How to transfer large volumes of data (millions of rows) between tasks in Airflow?

It is not recommended to pass large volumes of data directly between tasks via XCom in Airflow, as this can lead to performance and storage issues. Instead, it is better to use external storage:

• Save data in a database, such as PostgreSQL or ClickHouse.
• Use distributed file systems or cloud storage (S3, GCS).
• Pass only links or keys to this data between tasks.

Example:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def extract(**kwargs):
    # Save data to a file or database
    data_path = '/tmp/large_data.csv'
    # ... save code ...
    kwargs['ti'].xcom_push(key='data_path', value=data_path)

def process(**kwargs):
    ti = kwargs['ti']
    data_path = ti.xcom_pull(key='data_path', task_ids='extract')
    # Load and process data from data_path

with DAG('example_dag', start_date=datetime(2023,1,1), schedule_interval='@daily') as dag:
    t1 = PythonOperator(task_id='extract', python_callable=extract)
    t2 = PythonOperator(task_id='process', python_callable=process)
    t1 >> t2

This approach allows efficient handling of large data volumes without overloading Airflow.