Airflow 1.9.0 ExternalTaskSensor retry_delay=30 产生 TypeError: can't pickle _thread.RLock objects答案

【问题标题】：Airflow 1.9.0 ExternalTaskSensor retry_delay=30 yields TypeError: can't pickle _thread.RLock objectsAirflow 1.9.0 ExternalTaskSensor retry_delay=30 产生 TypeError: can't pickle _thread.RLock objects
【发布时间】：2018-02-22 17:09:59
【问题描述】：

正如标题所说；在 Airflow 1.9.0 中，如果您将 retry_delay=30（或任何其他数字）参数与 ExternalTaskSensor 一起使用，则 DAG 将运行得很好，直到您想要清除气流 GUI 中的任务实例 -> 它将返回以下错误: "TypeError: can't pickle _thread.RLock objects" （和一个不错的糟糕消息）但是，如果您使用 retry_delay=timedelta(seconds=30) 清除任务实例工作正常。

如果我查看 models.py 方法，deepcopy 应该没问题，所以对我来说这似乎是一种奇怪的行为。我错过了什么，还是这是一个错误？

您可以在下面找到一个最小的 DAG 示例。

from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.sensors import ExternalTaskSensor
from datetime import datetime, timedelta


dag_name = 'soft_fail_example'
schedule_interval = "0 * * * *"
default_args = {
            'owner': 'airflow',
            'depends_on_past': False,
            'start_date': datetime(2018, 1, 1),
            'email': [],
            'email_on_failure': False,
            'email_on_retry': False
        }

test_dag = DAG(dag_name, default_args=default_args, schedule_interval=schedule_interval, 
catchup=False, max_active_runs=1)


ets = ExternalTaskSensor(task_id="test_external_task_sensor", dag=test_dag, soft_fail=False, 
timeout=10, retries=0, poke_interval=1, retry_delay=30, external_dag_id="dependent_dag_id",
                         external_task_id="dependent_task_id")

dummy_task = DummyOperator(task_id="collection_task", dag=test_dag)

dummy_task << ets

