如何通过Airflow实现从DB2到Snowflake的增量加载？

针对DB2源表到Snowflake目标表的增量加载，结合Airflow编排，主要有两种实用方案，下面分别说明具体实现方式：

一、基于时间戳/自增ID的增量拉取（通用方案） #

这是最易实现的模式，适合绝大多数中小数据量、准实时同步场景，核心是通过记录上次同步的"水位值"（时间戳或自增主键），每次只拉取该水位之后的新增/更新数据。

1. 初始化同步水位 #

在Airflow中用Variable存储上次同步的最大时间戳（或自增ID），比如执行命令初始化：

airflow variables set db2_last_sync_ts '2024-01-01 00:00:00'

2. Airflow DAG编排流程 #

整个ETL流程分为4个核心任务，依赖关系为：获取水位值 → 抽取增量数据 → 加载到Snowflake → 更新水位值

完整DAG代码示例

from airflow import DAG
from airflow.providers.ibm.db2.hooks.db2 import Db2Hook
from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
from airflow.models import Variable
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'etl_team',
    'depends_on_past': False,
    'start_date': datetime(2024, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    dag_id='db2_to_snowflake_incremental_sync',
    default_args=default_args,
    schedule_interval='@hourly',  # 按需调整同步频率
    catchup=False
) as dag:

    def fetch_last_sync_watermark():
        """获取上次同步的时间戳水位"""
        return Variable.get('db2_last_sync_ts')

    def extract_incremental_from_db2(**context):
        """从DB2抽取增量数据"""
        last_sync_ts = context['ti'].xcom_pull(task_ids='fetch_watermark')
        db2_hook = Db2Hook(db2_conn_id='db2_prod_conn')  # 提前在Airflow配置DB2连接

        # 构建增量查询SQL，假设源表有update_ts字段记录最后更新时间
        extract_sql = f"""
            SELECT id, col1, col2, update_ts
            FROM db2_source_schema.source_table
            WHERE update_ts > TIMESTAMP('{last_sync_ts}')
            ORDER BY update_ts ASC
        """

        # 用pandas读取数据，大数据量建议写入S3/GCS而非XCom
        incremental_df = db2_hook.get_pandas_df(extract_sql)
        
        # 推送数据和本次最大时间戳到XCom
        if not incremental_df.empty:
            context['ti'].xcom_push(
                key='incremental_csv', 
                value=incremental_df.to_csv(index=False, encoding='utf-8')
            )
            context['ti'].xcom_push(
                key='current_max_ts', 
                value=incremental_df['update_ts'].max().strftime('%Y-%m-%d %H:%M:%S')
            )

    def load_to_snowflake(**context):
        """将增量数据加载到Snowflake"""
        incremental_csv = context['ti'].xcom_pull(task_ids='extract_incremental', key='incremental_csv')
        if not incremental_csv:
            return

        snowflake_hook = SnowflakeHook(snowflake_conn_id='snowflake_prod_conn')  # 提前配置Snowflake连接

        # 创建临时阶段存储用于加载CSV
        snowflake_hook.run("""
            CREATE OR REPLACE TEMP STAGE temp_incremental_stage
            FILE_FORMAT = (TYPE = CSV FIELD_OPTIONALLY_ENCLOSED_BY = '"' SKIP_HEADER = 1)
        """)

        # 上传CSV到临时阶段
        snowflake_hook.stage_file(
            file_obj=incremental_csv,
            stage_location='@temp_incremental_stage',
            file_name='db2_incremental_data.csv'
        )

        # 用COPY INTO加载，大数据量推荐此方式；小数据量可直接用pandas.to_sql
        snowflake_hook.run("""
            COPY INTO snowflake_target_schema.target_table
            FROM @temp_incremental_stage/db2_incremental_data.csv
            MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE
            ON_ERROR = CONTINUE
        """)

    def update_sync_watermark(**context):
        """更新同步水位值"""
        current_max_ts = context['ti'].xcom_pull(task_ids='extract_incremental', key='current_max_ts')
        if current_max_ts:
            Variable.set('db2_last_sync_ts', current_max_ts)

    # 定义任务
    fetch_watermark_task = PythonOperator(
        task_id='fetch_watermark',
        python_callable=fetch_last_sync_watermark
    )

    extract_task = PythonOperator(
        task_id='extract_incremental',
        python_callable=extract_incremental_from_db2,
        provide_context=True
    )

    load_task = PythonOperator(
        task_id='load_to_snowflake',
        python_callable=load_to_snowflake,
        provide_context=True
    )

    update_watermark_task = PythonOperator(
        task_id='update_watermark',
        python_callable=update_sync_watermark,
        provide_context=True
    )

    # 设置任务依赖
    fetch_watermark_task >> extract_task >> load_task >> update_watermark_task

关键注意事项 #

• 水位可靠性：DB2源表的update_ts字段必须是每次数据更新时自动刷新的（可通过DB2触发器实现），否则会遗漏更新数据。
• 大数据量优化：不要用XCom传递大体积数据，应将抽取的增量数据写入云存储（S3/GCS），再通过Snowflake的COPY INTO加载，效率提升明显。
• 去重与更新：如果需要处理重复或更新数据，将COPY INTO替换为MERGE语句，根据主键匹配进行插入、更新或删除操作。

二、基于DB2 CDC的实时增量同步（高实时性场景） #

如果需要近实时（分钟级以内）同步，可采用DB2的变更数据捕获（CDC）方案：

1. 开启DB2的CDC功能（如IBM InfoSphere CDC或DB2原生CDCapture），配置捕获源表的INSERT/UPDATE/DELETE操作。
2. 将CDC变更日志输出到中间消息队列（如Kafka）。
3. 通过Snowflake的Kafka Connector直接消费队列数据到Snowflake临时表，或用Airflow的Kafka相关Operator读取数据。
4. Airflow负责编排合并变更数据到目标表、CDC任务状态监控、异常告警等任务。

合并变更数据的Airflow任务示例

def merge_cdc_data_to_target(**context):
    snowflake_hook = SnowflakeHook(snowflake_conn_id='snowflake_prod_conn')
    # 假设CDC数据已同步到临时表cdc_staging，包含op_type字段（INSERT/UPDATE/DELETE）
    merge_sql = """
        MERGE INTO snowflake_target_schema.target_table t
        USING snowflake_staging_schema.cdc_staging s
        ON t.id = s.id
        WHEN MATCHED AND s.op_type = 'UPDATE' THEN 
            UPDATE SET t.col1 = s.col1, t.col2 = s.col2, t.update_ts = s.update_ts
        WHEN MATCHED AND s.op_type = 'DELETE' THEN 
            DELETE
        WHEN NOT MATCHED AND s.op_type = 'INSERT' THEN 
            INSERT (id, col1, col2, update_ts) VALUES (s.id, s.col1, s.col2, s.update_ts)
    """
    snowflake_hook.run(merge_sql)

内容的提问来源于stack exchange，提问作者wulalala