Воздушный поток, как передать переменные из задачи BashOperator в другую - PullRequest
Новая
       21

Воздушный поток, как передать переменные из задачи BashOperator в другую

0 голосов
/ 14 октября 2019

Я пытаюсь найти наиболее общий / эффективный способ передачи переменных из одной задачи BashOperator в другую. Я придумал решение, в котором я перенес вывод первой задачи BashOperator в xcom. Затем эта строка извлекается задачей PythonOperator, которая анализирует строку с парами ключ-значение, которые затем передаются в xcom. Наконец, эти пары kv могут быть извлечены и использованы второй задачей BashOperator.

Я хотел бы получить некоторые комментарии от более опытных пользователей воздушного потока, является ли этот подход чрезмерно сложным или это нормально? 

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
import datetime as dt 
from airflow.utils.dates import days_ago    

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': dt.datetime(2019,10,14,10,0),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': dt.timedelta(minutes=1),
}

dag = DAG('communication-between-tasks-v1',
      catchup=False,
      default_args=default_args,
      schedule_interval='*/5 * * * *')

def parse_function(**context):
    ti = context['ti']
    msg = ti.xcom_pull('start_task')
    parsed_output_parameters = dict([x.split('=') for x in msg.split()])

    # Option1: Dictionary parsed from the string is pushed to xcom
    ti.xcom_push(key='parameters', value=parsed_output_parameters)

    # Option2: We can push every key value separately
    for k, v in parsed_output_parameters.items():
        ti.xcom_push(key=k, value=v)


start_task = BashOperator(
    task_id='start_task',
    bash_command='echo "FILE1=file1.h5 FILE2=file2.txt VARIABLE=400"',
    xcom_push=True,
    dag=dag)

parse = PythonOperator(
    task_id='parse',
    python_callable=parse_function,
    provide_context=True,
    dag=dag
)

end_task = BashOperator(
    task_id='end_task',
    bash_command='echo start_task params: {{ ti.xcom_pull(task_ids="parse", 
    key="parameters")["FILE1"] }} \
    {{ ti.xcom_pull(task_ids="parse", key="FILE2") }}',
    dag=dag
)

start_task >> parse >> end_task

1 Ответ

0 голосов
/ 15 октября 2019

Xcom лучше всего работает с действительно небольшими объемами передаваемых данных и должен использоваться экономно (так как все это записывается в базу данных воздушного потока).

Если у вас есть 2 разные задачи BashOperator и вы хотите передать данныеот одного к другому, почему бы просто не записать вывод в файл в первом задании и прочитать его во втором? (Вы можете включить строку во вторую задачу BashOperator, которая проверяет, содержит ли файл данные и rm файл после прочтения его содержимого.)

...