Airflow新手求助：如何用内置组件传递SqlSensor查询结果至后续任务？

可以通过SqlOperator结合分支逻辑+XCom实现你的需求，无需两次查询，也不用完全自定义组件，具体方案如下：

核心思路 #

用MySqlOperator一次性执行查询并将结果推送到XCom，再通过分支任务判断结果是否为空，决定是否触发后续的导出流程。

1. 执行查询并推送结果到XCom #

使用MySqlOperator（Airflow内置的MySQL操作符），开启do_xcom_push=True，查询结果会自动保存到XCom中。如果结果数据量较大，可以搭配你提到的自定义XCom后端（比如将结果存到本地文件或对象存储），避免占用元数据库资源。

示例代码片段：

from airflow.providers.mysql.operators.mysql import MySqlOperator
from airflow.operators.python import BranchPythonOperator
from airflow.operators.dummy import DummyOperator
from airflow.models.dag import DAG
from datetime import datetime

def check_records_exists(ti):
    # 从XCom拉取查询结果
    query_result = ti.xcom_pull(task_ids='execute_target_query')
    # 根据查询结果的格式判断是否有记录（示例为列表格式，判断长度）
    return 'export_records_to_file' if len(query_result) > 0 else 'skip_export'

with DAG(
    dag_id='mysql_export_conditional',
    start_date=datetime(2024, 1, 1),
    schedule_interval='@daily',
    catchup=False
) as dag:
    # 执行目标查询并推送结果到XCom
    execute_query = MySqlOperator(
        task_id='execute_target_query',
        mysql_conn_id='your_mysql_connection_id',  # 提前在Airflow配置好的MySQL连接ID
        sql='SELECT * FROM your_target_table WHERE your_condition;',  # 你的查询语句
        do_xcom_push=True,
    )

    # 分支判断是否有记录
    branch_task = BranchPythonOperator(
        task_id='check_records',
        python_callable=check_records_exists,
    )

    # 导出任务（替换成你实际的导出逻辑，比如用PythonOperator写文件导出，或用内置的转储操作符）
    export_task = DummyOperator(task_id='export_records_to_file')
    # 跳过分支的占位任务
    skip_task = DummyOperator(task_id='skip_export')

    # 构建任务依赖
    execute_query >> branch_task >> [export_task, skip_task]

2. 替换导出任务为实际逻辑 #

你可以把export_task替换成具体的导出实现：

• 如果是导出到本地文件，用PythonOperator结合MySqlHook拉取XCom中的结果并写入文件
• 如果是导出到云存储（如S3/GCS），可以用MySqlToGCSOperator/MySqlToS3Operator等Airflow内置的转储操作符（注意这类操作符会自己执行查询，如果你想复用之前的查询结果，还是建议用PythonOperator处理XCom中的数据）

为什么不使用SqlSensor？ #

SqlSensor的核心作用是等待满足条件的记录出现，它只会返回是否满足条件的布尔值，不会保存查询结果。如果用SqlSensor，你需要再执行一次查询来获取数据，会造成两次数据库请求，不符合你「仅查询一次」的需求。

大数据量场景适配 #

如果查询结果数据量很大，默认XCom（存储在Airflow元数据库）会有性能问题，此时可以配置自定义XCom后端，让MySqlOperator的查询结果直接存储到文件或对象存储中，后续任务从该后端读取数据即可，无需修改核心逻辑。

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