python - _pickle.PicklingError : Can't pickle : it's not found as jinja2. utils.MissingType

Question

我正在尝试运行 PythonVirtualenvOperator在dag内在 gcloud composer环境，但在创建虚拟环境时收到错误。谷歌搜索让我找到了https://github.com/uqfoundation/dill/issues/93但我不确定我是否明白问题到底出在哪里。

环境详细信息

镜像版本:composer-1.4.1-airflow-1.9.0Python版本:3

代码:

from airflow.operators.python_operator import PythonOperator,PythonVirtualenvOperator
from airflow.operators.bash_operator import BashOperator
from airflow import models
import traceback
import datetime
from datetime import datetime as dt

yesterday = dt.combine(dt.today() - datetime.timedelta(1), dt.min.time())

def dummy_func(**kwargs):
    config=kwargs.get('config',False)
    print("1")

default_dag_args = {
    # Setting start date as yesterday starts the DAG immediately when it is
    # detected in the Cloud Storage bucket.
    'start_date':yesterday,
    # To email on failure or retry set 'email' arg to your email and enable
    # emailing here.
    'email_on_failure': False,
    'email_on_retry': False,
    # If a task fails, retry it once after waiting at least 5 minutes
    'retries': 0,
    #'retry_delay': datetime.timedelta(minutes=5),
    'project_id': 'abc'
    #models.Variable.get('gcp_project')
}

try:
  # [START composer_quickstart_schedule]
    DAG=models.DAG(
        'erised_map_dummy', #dag name
        # Continue to run DAG once per day
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args)
    # [END composer_quickstart_schedule]


    config={"a":1}

    # INITIALIZATION
    # Initializing Configurations variable

    # Read from GCS, Process and Write back to GCS
    Start = PythonVirtualenvOperator(
                    task_id="start_it",
                    python_callable=dummy_func,
                    requirements=["pandas==0.23.4","numpy","dill"],
                    use_dill=True,
                    system_site_packages=False,
                    #op_kwargs={'config':config},
                    provide_context=True,
                    dag=DAG
                    )

    op_end = BashOperator(task_id='Finished', bash_command='echo Success!', dag=DAG)

    Start >> op_end

except Exception as e:
    traceback.print_exc()
    raise(e)
# [END composer_quickstart]

错误: _pickle.PicklingError: Can't pickle <class 'jinja2.utils.MissingType'>: it's not found as jinja2.utils.MissingType

回溯堆栈:

Traceback (most recent call last):
  File "/usr/local/lib/python3.6/pickle.py", line 269, in _getattribute
    obj = getattr(obj, subpath)
AttributeError: module 'jinja2.utils' has no attribute 'MissingType'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.6/pickle.py", line 918, in save_global
    obj2, parent = _getattribute(module, name)
  File "/usr/local/lib/python3.6/pickle.py", line 272, in _getattribute
    .format(name, obj))
