Google数据流：导入自定义Python模块

Question

我尝试在Google Cloud数据流中运行Apache Beam管道（Python），由Google Cloud Coomposer中的DAG触发。

我的dags文件夹在各自的GCS桶中的结构如下：

/dags/
  dataflow.py <- DAG
  dataflow/
    pipeline.py <- pipeline
    setup.py
    my_modules/
      __init__.py
      commons.py <- the module I want to import in the pipeline

setup.py是非常基本的，但是根据Apache Beam文档和SO上的答案：

import setuptools

setuptools.setup(setuptools.find_packages())

在DAG文件（dataflow.py）中，我设置了setup_file选项并将其传递给Dataflow：

default_dag_args = {
    ... ,
    'dataflow_default_options': {
        ... ,
        'runner': 'DataflowRunner',
        'setup_file': os.path.join(configuration.get('core', 'dags_folder'), 'dataflow', 'setup.py')
    }
}

在管道文件（pipeline.py）中，我尝试使用

from my_modules import commons

但这失败了。Google Cloud Composer（Apache Airflow）中的html" target="_blank">日志显示：

gcp_dataflow_hook.py:132} WARNING - b'  File "/home/airflow/gcs/dags/dataflow/dataflow.py", line 11\n    from my_modules import commons\n           ^\nSyntaxError: invalid syntax'

setup.py文件背后的基本思想记录在这里

此外，在SO上也有类似的问题帮助了我：

Google Dataflow-导入自定义python模块失败

Dataflow/apachebeam：管理自定义模块依赖项

我实际上想知道为什么我的管道会出现语法错误，而不是模块找不到类型的错误...

Answer 1

我试图重现您的问题，然后尝试解决它，所以我创建了与您已经拥有的相同的文件夹结构：

/dags/
  dataflow.py
  dataflow/
     pipeline.py -> pipeline
     setup.py
     my_modules/
        __init__.py
        common.py

因此，为了使其正常工作，我所做的更改是将这些文件夹复制到实例正在运行的位置，以便代码能够找到它，例如在实例的/tmp/文件夹中。

所以，我的DAG应该是这样的：

1-我首先声明我的论点：

default_args = {
   'start_date': datetime(xxxx, x, x),
   'retries': 1,
   'retry_delay': timedelta(minutes=5),
   'dataflow_default_options': {
       'project': '<project>',
       'region': '<region>',
       'stagingLocation': 'gs://<bucket>/stage',
       'tempLocation': 'gs://<bucket>/temp',
       'setup_file': <setup.py>,
       'runner': 'DataflowRunner'
   }
} 

2-在此之后，我创建了DAG，在运行数据流任务之前，我将上面创建的整个文件夹目录复制到实例任务t1的/tmp/文件夹中，然后，我从/tmp/目录任务t2运行管道：

with DAG(
    'composer_df',
     default_args=default_args,
     description='datflow dag',
     schedule_interval="xxxx") as dag:

     def copy_dependencies():
          process = subprocess.Popen(['gsutil','cp', '-r' ,'gs://<bucket>/dags/*', 
          '/tmp/'])
          process.communicate()


     t1 = python_operator.PythonOperator(
        task_id='copy_dependencies',
        python_callable=copy_dependencies,
        provide_context=False
     )


     t2 = DataFlowPythonOperator(task_id="composer_dataflow", 
          py_file='/tmp/dataflow/pipeline.py', job_name='job_composer')

     t1 >> t2

这就是我创建DAG文件数据流的方式。py，然后，在管道中。py要导入的包如下所示：

from my_modules import commons

它应该可以正常工作，因为VM可以理解文件夹目录。