Нет модуля с именем airfow.gcp - как запустить задание потока данных, использующее python3/beam 2.15?

Question

Когда я использую операторы / перехватчики, такие как BigQueryHook, я вижу сообщение о том, что эти операторы устарели и используют версию оператора airflow.gcp... Однако, когда я пытаюсь использовать его в своем даге, он терпит неудачу и говорит, что нет модуля с именем airflow.gcp. У меня самая последняя версия композитора воздушного потока с бета-функциями, python3. Можно ли как-нибудь установить эти операторы?

Я пытаюсь запустить задание потока данных в python 3, используя луч 2.15. Я пробовал оператор virtualenv, но это не сработало, потому что он позволяет использовать только python2.7. Как я могу это сделать?

Answer 1

Новейшая версия Airflow, доступная в Composer, - это 1.10.2 или 1.10.3 (в зависимости от региона). К тому времени эти операторы были в contrib раздел.

Сосредоточившись на том, как запускать задания потока данных Python 3 с помощью Composer, вам понадобится выпуск новой версии. Однако, если вам нужно немедленное решение, вы можете попробовать выполнить обратный перенос исправления.

В этом случае я определил DataFlow3Hook что расширяет нормальный DataFlowHook но это не жестко кодирует python2 в start_python_dataflow метод:

class DataFlow3Hook(DataFlowHook):
    def start_python_dataflow(
        ...
        py_interpreter: str = "python3"
    ):

        ...

        self._start_dataflow(variables, name, [py_interpreter] + py_options + [dataflow],
                             label_formatter)

Тогда у нас будет свой обычай DataFlowPython3Operator вызов нового хука:

class DataFlowPython3Operator(DataFlowPythonOperator):

    def execute(self, context):
        ...
        hook = DataFlow3Hook(gcp_conn_id=self.gcp_conn_id,
                            delegate_to=self.delegate_to,
                            poll_sleep=self.poll_sleep)
        ...
        hook.start_python_dataflow(
            self.job_name, formatted_options,
            self.py_file, self.py_options, py_interpreter="python3")

Наконец, в нашем DAG мы просто используем новый оператор:

task = DataFlowPython3Operator(
    py_file='/home/airflow/gcs/data/main.py',
    task_id=JOB_NAME,
    dag=dag)

Смотрите полный код здесь. Работа выполняется с Python 3.6:

Используемые детали среды и зависимости (минимальным примером была работа Beam):

softwareConfig:
  imageVersion: composer-1.8.0-airflow-1.10.3
  pypiPackages:
    apache-beam: ==2.15.0
    google-api-core: ==1.14.3
    google-apitools: ==0.5.28
    google-cloud-core: ==1.0.3
  pythonVersion: '3'

Дайте мне знать, если это сработает для вас. Если это так, я бы рекомендовал перенести код в плагин для удобства чтения и повторно использовать его в DAG.

Answer 2

В качестве альтернативы вы можете использовать PythonVirtualenvOperatorна старых версиях с воздушным потоком. Учитывая некоторый балочный конвейер (завернутый в функцию), сохраненный какdataflow_python3.py:

def main():
    import apache_beam as beam
    from apache_beam.options.pipeline_options import PipelineOptions
    from apache_beam.options.pipeline_options import SetupOptions
    import argparse
    import logging

    class ETL(beam.DoFn):
        def process(self, row):
            #do data processing


    def run(argv=None):
        parser = argparse.ArgumentParser()
        parser.add_argument(
            '--input',
            dest='input',
            default='gs://bucket/input/input.txt',
            help='Input file to process.'
            )
        known_args, pipeline_args = parser.parse_known_args(argv)
        pipeline_args.extend([
              '--runner=DataflowRunner',
              '--project=project_id',
              '--region=region',
              '--staging_location=gs://bucket/staging/',
              '--temp_location=gs://bucket/temp/',
              '--job_name=job_id',
              '--setup_file=./setup.py'
              ])

        pipeline_options = PipelineOptions(pipeline_args)
        pipeline_options.view_as(SetupOptions).save_main_session = True

        with beam.Pipeline(options=pipeline_options) as p:
            rows = (p | 'read rows' >> beam.io.ReadFromText(known_args.input))
            etl = (rows | 'process data' >> beam.ParDo(ETL()))


    logging.getLogger().setLevel(logging.DEBUG)
    run()

Вы можете запустить его, используя следующий файл DAG:

from airflow import DAG
from datetime import datetime, timedelta
from airflow.operators.python_operator import PythonVirtualenvOperator
import sys

import dataflow_python3 as py3 #import your beam pipeline file here 


default_args = {
    'owner': 'John Smith',
    'depends_on_past': False,
    'start_date': datetime(2016, 1, 1),
    'email': ['[e-mail]'],
    'email_on_failure': True,
    'email_on_retry': False,
    'retries': 3,
    'retry_delay': timedelta(minutes=1),
}

CONNECTION_ID = 'proj_id'

with DAG('Dataflow_Python3', schedule_interval='@once', template_searchpath=['/home/airflow/gcs/dags/'], max_active_runs=15, catchup=True, default_args=default_args) as dag:


    dataflow_python3 = PythonVirtualenvOperator(
        task_id='dataflow_python3',
        python_callable=py3.main, #this is your beam pipeline callable 
        requirements=['apache-beam[gcp]', 'pandas'],
        python_version=3,
        dag=dag
    )

dataflow_python3

Answer 3

Я запускал Python 3 Beam -2.17 с помощью DataflowTemplateOperator, и он работал как шарм.

Используйте команду ниже для создания шаблона:

python3 -m scriptname --runner DataflowRunner --project project_id --staging_location staging_location --temp_location temp_location --template_location template_location/script_metadata --region region --experiments use_beam_bq_sink --no_use_public_ips --subnetwork=subnetwork

scriptname будет именем вашего файла Python Dataflow (без расширения.py)

--template_location - место, где будет создан шаблон потока данных, не добавляйте к нему какие-либо расширения, например.json. Просто scriptname_metadata будет работать.

--experiments use_beam_bq_sink - этот параметр будет использоваться, если ваш приемник BigQuery, в противном случае его можно удалить.

import datetime as dt
import time
from airflow.models import DAG
from airflow.contrib.operators.dataflow_operator import DataflowTemplateOperator

lasthour = dt.datetime.now() - dt.timedelta(hours=1)

args = {
 'owner': 'airflow',
 'start_date': lasthour,
 'depends_on_past': False,
 'dataflow_default_options': {
     'project': "project_id",
     'staging_location': "staging_location",
     'temp_location': "temp_location",
     'region': "region",
     'runner': "DataflowRunner",
     'job_name': 'job_name' + str(time.time()),
 },
}
dag = DAG(
 dag_id='employee_dataflow_dag',
 schedule_interval=None,
 default_args=args
)

Dataflow_Run = DataflowTemplateOperator(
                task_id='dataflow_pipeline',
                template='template_location/script_metadata',
                parameters ={
                    'input':"employee.csv",
                    'output':'project_id:dataset_id.table',
                    'region':"region"
                        },
                gcp_conn_id='google_cloud_default',
                poll_sleep=15,
                dag=dag
            )

Dataflow_Run