google-cloud-platform - PythonVirtualenvOperator 失败并出现 TypeError : cannot pickle 'module' object

Question

以下发生在 Cloud Composer 2.1.xx 上

我正在尝试使用带有模板化参数的 PythonVirtualenvOperator。不幸的是，运算符(operator)因以下错误而失败:

TypeError: cannot pickle 'module' object

这是我的狗的代码:

with models.DAG(
    'name',
    schedule_interval=None,
    start_date=datetime.datetime(2018, 1, 1),
    catchup=False) as dag:

t1 = PythonVirtualenvOperator(
    task_id='download',
    python_callable=update_files,
    requirements=['google-cloud-firestore==2.2.0'],
    python_version='3.8',
    op_kwargs={
        "inputPaths": '{{ dag_run.conf["inputPaths"] }}'
    }
)

Python 函数的代码如下所示:

def update_files(**kwargs):
    from google.cloud import firestore
    import datetime
    paths = kwargs['inputPaths']
    .....

我尝试使用参数 use_dill = True 和 How to use PythonVirtualenvOperator in airflow? 上的建议答案但它并没有使任何事情变得更好。

Answer 1

从 airflow 2.x 版本开始，我们将无法在 virtualEnv python_callable 方法中获取 op_kwargs 字典。相反，我们将只能在可调用对象中获得值(value)在你的情况下，你只改变了你的可调用

def update_files(inputPaths): // def callable(<DICT_KEY>):
from google.cloud import firestore
import datetime
print(inputPaths)