Pass None (default) to remove the filter. If provided, only XCom with matching keys will be returned. tikey ( ) The TaskInstanceKey to look up the XCom for. Revised code with 2. If multiple XCom entries match the criteria, an arbitrary one is returned. This feature allows you to access the output of tasks using a simple task.output expression. If you are using Airflow 2.0, the code can actually be simplified to use the new XComArg feaure. Task_id='hello_world', python_callable=hello_world, dag=dag, provide_context=True In traditional Airflow operators, output can be pushed to XCom using the xcompush method. Load_DATE_FUNCTION>push_dag_status_to_xcomįrom _operator import PythonOperatorĭef hello_world(ti, execution_date, **context): output to extract outputs from traditional operators. Push_dag_status_to_xcom = PythonOperator( Given below is an Airflow DAG for extracting data and transforming data showing XCom push pull syntax. Kwargs.xcom_push(key='dag_status', value='no_successful_runs') Kwargs.xcom_push(key='dag_status', value='failed') Kwargs.xcom_push(key='dag_status', value='success') ('Latest DAG run ID: %s, Status: %s', latest_dag_run_id, latest_dag_run_status) Example of pulling an XCom from another DAG from airflow.models import TaskInstance def pullfunction(kwargs): value TaskInstance. XComs are stored in Airflows metadata database and can be 'pushed' (sent) or 'pulled' (received) by tasks. I see a lot of examples on how to use xcompush and xcompull with PythonOperators in Airflow. They are primarily used for inter-task communication. Latest_dag_run_status = latest_dag_run.state XComs, short for 'cross-communication', are designed to allow tasks to exchange messages or small amounts of data. Latest_dag_run_id = latest_dag_run.run_id If latest_dag_run is None or dag_run.execution_date > latest_dag_run.execution_date: # Add this function at the end of each DAG (A, B, and C)įor dag_run in DagRun.find(dag_id='testing', state='success'): Kwargs.xcom_push(key='load_date', value=load_date) Kwargs.xcom_push(key='partition_date', value=partition_date) Kwargs.xcom_push(key='Snapshot_Month_Date', value=str(Snapshot_Month_Date)) Snapshot_Month_Date = Snapshot_Month_Date.strftime('%Y-%m-%d') Best Practices: For large data, store the data externally and push/pull the reference or path. However, its important to be cautious as large messages can impact the databases performance. Load_date = str(today_date.strftime('%Y-%m-%d')) Default Behavior: By default, Airflow does not impose a hard limit on the size of the messages that can be passed via XCom. Its because the entire data argument can be templated. I suspect you might be wondering how its used for the start key in the data dict. Partition_date = str(Snapshot_Month_Date.strftime('%Y-%m-%d').replace('-', '')) user3595632 For the SimpleHttpOperator in that example, the data parameter is a template field so Jinja templating is completely fine to use. Snapshot_Month_Date = today_date.replace(day=1) - timedelta(days=1) Today_date = datetime.now() + timedelta(hours=5, minutes=30) Usage: - push and pull Uses the Airflow metadatabase (Postgres / MySQL) xcompush( key ‘returnvalue’, value ‘my value’. LOAD DAG : from datetime import timedelta, datetimeįrom import gcs_to_bqįrom _operator import PythonOperatorįrom _operator import BigQueryOperator Information on how to use xcom you can get from airflow examples. From the other task get table2 using xcom.pull and read it with df.readsql (). The custom python operator is pushing the required information in the XCOM key s3datasettype, this key will be used in the downstream task. Pass the name of the table using xcom.push. while doing so i am getting returned output as None. In task1 you can download data from table1 in some dataframe, process it and save in another table2 (df.tosql ()). I have a DAG where i am doing some data loading activity and at the end i am fetching the DAG run state in a Xcom and using that Xcom i want to pull its value to another dag.

Airflow xcom push example