You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@airflow.apache.org by Maxime Beauchemin <ma...@gmail.com> on 2016/10/04 04:11:13 UTC
Re: [jira] [Commented] (AIRFLOW-351) Failed to clear downstream tasks
I've noted before that it'd be nice to have a `DAG.validate` method along
with a `airflow validate` CLI command that would make sure that the DAG is
deepcopyable, has a normal memory footprint, that templates render without
issues, and that sort of things. Perhaps it could print related stats
around the DAG as well (parsing time, number of tasks by operator type,
template render time, ...).
This is a nice, easy-ish, contained yet useful task if someone wants to
give it a shot.
Max
On Mon, Oct 3, 2016 at 12:19 AM, Jim Zheng (JIRA) <ji...@apache.org> wrote:
>
> [ https://issues.apache.org/jira/browse/AIRFLOW-351?page=
> com.atlassian.jira.plugin.system.issuetabpanels:comment-
> tabpanel&focusedCommentId=15541745#comment-15541745 ]
>
> Jim Zheng commented on AIRFLOW-351:
> -----------------------------------
>
> [~maxime.beauchemin@apache.org] is right - thanks for the help! For us,
> we were attaching a boto3 client to a base operator. We didn't notice the
> regression until much later.
>
> This might be good to document as a gotcha, since it's really hard to tell
> from the backtrace what failed.
>
> > Failed to clear downstream tasks
> > --------------------------------
> >
> > Key: AIRFLOW-351
> > URL: https://issues.apache.org/jira/browse/AIRFLOW-351
> > Project: Apache Airflow
> > Issue Type: Bug
> > Components: models, subdag, webserver
> > Affects Versions: Airflow 1.7.1.3
> > Reporter: Adinata
> > Attachments: dag_error.py, error_on_clear_dag.txt
> >
> >
> > {code}
> > ____/ ( ( ) ) \___
> > /( ( ( ) _ )) ) )\
> > (( ( )( ) ) ( ) )
> > ((/ ( _( ) ( _) ) ( () ) )
> > ( ( ( (_) (( ( ) .((_ ) . )_
> > ( ( ) ( ( ) ) ) . ) ( )
> > ( ( ( ( ) ( _ ( _) ). ) . ) ) ( )
> > ( ( ( ) ( ) ( )) ) _)( ) ) )
> > ( ( ( \ ) ( (_ ( ) ( ) ) ) ) )) ( )
> > ( ( ( ( (_ ( ) ( _ ) ) ( ) ) )
> > ( ( ( ( ( ) (_ ) ) ) _) ) _( ( )
> > (( ( )( ( _ ) _) _(_ ( (_ )
> > (_((__(_(__(( ( ( | ) ) ) )_))__))_)___)
> > ((__) \\||lll|l||/// \_))
> > ( /(/ ( ) ) )\ )
> > ( ( ( ( | | ) ) )\ )
> > ( /(| / ( )) ) ) )) )
> > ( ( ((((_(|)_))))) )
> > ( ||\(|(|)|/|| )
> > ( |(||(||)|||| )
> > ( //|/l|||)|\\ \ )
> > (/ / // /|//||||\\ \ \ \ _)
> > ------------------------------------------------------------
> -------------------
> > Node: 9889a7c79e9b
> > ------------------------------------------------------------
> -------------------
> > Traceback (most recent call last):
> > File "/usr/local/lib/python2.7/dist-packages/flask/app.py", line
> 1817, in wsgi_app
> > response = self.full_dispatch_request()
> > File "/usr/local/lib/python2.7/dist-packages/flask/app.py", line
> 1477, in full_dispatch_request
> > rv = self.handle_user_exception(e)
> > File "/usr/local/lib/python2.7/dist-packages/flask/app.py", line
> 1381, in handle_user_exception
> > reraise(exc_type, exc_value, tb)
> > File "/usr/local/lib/python2.7/dist-packages/flask/app.py", line
> 1475, in full_dispatch_request
> > rv = self.dispatch_request()
> > File "/usr/local/lib/python2.7/dist-packages/flask/app.py", line
> 1461, in dispatch_request
> > return self.view_functions[rule.endpoint](**req.view_args)
> > File "/usr/local/lib/python2.7/dist-packages/flask_admin/base.py",
> line 68, in inner
> > return self._run_view(f, *args, **kwargs)
> > File "/usr/local/lib/python2.7/dist-packages/flask_admin/base.py",
> line 367, in _run_view
> > return fn(self, *args, **kwargs)
> > File "/usr/local/lib/python2.7/dist-packages/flask_login.py", line
> 755, in decorated_view
> > return func(*args, **kwargs)
> > File "/usr/local/lib/python2.7/dist-packages/airflow/www/utils.py",
> line 118, in wrapper
> > return f(*args, **kwargs)
> > File "/usr/local/lib/python2.7/dist-packages/airflow/www/utils.py",
> line 167, in wrapper
> > return f(*args, **kwargs)
> > File "/usr/local/lib/python2.7/dist-packages/airflow/www/views.py",
> line 1017, in clear
> > include_upstream=upstream)
> > File "/usr/local/lib/python2.7/dist-packages/airflow/models.py", line
> 2870, in sub_dag
> > dag = copy.deepcopy(self)
> > File "/usr/lib/python2.7/copy.py", line 174, in deepcopy
> > y = copier(memo)
> > File "/usr/local/lib/python2.7/dist-packages/airflow/models.py", line
> 2856, in __deepcopy__
> > setattr(result, k, copy.deepcopy(v, memo))
> > File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
> > y = copier(x, memo)
> > File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
> > y[deepcopy(key, memo)] = deepcopy(value, memo)
> > File "/usr/lib/python2.7/copy.py", line 174, in deepcopy
> > y = copier(memo)
> > File "/usr/local/lib/python2.7/dist-packages/airflow/models.py", line
> 1974, in __deepcopy__
> > setattr(result, k, copy.deepcopy(v, memo))
> > File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
> > y = _reconstruct(x, rv, 1, memo)
> > File "/usr/lib/python2.7/copy.py", line 334, in _reconstruct
> > state = deepcopy(state, memo)
> > File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
> > y = copier(x, memo)
> > File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
> > y[deepcopy(key, memo)] = deepcopy(value, memo)
> > File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
> > y = _reconstruct(x, rv, 1, memo)
> > File "/usr/lib/python2.7/copy.py", line 334, in _reconstruct
> > state = deepcopy(state, memo)
> > File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
> > y = copier(x, memo)
> > File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
> > y[deepcopy(key, memo)] = deepcopy(value, memo)
> > File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
> > y = _reconstruct(x, rv, 1, memo)
> > File "/usr/lib/python2.7/copy.py", line 334, in _reconstruct
> > state = deepcopy(state, memo)
> > File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
> > y = copier(x, memo)
> > File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
> > y[deepcopy(key, memo)] = deepcopy(value, memo)
> > File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
> > y = copier(x, memo)
> > File "/usr/lib/python2.7/copy.py", line 230, in _deepcopy_list
> > y.append(deepcopy(a, memo))
> > File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
> > y = _reconstruct(x, rv, 1, memo)
> > File "/usr/lib/python2.7/copy.py", line 334, in _reconstruct
> > state = deepcopy(state, memo)
> > File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
> > y = copier(x, memo)
> > File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
> > y[deepcopy(key, memo)] = deepcopy(value, memo)
> > File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
> > y = _reconstruct(x, rv, 1, memo)
> > File "/usr/lib/python2.7/copy.py", line 334, in _reconstruct
> > state = deepcopy(state, memo)
> > File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
> > y = copier(x, memo)
> > File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
> > y[deepcopy(key, memo)] = deepcopy(value, memo)
> > File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
> > y = _reconstruct(x, rv, 1, memo)
> > File "/usr/lib/python2.7/copy.py", line 329, in _reconstruct
> > y = callable(*args)
> > File "/usr/lib/python2.7/copy_reg.py", line 93, in __newobj__
> > return cls.__new__(cls, *args)
> > TypeError: object.__new__(thread.lock) is not safe, use
> thread.lock.__new__()
> > {code}
>
>
>
> --
> This message was sent by Atlassian JIRA
> (v6.3.4#6332)
>