从谷歌云作曲家运行数据流时导入依赖项的问题

Question

我正在从谷歌云作曲家运行数据流，数据流脚本包含一些非标准依赖项，如zeep，googleads。这些都需要安装在数据流工作节点上，所以我用setup.py打包它们。当我尝试在dag中运行它时，作曲家正在验证数据流文件并抱怨No module names Zeep , googleads。所以我创建了pythonvirtualenvoperator并安装了所需的所有非标准依赖项，并尝试运行数据流作业，它仍然抱怨引入zeep和googleads。

这是我的代码库：

PULL_DATA = PythonVirtualenvOperator(
    task_id=PROCESS_TASK_ID,
    python_callable=execute_dataflow,
    op_kwargs={
        'main': 'main.py',
        'project': PROJECT,
        'temp_location': 'gs://bucket/temp',
        'setup_file': 'setup.py',
        'max_num_workers': 2,
        'output': 'gs://bucket/output',
        'project_id': PROJECT_ID},
    requirements=['google-cloud-storage==1.10.0', 'zeep==3.2.0',
                  'argparse==1.4.0', 'google-cloud-kms==0.2.1',
                  'googleads==15.0.2', 'dill'],
    python_version='2.7',
    use_dill=True,
    system_site_packages=True,
    on_failure_callback=on_failure_handler,
    on_success_callback=on_success_handler,
    dag='my-dag')

和我的python可调用代码：

def execute_dataflow(**kwargs):
        import subprocess
        TEMPLATED_COMMAND = """
                          python main.py \
                                 --runner DataflowRunner \
                                 --project {project} \
                                 --region us-central1 \
                                 --temp_location {temp_location} \
                                 --setup_file {setup_file} \
                                 --output {output} \
                                 --project_id {project_id} 
                          """.format(**kwargs)
        process = subprocess.Popen(['/bin/bash', '-c', TEMPLATED_COMMAND])
        process.wait()
        return process.returncode

我的main.py文件

import zeep
import googleads

{Apache-beam-code to construct dataflow pipeline}

有什么建议？

Answer 1

我的工作有requirements.txt。它不是像你的那样使用--setup_file选项，而是指定以下内容：

--requirements_file prod_requirements.txt

这告诉DataFlow在运行作业之前在requirements.txt中安装库。

参考：https://beam.apache.org/documentation/sdks/python-pipeline-dependencies/

Answer 2

使用带有import googleads, zeep的示例Dataflow管道脚本，我设置了一个测试Composer环境。 DAG就像你的一样，我也得到同样的错误。然后我做了一些更改，以确保可以在工作计算机上找到依赖项。

在DAG中，我使用普通的PythonOperator，而不是PythonVirtualenvOperator。我有我的数据流管道和设置文件（main.py和setup.py）in a Google Cloud Storage bucket，因此Composer可以找到它们。设置文件包含我需要的要求列表，例如zeep和googleads。我改编了here的样本设置文件，改变了这个：

REQUIRED_PACKAGES = [
    'google-cloud-storage==1.10.0', 'zeep==3.2.0',
'argparse==1.4.0', 'google-cloud-kms==0.2.1',
'googleads==15.0.2', 'dill'
    ]

setuptools.setup(
    name='Imports test',
    version='1',
    description='Imports test workflow package.',
    install_requires=REQUIRED_PACKAGES,
    packages=setuptools.find_packages(),
    cmdclass={
        # Command class instantiated and run during pip install scenarios.
        'build': build,
        'CustomCommands': CustomCommands,
        }
    )

我的日子

with models.DAG(  'composer_sample',
        schedule_interval=datetime.timedelta(days=1),
        default_args=default_dag_args) as dag:

    PULL_DATA = PythonOperator(
    task_id='PULL_DATA',
    python_callable=execute_dataflow,
    op_kwargs={
        'main': '/home/airflow/gcs/data/main.py',
        'project': PROJECT,
        'temp_location': 'gs://dataflow-imports-test/temp',
        'setup_file': '/home/airflow/gcs/data/setup.py',
        'max_num_workers': 2,
        'output': 'gs://dataflow-imports-test/output',
        'project_id': PROJECT_ID})
    PULL_DATA

没有更改Python可调用。但是，使用此配置我仍然会收到错误。

Next step，在Google云端平台（GCP）控制台中，我通过导航菜单转到“Composer”，然后单击环境的名称。在“PyPI包”选项卡上，我添加了zeep和googleads，然后单击“提交”。更新环境需要一段时间，但它确实有效。

在此步骤之后，我的管道能够导入依赖项并成功运行。我还尝试使用GCP控制台上指示的依赖项运行DAG，但不是在setup.py的要求中运行。工作流程再次中断，但在不同的地方。所以一定要在两个地方都注明。

---