如何将Airflow中xcom_pull返回的列表转为SQL IN子句可用的逗号分隔字符串

问题根源 #

你当前的代码存在两个核心问题：

1. XCom存储的数据格式错误：fetchall()返回的是元组列表（如[(1,), (2,)]），直接存储会导致后续处理时包含冗余的元组结构。
2. 模板解析时机错误：parameters中的Python代码在DAG解析阶段执行，此时Jinja模板字符串{{ti.xcom_pull(...)}}还未被渲染为实际的ID列表，最终传递给SQL的是模板字符串本身，而非预期的逗号分隔ID。

修正方案 #

步骤1：修正Task1的XCom存储内容

先从数据库查询中提取纯ID列表，而非元组列表：

def get_processed_files(ti):
    sql = "select id from files where status='DONE'"
    pg_hook = PostgresHook(postgres_conn_id="conn_id")
    # 用get_records简化查询，直接获取结果集
    file_tuples = pg_hook.get_records(sql)
    # 提取每个元组中的ID，得到纯整数列表
    file_ids = [item[0] for item in file_tuples]
    ti.xcom_push(key="file_ids_to_update", value=file_ids)

步骤2：选择合适的Task2实现方式

方式一：直接在PostgresOperator的SQL中用Jinja过滤器处理

利用Jinja的join过滤器在SQL模板中直接将列表转为逗号分隔字符串：

archive_file = PostgresOperator(
    task_id="archive_processed_file",
    postgres_conn_id="upflow",
    sql="""
        update files set update_date=now() 
        where id in ({{ ti.xcom_pull(key='file_ids_to_update') | join(',') }})
    """
)

若ID是字符串类型，需添加引号处理：where id in ('{{ ti.xcom_pull(key='file_ids_to_update') | map('string') | join("','") }}')

方式二：用PythonOperator实现更灵活的逻辑

如果需要额外的判断（比如无ID时跳过执行），推荐用PythonOperator封装数据库操作：

def archive_files(ti):
    file_ids = ti.xcom_pull(key='file_ids_to_update')
    # 无待更新ID时直接返回，避免执行无效SQL
    if not file_ids:
        return
    
    sql = "update files set update_date=now() where id in %s"
    pg_hook = PostgresHook(postgres_conn_id="upflow")
    # 将列表转为元组，PostgresHook会自动处理为合法的IN子句格式
    pg_hook.execute(sql, parameters=(tuple(file_ids),))

archive_file = PythonOperator(
    task_id="archive_processed_file",
    python_callable=archive_files,
)

为什么原代码不生效？ #

原代码中parameters={"list_of_ids": ",".join(map(str, "{{ti.xcom_pull(key='file_ids_to_update')}}"))}是在DAG加载阶段执行的，此时Jinja模板还未被渲染，"{{ti.xcom_pull(...)}}"只是一个普通字符串，而非实际的ID列表。最终传递给SQL的是这个模板字符串本身，导致IN子句语法错误。

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