airflow中跨dag/task的數據傳送

默認在airflow中每個task都是獨立的進程，無法進行數據交換，但airflow還提供了一個XCom功能，以滿足用戶的類似需求

下面我們創建兩個dag,其中一個push數據，一個pull數據，如下的dag中的task（push）執行完后會推送一條數據到xcom表，key=push ,value=True

dag = DAG(
    dag_id='migrate_mongo',
    default_args=args,
    schedule_interval='0 17 * * *',
    catchup=False
)


def push(**kwargs):
    kwargs['ti'].xcom_push(key='status', value=True)

task1 = PythonOperator(task_id='push',
                       python_callable=push,
                       provide_context=True,
                       dag=dag)

另起一個dag,創建一個task（pull）可以將上面的dag中的數據獲取到

dag = DAG(
    dag_id='migrate_mongo2',
    default_args=args,
    schedule_interval='0 17 * * *',
    catchup=False
)


def pull(**kwargs):
    execution_date = kwargs['execution_date']
    v1 = XCom.get_one(dag_id='migrate_mongo', task_id='push', execution_date=execution_date)
    print(v1)

task2 = PythonOperator(task_id='pull',
                       python_callable=pull,
                       provide_context=True,
                       dag=dag)

兩個dag之間其實通過airflow底層的一張xcom表進行數據交換的