编辑：并根据要求堆栈跟踪：

    Ooops.

                              ____/ (  (    )   )  \___
                             /( (  (  )   _    ))  )   )\
                           ((     (   )(    )  )   (   )  )
                         ((/  ( _(   )   (   _) ) (  () )  )
                        ( (  ( (_)   ((    (   )  .((_ ) .  )_
                       ( (  )    (      (  )    )   ) . ) (   )
                      (  (   (  (   ) (  _  ( _) ).  ) . ) ) ( )
                      ( (  (   ) (  )   (  ))     ) _)(   )  )  )
                     ( (  ( \ ) (    (_  ( ) ( )  )   ) )  )) ( )
                      (  (   (  (   (_ ( ) ( _    )  ) (  )  )   )
                     ( (  ( (  (  )     (_  )  ) )  _)   ) _( ( )
                      ((  (   )(    (     _    )   _) _(_ (  (_ )
                       (_((__(_(__(( ( ( |  ) ) ) )_))__))_)___)
                       ((__)        \\||lll|l||///          \_))
                                (   /(/ (  )  ) )\   )
                              (    ( ( ( | | ) ) )\   )
                               (   /(| / ( )) ) ) )) )
                             (     ( ((((_(|)_)))))     )
                              (      ||\(|(|)|/||     )
                            (        |(||(||)||||        )
                              (     //|/l|||)|\\ \     )
                            (/ / //  /|//||||\\  \ \  \ _)
    -------------------------------------------------------------------------------
    Node: jb-VirtualBox
    -------------------------------------------------------------------------------
    Traceback (most recent call last):
      File "/home/jb/Documents/p3_cdc_data_flow/lib/python3.6/site-packages/flask/app.py", line 1988, in wsgi_app
        response = self.full_dispatch_request()
      File "/home/jb/Documents/p3_cdc_data_flow/lib/python3.6/site-packages/flask/app.py", line 1641, in full_dispatch_request
        rv = self.handle_user_exception(e)
      File "/home/jb/Documents/p3_cdc_data_flow/lib/python3.6/site-packages/flask/app.py", line 1544, in handle_user_exception
        reraise(exc_type, exc_value, tb)
      File "/home/jb/Documents/p3_cdc_data_flow/lib/python3.6/site-packages/flask/_compat.py", line 33, in reraise
        raise value
      File "/home/jb/Documents/p3_cdc_data_flow/lib/python3.6/site-packages/flask/app.py", line 1639, in full_dispatch_request
        rv = self.dispatch_request()
      File "/home/jb/Documents/p3_cdc_data_flow/lib/python3.6/site-packages/flask/app.py", line 1625, in dispatch_request
        return self.view_functions[rule.endpoint](**req.view_args)
      File "/home/jb/Documents/p3_cdc_data_flow/lib/python3.6/site-packages/flask_admin/base.py", line 69, in inner
        return self._run_view(f, *args, **kwargs)
      File "/home/jb/Documents/p3_cdc_data_flow/lib/python3.6/site-packages/flask_admin/base.py", line 368, in _run_view
        return fn(self, *args, **kwargs)
      File "/home/jb/Documents/p3_cdc_data_flow/lib/python3.6/site-packages/flask_login.py", line 755, in decorated_view
        return func(*args, **kwargs)
      File "/home/jb/Documents/p3_cdc_data_flow/lib/python3.6/site-packages/airflow/www/utils.py", line 262, in wrapper
        return f(*args, **kwargs)
      File "/home/jb/Documents/p3_cdc_data_flow/lib/python3.6/site-packages/airflow/www/utils.py", line 309, in wrapper
        return f(*args, **kwargs)
      File "/home/jb/Documents/p3_cdc_data_flow/lib/python3.6/site-packages/airflow/www/views.py", line 989, in clear
        include_upstream=upstream)
      File "/home/jb/Documents/p3_cdc_data_flow/lib/python3.6/site-packages/airflow/models.py", line 3527, in sub_dag
        dag = copy.deepcopy(self)
      File "/usr/lib/python3.6/copy.py", line 161, in deepcopy
        y = copier(memo)
      File "/home/jb/Documents/p3_cdc_data_flow/lib/python3.6/site-packages/airflow/models.py", line 3512, in __deepcopy__
        setattr(result, k, copy.deepcopy(v, memo))
      File "/usr/lib/python3.6/copy.py", line 150, in deepcopy
        y = copier(x, memo)
      File "/usr/lib/python3.6/copy.py", line 240, in _deepcopy_dict
        y[deepcopy(key, memo)] = deepcopy(value, memo)
      File "/usr/lib/python3.6/copy.py", line 161, in deepcopy
        y = copier(memo)
      File "/home/jb/Documents/p3_cdc_data_flow/lib/python3.6/site-packages/airflow/models.py", line 2437, in __deepcopy__
        setattr(result, k, copy.deepcopy(v, memo))
      File "/usr/lib/python3.6/copy.py", line 180, in deepcopy
        y = _reconstruct(x, memo, *rv)
      File "/usr/lib/python3.6/copy.py", line 280, in _reconstruct
        state = deepcopy(state, memo)
      File "/usr/lib/python3.6/copy.py", line 150, in deepcopy
        y = copier(x, memo)
      File "/usr/lib/python3.6/copy.py", line 240, in _deepcopy_dict
        y[deepcopy(key, memo)] = deepcopy(value, memo)
      File "/usr/lib/python3.6/copy.py", line 180, in deepcopy
        y = _reconstruct(x, memo, *rv)
      File "/usr/lib/python3.6/copy.py", line 280, in _reconstruct
        state = deepcopy(state, memo)
      File "/usr/lib/python3.6/copy.py", line 150, in deepcopy
        y = copier(x, memo)
      File "/usr/lib/python3.6/copy.py", line 240, in _deepcopy_dict
        y[deepcopy(key, memo)] = deepcopy(value, memo)
      File "/usr/lib/python3.6/copy.py", line 180, in deepcopy
        y = _reconstruct(x, memo, *rv)
      File "/usr/lib/python3.6/copy.py", line 280, in _reconstruct
        state = deepcopy(state, memo)
      File "/usr/lib/python3.6/copy.py", line 150, in deepcopy
        y = copier(x, memo)
      File "/usr/lib/python3.6/copy.py", line 240, in _deepcopy_dict
        y[deepcopy(key, memo)] = deepcopy(value, memo)
      File "/usr/lib/python3.6/copy.py", line 150, in deepcopy
        y = copier(x, memo)
      File "/usr/lib/python3.6/copy.py", line 215, in _deepcopy_list
        append(deepcopy(a, memo))
      File "/usr/lib/python3.6/copy.py", line 180, in deepcopy
        y = _reconstruct(x, memo, *rv)
      File "/usr/lib/python3.6/copy.py", line 280, in _reconstruct
        state = deepcopy(state, memo)
      File "/usr/lib/python3.6/copy.py", line 150, in deepcopy
        y = copier(x, memo)
      File "/usr/lib/python3.6/copy.py", line 240, in _deepcopy_dict
        y[deepcopy(key, memo)] = deepcopy(value, memo)
      File "/usr/lib/python3.6/copy.py", line 169, in deepcopy
        rv = reductor(4)
    TypeError: can't pickle _thread.RLock objects

【问题讨论】：

请发布您的错误的整个堆栈跟踪。
@joeb 我添加了堆栈跟踪。
好的。我以前见过这个（我有一些 Airflow 的自定义扩展），你会想在 Airflow 的代码中添加一行来帮助调试。在airflow/models.py 中的第3512 行之前添加“print(k)”——再次导致错误并共享在跟踪之前打印的行。
@joeb 我得到以下几行：default_args, _dag_id, _full_filepath, _concurrency,_pickle_id, _description, fileloc,task_dict 在最后一行之后我得到这个错误：ERROR in app: Exception on /admin/airflow/clear [GET]
@joeb，再次浏览代码后，我找到了答案，基本上我错过了一些东西。谢谢你的调查。

标签： python airflow airflow-scheduler

【解决方案1】：

再次查看这个问题后，文档明确指出retry_delay应该是一个timedelta。因此，如果您为 retry_delay 输入整数而不是 timedelta，那么 DAG 可以正常工作。

在models.py中，BaseOperator：

   :param retry_delay: delay between retries
   :type retry_delay: timedelta

【讨论】：

我看到了这个错误，我的重试延迟都在使用 timedelta retry_delay=timedelta(minutes=5)。对我来说，只有当我在运行过程中停止 DAG（通过关闭 DAG）然后清除所有 Past, Future, Downstream, Upstream, Recursive 任务时才会发生这种情况。这可能是因为计划重试任务并且我正在尝试清除它们或其他原因吗？