如何在气流中使用PythonVirtualenvOperator？

Question

基本上，我在研究气流，并开发了一个任务，从外部源下载文件。在

t1 = PythonOperator(
        task_id='download',
        python_callable=download,
        provide_context=True,
        dag=dag)

这种气流是在虚拟环境（pipenv）中运行的。在

下载功能是：

^{pr2}$

所以基本上我用xcon把数据从一个任务传递到另一个任务…使用这种配置不可能管理每个DAG的所有依赖关系。。。在

在文档中，我发现了一个名为“pythonvirtualenvooperator”的类，为了实现我写的：

t1 = PythonVirtualenvOperator(
        task_id='download',
        python_callable=download,
        requirements=['requests'],
        python_version='3.8',
        provide_context=True,
        dag=dag
    )

它给了我以下错误：

TypeError: can't pickle module objects

download_file函数是另一个文件中的API连接。在

有什么建议可以帮助我管理环境并在任务之间建立联系？在

Answer 1

根据PythonVirtualenvOperator的定义：

The function must be defined using def, and not be
part of a class. All imports must happen inside the function
and no variables outside of the scope may be referenced.

我猜在你的download函数中调用的代码链的某个地方，有一个方法是使用顶级导入从另一个文件导入的。也许把这个导入移到你的download函数中就足够了？在