PythonVirtualenvOperator 因 TypeError 失败：无法 pickle 'module' 对象

Question

Cloud Composer 2.1.xx 上会发生以下情况

我正在尝试将 PythonVirtualenvOperator 与模板化参数一起使用。不幸的是，操作员失败并出现以下错误：

TypeError: cannot pickle 'module' object

这是我的 dag 的代码：

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和建议的答案如何在气流中使用PythonVirtualenvOperator？但这并没有让任何事情变得更好。

Answer 1

从airflow 2.x版本开始，我们将无法在virtualEnv python_callable方法中获取op_kwargs dict。相反，我们将只能获取可调用对象中的值在您的情况下，您只能将可调用对象更改为

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