Airflow Xcom如何拆分字符串并将拆分值存为独立变量供后续任务使用

你遇到的是XCom存储时被序列化为字符串、拉取后无法直接按列表使用的问题，按使用场景选对应方案即可：

• 优先推荐从上游根源修复：如果能修改上游推送XCom的任务代码，直接返回原生列表对象['val1', 'val2', 'val3']即可，不要手动转成字符串再返回。Airflow内置了对列表、字典这类基础Python类型的序列化能力，修复后下游用ti.xcom_pull()拉到的直接就是列表类型，不用额外解析，直接val1, val2, val3 = ti.xcom_pull(task_ids='上游task_id')就能拆分出三个独立值。

场景1：下游为Python类任务（PythonOperator、@task装饰器定义的任务等） #

直接用ast.literal_eval把字符串格式的列表转为原生Python列表再拆分即可，禁止用eval()做转换，存在任意代码执行的安全风险，literal_eval仅会解析合法的Python基础类型字面量，无安全问题。
示例代码：

import ast
from airflow.decorators import task

@task
def downstream_task(ti=None):
    # 拉取字符串格式的XCom值
    xcom_raw = ti.xcom_pull(task_ids="替换为你的上游任务ID")
    # 转为原生列表
    val_list = ast.literal_eval(xcom_raw)
    # 直接解构为三个独立变量
    val1, val2, val3 = val_list

    # 后续业务逻辑直接调用三个变量即可
    print(f"拆分后的值：{val1}, {val2}, {val3}")

场景2：下游为使用Jinja模板的非Python类任务（BashOperator、SparkSubmitOperator等） #

先在DAG初始化时注册一个自定义Jinja过滤器做列表解析，再在模板里拆分取值即可：

1. 注册自定义过滤器

import ast
from airflow import DAG

def str_to_list(input_str):
    return ast.literal_eval(input_str)

dag = DAG(
    dag_id="替换为你的DAG ID",
    # 补全start_date、schedule_interval等其他DAG配置
    user_defined_filters={"to_list": str_to_list}
)

2. 在算子模板字段中直接拆分取值，以BashOperator为例：

from airflow.operators.bash import BashOperator

use_split_val = BashOperator(
    task_id="use_split_val",
    bash_command="""
    # 模板渲染阶段直接拆分出三个值注入bash变量
    v1="{{ ti.xcom_pull(task_ids='上游任务ID') | to_list | first }}"
    v2="{{ ti.xcom_pull(task_ids='上游任务ID') | to_list | slice(1,2) | first }}"
    v3="{{ ti.xcom_pull(task_ids='上游任务ID') | to_list | last }}"
    echo "拆分结果：$v1 $v2 $v3"
    """,
    dag=dag
)

如果只是临时调试不想注册过滤器，且确定XCom字符串格式固定为['val1', 'val2', 'val3']这种格式，也可以直接用字符串方法快速拆分，正式生产场景不推荐使用，格式变动时容易出错：

# 临时取巧拆分方式
v1="{{ ti.xcom_pull(task_ids='上游任务ID').strip('[]').split(',')[0].strip("'\" ") }}"

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