Airflow (Google Composer) TypeError: can't pickle _thread.RLock objects

user_6408784 0
Viewed 1813

I'm using airflow(Google composer), but experienced some exceptions below

TypeError: can't pickle _thread.RLock objects

Ooops.

                          ____/ (  (    )   )  \___
                         /( (  (  )   _    ))  )   )\
                       ((     (   )(    )  )   (   )  )
                     ((/  ( _(   )   (   _) ) (  () )  )
                    ( (  ( (_)   ((    (   )  .((_ ) .  )_
                   ( (  )    (      (  )    )   ) . ) (   )
                  (  (   (  (   ) (  _  ( _) ).  ) . ) ) ( )
                  ( (  (   ) (  )   (  ))     ) _)(   )  )  )
                 ( (  ( \ ) (    (_  ( ) ( )  )   ) )  )) ( )
                  (  (   (  (   (_ ( ) ( _    )  ) (  )  )   )
                 ( (  ( (  (  )     (_  )  ) )  _)   ) _( ( )
                  ((  (   )(    (     _    )   _) _(_ (  (_ )
                   (_((__(_(__(( ( ( |  ) ) ) )_))__))_)___)
                   ((__)        \\||lll|l||///          \_))
                            (   /(/ (  )  ) )\   )
                          (    ( ( ( | | ) ) )\   )
                           (   /(| / ( )) ) ) )) )
                         (     ( ((((_(|)_)))))     )
                          (      ||\(|(|)|/||     )
                        (        |(||(||)||||        )
                          (     //|/l|||)|\\ \     )
                        (/ / //  /|//||||\\  \ \  \ _)
-------------------------------------------------------------------------------
Node: d93e048dc08a
-------------------------------------------------------------------------------
Traceback (most recent call last):
  File "/opt/python3.6/lib/python3.6/site-packages/flask/app.py", line 2447, in wsgi_app
    response = self.full_dispatch_request()
  File "/opt/python3.6/lib/python3.6/site-packages/flask/app.py", line 1952, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/opt/python3.6/lib/python3.6/site-packages/flask/app.py", line 1821, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/opt/python3.6/lib/python3.6/site-packages/flask/_compat.py", line 39, in reraise
    raise value
  File "/opt/python3.6/lib/python3.6/site-packages/flask/app.py", line 1950, in full_dispatch_request
    rv = self.dispatch_request()
  File "/opt/python3.6/lib/python3.6/site-packages/flask/app.py", line 1936, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File "/opt/python3.6/lib/python3.6/site-packages/flask_admin/base.py", line 69, in inner
    return self._run_view(f, *args, **kwargs)
  File "/opt/python3.6/lib/python3.6/site-packages/flask_admin/base.py", line 368, in _run_view
    return fn(self, *args, **kwargs)
  File "/opt/python3.6/lib/python3.6/site-packages/flask_login/utils.py", line 258, in decorated_view
    return func(*args, **kwargs)
  File "/usr/local/lib/airflow/airflow/www/utils.py", line 290, in wrapper
    return f(*args, **kwargs)
  File "/usr/local/lib/airflow/airflow/www/utils.py", line 337, in wrapper
    return f(*args, **kwargs)
  File "/usr/local/lib/airflow/airflow/www/views.py", line 1335, in clear
    include_upstream=upstream)
  File "/usr/local/lib/airflow/airflow/models/dag.py", line 1243, in sub_dag
    for t in regex_match + also_include}
  File "/usr/local/lib/airflow/airflow/models/dag.py", line 1243, in <dictcomp>
    for t in regex_match + also_include}
  File "/opt/python3.6/lib/python3.6/copy.py", line 161, in deepcopy
    y = copier(memo)
  File "/usr/local/lib/airflow/airflow/models/baseoperator.py", line 678, in __deepcopy__
    setattr(result, k, copy.deepcopy(v, memo))
  File "/opt/python3.6/lib/python3.6/copy.py", line 180, in deepcopy
    y = _reconstruct(x, memo, *rv)
  File "/opt/python3.6/lib/python3.6/copy.py", line 280, in _reconstruct
    state = deepcopy(state, memo)
  File "/opt/python3.6/lib/python3.6/copy.py", line 150, in deepcopy
    y = copier(x, memo)
  File "/opt/python3.6/lib/python3.6/copy.py", line 240, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/opt/python3.6/lib/python3.6/copy.py", line 180, in deepcopy
    y = _reconstruct(x, memo, *rv)
  File "/opt/python3.6/lib/python3.6/copy.py", line 280, in _reconstruct
    state = deepcopy(state, memo)
  File "/opt/python3.6/lib/python3.6/copy.py", line 150, in deepcopy
    y = copier(x, memo)
  File "/opt/python3.6/lib/python3.6/copy.py", line 240, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/opt/python3.6/lib/python3.6/copy.py", line 180, in deepcopy
    y = _reconstruct(x, memo, *rv)
  File "/opt/python3.6/lib/python3.6/copy.py", line 280, in _reconstruct
    state = deepcopy(state, memo)
  File "/opt/python3.6/lib/python3.6/copy.py", line 150, in deepcopy
    y = copier(x, memo)
  File "/opt/python3.6/lib/python3.6/copy.py", line 240, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/opt/python3.6/lib/python3.6/copy.py", line 150, in deepcopy
    y = copier(x, memo)
  File "/opt/python3.6/lib/python3.6/copy.py", line 215, in _deepcopy_list
    append(deepcopy(a, memo))
  File "/opt/python3.6/lib/python3.6/copy.py", line 180, in deepcopy
    y = _reconstruct(x, memo, *rv)
  File "/opt/python3.6/lib/python3.6/copy.py", line 280, in _reconstruct
    state = deepcopy(state, memo)
  File "/opt/python3.6/lib/python3.6/copy.py", line 150, in deepcopy
    y = copier(x, memo)
  File "/opt/python3.6/lib/python3.6/copy.py", line 240, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/opt/python3.6/lib/python3.6/copy.py", line 169, in deepcopy
    rv = reductor(4)
TypeError: can't pickle _thread.RLock objects

What I have tried?

  1. From airflow UI interface to clear task, the task, not work
  2. From command like use command backfill, not work
  3. Restart Airflow webservice, not work
  4. Changed DAG retry_delay=timedelta(seconds=5)

Can anyone help with the above? Thanks a lot

I was noticed there have some similar questions on StackOverflow, but those questions not really get the fix.

Airflow can't pickle _thread._local objects

Airflow 1.9.0 ExternalTaskSensor retry_delay=30 yields TypeError: can't pickle _thread.RLock objects

google-cloud-composer airflow-scheduler airflow
1 Answers

I guess that the similar problem was reported on Apache Jira tracker, looking into the discussion threads there, I can point out some concerns that might help to overcome this issue:

  • I would advise to look through the particular DAG, inspecting the correct type of the default arguments for dedicated DAG operator, despite the fact that retry_delay has been already checked, it would worth to review the rest of the parameters, the link was already mentioned in the question;

  • To further debug, verify whether you DAG Operator consumes only pickable(serializable) object, as per comment being posted here.

  • I assume that we are still receiving some issues from the users with clearing Airflow DAG tasks via Airflow WEB UI, just check out this thread. In order to mitigate this problem you can either delete the failing task within Airflow command-line tool (example here) or as a last resort remove the task_id record from Airflow metadata database.

    Connecting to the one of Composer's workers:

    kubectl -it exec $(kubectl get po -l run=airflow-worker -o jsonpath='{.items[0].metadata.name}' \
    -n $(kubectl get ns| grep composer*| awk '{print $1}')) -n $(kubectl get ns| grep composer*| awk '{print $1}') \
    -c airflow-worker -- mysql -u root -h airflow-sqlproxy-service.default

    Using mysql client:

    mysql> show databases;
+-----------------------------------------+
| Database                                |
+-----------------------------------------+
| information_schema                      |
| composer-1-11-3-airflow-1-10-6-* |
| mysql                                   |
| performance_schema                      |
| sys                                     |
+-----------------------------------------+
5 rows in set (0.01 sec)

    Launch connection to composer-1-11-3-airflow-1-10-6-* schema:

    mysql> use composer-1-11-3-airflow-1-10-6-*;

    Delete the failing task_id:

    delete from task_instance where task_id='<task_id>' AND execution_date='<execution_date>'
user_99288090
Related