Save result of operator in Apache Airflow

Question

Several operators allow to pull data but I never managed to use the results.

For example: https://github.com/apache/incubator-airflow/blob/master/airflow/contrib/operators/bigquery_get_data.py

This operator can be called as follow:

get_data = BigQueryGetDataOperator(
      task_id='get_data_from_bq',
      dataset_id='test_dataset',
      table_id='Transaction_partitions',
      max_results='100',
      selected_fields='DATE',
      bigquery_conn_id='airflow-service-account'
      )

Yet, get_data is of type DAG but line 116 says "return table_data". To be clear, the operator works and retrieve the data, I just don't understand how to use the data retrieve/where it is located.

How do I get the data using "get_data" above?

kaxil · Accepted Answer · 2019-07-17 18:52:08Z

11

+50

The way you would use get_data is in the next task can be a PythonOperator which you can then use to process the data.

get_data = BigQueryGetDataOperator(
      task_id='get_data_from_bq',
      dataset_id='test_dataset',
      table_id='Transaction_partitions',
      max_results='100',
      selected_fields='DATE',
      bigquery_conn_id='airflow-service-account'
      )

def process_data_from_bq(**kwargs):
      ti = kwargs['ti']
      bq_data = ti.xcom_pull(task_ids='get_data_from_bq')
      # Now bq_data here would have your data in Python list
      print(bq_data)

process_data = PythonOperator(
      task_id='process_data_from_bq',
      python_callable=process_bq_data,
      provide_context=True
      )

get_data >> process_data

PS: I am the author of BigQueryGetDataOperator and Airflow committer / PMC

edited Jul 17, 2019 at 18:52

answered Jan 2, 2019 at 12:08

kaxil

19.1k3 gold badges67 silver badges87 bronze badges

Sign up to request clarification or add additional context in comments.

3 Comments

dboshardy Over a year ago

Nitpicking here: your print statement will fail because data is undeclared.

kaxil Over a year ago

@dboshardy Updated

user13237611 Over a year ago

"An XCom is identified by a key (essentially its name), as well as the task_id and dag_id it came from. They can have any (serializable) value, but they are only designed for small amounts of data; do not use them to pass around large values, like dataframes." (source: Airflow documentation)

SergiyKolesnikov · Accepted Answer · 2019-01-01 22:36:36Z

1

The return value is saved in an Xcom. You can access it from another operator as it is shown in this example.

data = ti.xcom_pull(task_ids='get_data_from_bq')

answered Jan 1, 2019 at 22:36

SergiyKolesnikov

7,8254 gold badges33 silver badges50 bronze badges

Collectives™ on Stack Overflow

Save result of operator in Apache Airflow

2 Answers 2

3 Comments

Comments

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

3 Comments

Comments

Related