AttributeError: Can't get attribute 'MissingType' on <module 'jinja2.utils' from '/usr/local/lib/python3.6/site-packages/jinja2/utils.py'>

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/bin/airflow", line 6, in <module>
    exec(compile(open(__file__).read(), __file__, 'exec'))
  File "/usr/local/lib/airflow/airflow/bin/airflow", line 27, in <module>
    args.func(args)
  File "/usr/local/lib/airflow/airflow/bin/cli.py", line 528, in test
    ti.run(ignore_task_deps=True, ignore_ti_state=True, test_mode=True)
  File "/usr/local/lib/airflow/airflow/utils/db.py", line 50, in wrapper
    result = func(*args, **kwargs)
  File "/usr/local/lib/airflow/airflow/models.py", line 1583, in run
    session=session)
  File "/usr/local/lib/airflow/airflow/utils/db.py", line 50, in wrapper
    result = func(*args, **kwargs)
  File "/usr/local/lib/airflow/airflow/models.py", line 1492, in _run_raw_task
    result = task_copy.execute(context=context)
  File "/usr/local/lib/airflow/airflow/operators/python_operator.py", line 89, in execute
    return_value = self.execute_callable()
  File "/usr/local/lib/airflow/airflow/operators/python_operator.py", line 245, in execute_callable
    self._write_args(input_filename)
  File "/usr/local/lib/airflow/airflow/operators/python_operator.py", line 283, in _write_args
    dill.dump(arg_dict, f)
  File "/usr/local/lib/python3.6/site-packages/dill/_dill.py", line 286, in dump
    pik.dump(obj)
  File "/usr/local/lib/python3.6/pickle.py", line 409, in dump
    self.save(obj)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/site-packages/dill/_dill.py", line 893, in save_module_dict
    StockPickler.save_dict(pickler, obj)
  File "/usr/local/lib/python3.6/pickle.py", line 821, in save_dict
    self._batch_setitems(obj.items())
  File "/usr/local/lib/python3.6/pickle.py", line 847, in _batch_setitems
    save(v)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/site-packages/dill/_dill.py", line 893, in save_module_dict
    StockPickler.save_dict(pickler, obj)
  File "/usr/local/lib/python3.6/pickle.py", line 821, in save_dict
    self._batch_setitems(obj.items())
  File "/usr/local/lib/python3.6/pickle.py", line 847, in _batch_setitems
    save(v)
  File "/usr/local/lib/python3.6/pickle.py", line 521, in save
    self.save_reduce(obj=obj, *rv)
  File "/usr/local/lib/python3.6/pickle.py", line 634, in save_reduce
    save(state)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/site-packages/dill/_dill.py", line 893, in save_module_dict
    StockPickler.save_dict(pickler, obj)
  File "/usr/local/lib/python3.6/pickle.py", line 821, in save_dict
    self._batch_setitems(obj.items())
  File "/usr/local/lib/python3.6/pickle.py", line 847, in _batch_setitems
    save(v)
  File "/usr/local/lib/python3.6/pickle.py", line 521, in save
    self.save_reduce(obj=obj, *rv)
  File "/usr/local/lib/python3.6/pickle.py", line 634, in save_reduce
    save(state)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/site-packages/dill/_dill.py", line 893, in save_module_dict
    StockPickler.save_dict(pickler, obj)
  File "/usr/local/lib/python3.6/pickle.py", line 821, in save_dict
    self._batch_setitems(obj.items())
  File "/usr/local/lib/python3.6/pickle.py", line 847, in _batch_setitems
    save(v)
  File "/usr/local/lib/python3.6/pickle.py", line 521, in save
    self.save_reduce(obj=obj, *rv)
  File "/usr/local/lib/python3.6/pickle.py", line 634, in save_reduce
    save(state)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/site-packages/dill/_dill.py", line 893, in save_module_dict
    StockPickler.save_dict(pickler, obj)
  File "/usr/local/lib/python3.6/pickle.py", line 821, in save_dict
    self._batch_setitems(obj.items())
  File "/usr/local/lib/python3.6/pickle.py", line 847, in _batch_setitems
    save(v)
  File "/usr/local/lib/python3.6/pickle.py", line 521, in save
    self.save_reduce(obj=obj, *rv)
  File "/usr/local/lib/python3.6/pickle.py", line 634, in save_reduce
    save(state)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/site-packages/dill/_dill.py", line 893, in save_module_dict
    StockPickler.save_dict(pickler, obj)
  File "/usr/local/lib/python3.6/pickle.py", line 821, in save_dict
    self._batch_setitems(obj.items())
  File "/usr/local/lib/python3.6/pickle.py", line 847, in _batch_setitems
    save(v)
  File "/usr/local/lib/python3.6/pickle.py", line 521, in save
    self.save_reduce(obj=obj, *rv)
  File "/usr/local/lib/python3.6/pickle.py", line 634, in save_reduce
    save(state)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/site-packages/dill/_dill.py", line 893, in save_module_dict
    StockPickler.save_dict(pickler, obj)
  File "/usr/local/lib/python3.6/pickle.py", line 821, in save_dict
    self._batch_setitems(obj.items())
  File "/usr/local/lib/python3.6/pickle.py", line 847, in _batch_setitems
    save(v)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/site-packages/dill/_dill.py", line 893, in save_module_dict
    StockPickler.save_dict(pickler, obj)
  File "/usr/local/lib/python3.6/pickle.py", line 821, in save_dict
    self._batch_setitems(obj.items())
  File "/usr/local/lib/python3.6/pickle.py", line 847, in _batch_setitems
    save(v)
  File "/usr/local/lib/python3.6/pickle.py", line 521, in save
    self.save_reduce(obj=obj, *rv)
  File "/usr/local/lib/python3.6/pickle.py", line 634, in save_reduce
    save(state)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/site-packages/dill/_dill.py", line 893, in save_module_dict
    StockPickler.save_dict(pickler, obj)
  File "/usr/local/lib/python3.6/pickle.py", line 821, in save_dict
    self._batch_setitems(obj.items())
  File "/usr/local/lib/python3.6/pickle.py", line 847, in _batch_setitems
    save(v)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/pickle.py", line 781, in save_list
    self._batch_appends(obj)
  File "/usr/local/lib/python3.6/pickle.py", line 808, in _batch_appends
    save(tmp[0])
  File "/usr/local/lib/python3.6/pickle.py", line 521, in save
    self.save_reduce(obj=obj, *rv)
  File "/usr/local/lib/python3.6/pickle.py", line 634, in save_reduce
    save(state)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/site-packages/dill/_dill.py", line 893, in save_module_dict
    StockPickler.save_dict(pickler, obj)
  File "/usr/local/lib/python3.6/pickle.py", line 821, in save_dict
    self._batch_setitems(obj.items())
  File "/usr/local/lib/python3.6/pickle.py", line 847, in _batch_setitems
    save(v)
  File "/usr/local/lib/python3.6/pickle.py", line 521, in save
    self.save_reduce(obj=obj, *rv)
  File "/usr/local/lib/python3.6/pickle.py", line 634, in save_reduce
    save(state)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/site-packages/dill/_dill.py", line 893, in save_module_dict
    StockPickler.save_dict(pickler, obj)
  File "/usr/local/lib/python3.6/pickle.py", line 821, in save_dict
    self._batch_setitems(obj.items())
  File "/usr/local/lib/python3.6/pickle.py", line 847, in _batch_setitems
    save(v)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/site-packages/dill/_dill.py", line 1377, in save_function
    obj.__dict__), obj=obj)
  File "/usr/local/lib/python3.6/pickle.py", line 610, in save_reduce
    save(args)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/pickle.py", line 751, in save_tuple
    save(element)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/site-packages/dill/_dill.py", line 893, in save_module_dict
    StockPickler.save_dict(pickler, obj)
  File "/usr/local/lib/python3.6/pickle.py", line 821, in save_dict
    self._batch_setitems(obj.items())
  File "/usr/local/lib/python3.6/pickle.py", line 847, in _batch_setitems
    save(v)
  File "/usr/local/lib/python3.6/pickle.py", line 521, in save
    self.save_reduce(obj=obj, *rv)
  File "/usr/local/lib/python3.6/pickle.py", line 605, in save_reduce
    save(cls)
  File "/usr/local/lib/python3.6/pickle.py", line 476, in save
    f(self, obj) # Call unbound method with explicit self
  File "/usr/local/lib/python3.6/site-packages/dill/_dill.py", line 1321, in save_type
    StockPickler.save_global(pickler, obj)
  File "/usr/local/lib/python3.6/pickle.py", line 922, in save_global
    (obj, module_name, name))
_pickle.PicklingError: Can't pickle <class 'jinja2.utils.MissingType'>: it's not found as jinja2.utils.MissingType

Answer 1

我将 provide_context 设置为 False 后解决了。根据文档 :param Provide_context: 如果设置为 true，Airflow 将传递一组可在您的函数中使用的关键字参数。这组 kwargs 与您可以在 jinja 模板中使用的内容完全对应。为此，您需要在函数头中定义 **kwargs。这会导致错误。