首先,你不能在通用的 Python 3 Airflow 集群上运行 Python 2 Airflow 工作程序:
Airflow 使用 SQLAlchemy(我认为是为了读写有关 DAG 的元数据到数据库中)。当你在工作程序上运行 DAG 时,它将从数据库中读取有关该 DAG 的 pickled 信息。如果你的其他非工作程序组件使用的是 Python 3,则它们将以 pickle 4 的形式写入数据库,而工作程序将尝试以 Python 2 的形式从数据库中读取。
特别地,在 SQLAlchemy 中查看 sqltypes.py:
class PickleType(TypeDecorator):
"""Holds Python objects, which are serialized using pickle.
PickleType builds upon the Binary type to apply Python's
``pickle.dumps()`` to incoming objects, and ``pickle.loads()`` on
the way out, allowing any pickleable Python object to be stored as
a serialized binary field.
To allow ORM change events to propagate for elements associated
with :class:`.PickleType`, see :ref:`mutable_toplevel`.
"""
impl = LargeBinary
def __init__(self, protocol=pickle.HIGHEST_PROTOCOL,
pickler=None, comparator=None):
然后在compat.py中进行最终的序列化,这个过程最终会在sqltypes.py中完成。
py36 = sys.version_info >= (3, 6)
py33 = sys.version_info >= (3, 3)
py35 = sys.version_info >= (3, 5)
py32 = sys.version_info >= (3, 2)
py3k = sys.version_info >= (3, 0)
py2k = sys.version_info < (3, 0)
py265 = sys.version_info >= (2, 6, 5)
jython = sys.platform.startswith('java')
pypy = hasattr(sys, 'pypy_version_info')
win32 = sys.platform.startswith('win')
cpython = not pypy and not jython # TODO: something better for this ?
import collections
next = next
if py3k:
import pickle
else:
try:
import cPickle as pickle
except ImportError:
import pickle
此外,airflow 中的 donot_pickle = True 似乎对此没有影响???也许是因为根据 这里 的说明,它只在回填中相关?
PythonVirtualenvOperator
(https://github.com/apache/incubator-airflow/blob/e9babff4eb3334b0d71cda31c2f6cbfe7b741389/airflow/operators/python_operator.py#L200) 或设置您的工作程序安装Python 2但不在您的路径中,然后使用BashOperator
指定python 2的绝对路径(即/path/to/python2 /path/to/your_code_that_requires_py2
)以在Py2中执行脚本。 - cwurtz