Airflow中读取XCom值拼接SQL传入算子写入数据库表的方法

核心注意：XCom值是任务运行时才生成的，禁止在DAG顶层解析阶段直接调用xcom_pull，否则只会拿到空值，所有拉取逻辑必须放在算子的模板渲染段、或任务执行的可调用函数内

方案1：直接传入BigQuery相关算子执行 #

BigQueryInsertJobOperator（新版替代旧版BigQueryOperator）的query参数默认开启Jinja模板渲染，不需要额外写Python转换逻辑，直接在SQL语句中通过模板语法拉取XCom值拼接即可。
示例代码：

from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

# 假设之前推送key=db_val的任务ID为extract_db_val
insert_bq_task = BigQueryInsertJobOperator(
    task_id="insert_data_to_bq",
    configuration={
        "query": {
            "query": """
                INSERT INTO `项目ID.数据集名.表名` (目标字段1, 目标字段2)
                VALUES (
                    '{{ ti.xcom_pull(task_ids="extract_db_val", key="db_val") | replace("'", "\\'") }}',
                    CURRENT_TIMESTAMP()
                )
            """,
            "useLegacySql": False
        }
    },
    gcp_conn_id="你配置的GCP连接ID"
)

# 如果你仍在使用旧版BigQueryOperator，写法更简单：
# from airflow.providers.google.cloud.operators.bigquery import BigQueryOperator
# insert_bq_task = BigQueryOperator(
#     task_id="insert_data_to_bq",
#     sql="INSERT INTO 表名 (col) VALUES ('{{ ti.xcom_pull(task_ids=\"extract_db_val\", key=\"db_val\") }}')",
#     gcp_conn_id="gcp_conn",
#     use_legacy_sql=False
# )

注意事项：

• 如果拉取的值是字符串类型，必须用单引号包裹，同时通过replace过滤器转义值内部的单引号，避免SQL语法报错
• 如果拉取的值是数值、布尔类型，不需要外层包裹单引号
• 如果是列表、字典等复杂结构，可加| tojson过滤器转成合法SQL字符串

方案2：传入BashOperator执行 #

BashOperator的bash_command参数同样原生支持Jinja模板，可以直接在命令段拉取XCom值，拼接后通过数据库命令行客户端执行。
示例代码（以调用gcloud bq命令行写入BigQuery为例，可替换为MySQL、PG等其他数据库的客户端命令）：

from airflow.operators.bash import BashOperator

bash_insert_task = BashOperator(
    task_id="insert_data_via_bash",
    bash_command="""
        # 拉取XCom值存入shell变量，转义单引号避免shell语法错误
        TARGET_VAL='{{ ti.xcom_pull(task_ids="extract_db_val", key="db_val") | replace("'", "'\\''") }}'
        # 拼接SQL调用bq客户端执行
        bq query --use_legacy_sql=false \
        "INSERT INTO \`项目ID.数据集名.表名\` (target_col) VALUES ('${TARGET_VAL}')"
    """
)

复杂逻辑适配方案 #

如果XCom值需要做复杂校验、清洗、多字段拼接，直接写在Jinja模板里可读性差，可以单独用PythonOperator提前拼好完整SQL，再把SQL作为XCom传递给下游执行算子：

from airflow.operators.python import PythonOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

def build_insert_sql(ti):
    # 拉取目标值
    raw_val = ti.xcom_pull(task_ids="extract_db_val", key="db_val")
    # 做自定义清洗、校验、转义
    if not raw_val:
        raise ValueError("拉取到的XCom值为空，终止执行")
    safe_val = str(raw_val).replace("'", "\\'")
    # 拼接完整SQL
    return f"INSERT INTO 表名 (target_col) VALUES ('{safe_val}')"

build_sql_task = PythonOperator(
    task_id="build_insert_sql",
    python_callable=build_insert_sql
)

run_sql_task = BigQueryInsertJobOperator(
    task_id="run_insert_sql",
    configuration={
        "query": {
            # 直接拉取上游Python任务返回的完整SQL执行
            "query": "{{ ti.xcom_pull(task_ids='build_insert_sql') }}",
            "useLegacySql": False
        }
    }
)

build_sql_task >> run_sql_task

通用避坑提示 #

• 调用xcom_pull时必须指定正确的task_ids，也就是当初执行xcom_push(key="db_val", value=xxx)的任务ID，不指定会默认拉取所有上游任务的返回值，极易出现值匹配错误
• 拼接SQL前必须对传入值做特殊字符转义，避免SQL语法错误，若值来自外部不可信输入，必须额外做内容校验避免SQL注入
• 不要在DAG顶层、算子初始化的非模板参数位置直接调用xcom_pull，这类代码在DAG定时解析时就会执行，此时任务还未运行，XCom无有效值

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