Задание потока данных через Airflow DAG - PullRequest
Новая
       58

Задание потока данных через Airflow DAG

0 голосов
/ 25 сентября 2019

Я пытаюсь выполнить файл Python конвейера луча Apache, используя обработчик потока данных через BashOperator в Airflow.У меня есть идея, как передать параметры динамически в файл Python.Я с нетерпением жду оптимизации параметров, чтобы избежать отправки всех параметров по отдельности.Пример фрагмента:

text_context.py 

import sys

def run_awc_orders(*args, **kwargs):
    print("all arguments -> ",  args)

if __name__ == "__main__":
    print("all params -> ", sys.argv)
    run_awc_orders( sys.argv[1],  sys.argv[2], sys.argv[3])

my_dag.py 

test_DF_job = BashOperator(
    task_id='test_DF_job',
    provide_context=True,
    bash_command="python /usr/local/airflow/dags/test_context.py {{ execution_date }} {{ next_execution_date }} {{ params.db_params.new_text }}  --runner DataflowRunner --key path_to_creds_json_file --project project_name --staging_location staging_gcp_bucket_location --temp_location=temp_gcp_bucket_location --job_name test-job",
    params={
              'db_params': {
                'new_text': 'Hello World'
              }
            },
    dag=dag
)

Итак, вот чтомы можем увидеть в логах пользовательского интерфейса воздушного потока. 

[2019-09-25 06:44:44,103] {bash_operator.py:128} INFO - all params ->  ['/usr/local/airflow/dags/test_context.py', '2019-09-23T00:00:00+00:00', '2019-09-24T00:00:00+00:00', '127.0.0.1']
[2019-09-25 06:44:44,103] {bash_operator.py:128} INFO - all arguments ->  ('2019-09-23T00:00:00+00:00', '2019-09-24T00:00:00+00:00', '127.0.0.1')
[2019-09-25 06:44:44,106] {bash_operator.py:132} INFO - Command exited with return code 0

1 Ответ

0 голосов
/ 26 сентября 2019

Я считаю, что рекомендуемый способ сделать это с Airflow DataflowPythonOperator, который напрямую получает опции Python и Dataflow.

Вы бы сделали что-то вроде этого: 

test_DF_job = DataflowPythonOperator(
  py_file='/usr/local/airflow/dags/test_context.py',
  py_options=[...],
  dataflow_default_options={...},
  dag=dag
)
...