Airflow 1.X中XCom返回None，Spotify令牌传递失败求助

问题背景 #

使用Airflow 1.X版本通过Docker容器部署，DAG包含两个BashOperator任务：

1. 第一个任务成功从Spotify获取访问令牌（日志可见有效令牌），但Airflow UI的Admin > XCom中无对应存储记录
2. 第二个任务通过task_instance.xcom_pull获取令牌时返回None，导致后续请求失败

相关DAG代码：

dag = DAG(
    'spotify_etl_pipeline',
    start_date=datetime(2024, 11, 11),
    schedule_interval=None,
    params={"playlist_id": DEFAULT_PLAYLIST_ID}
)

#Task1
get_access_token = BashOperator(
    task_id='get_access_token',
    bash_command=f"""
    set -x
    ACCESS_TOKEN=$(curl -X POST "https://accounts.spotify.com/api/token" \
        -H "Authorization: Basic {encoded_credentials}" \
        -H "Content-Type: application/x-www-form-urlencoded" \
        -d "grant_type=client_credentials" \
        | jq -r '.access_token')
    echo $ACCESS_TOKEN
    """,
    
    do_xcom_push=True,
    dag=dag
)

# Task 2
get_playlist_tracks = BashOperator(
    task_id='get_playlist_tracks',
    bash_command="""
    ACCESS_TOKEN="{{ task_instance.xcom_pull(task_ids='get_access_token') }}"
    echo "ACCESS TOKEN FROM XCom: $ACCESS_TOKEN"
    curl -X "GET" "https://api.spotify.com/v1/playlists/{{ dag_run.conf['playlist_id'] }}/tracks" \
        -H "Accept: application/json" \
        -H "Content-Type: application/json" \
        -H "Authorization: Bearer $ACCESS_TOKEN" \
        | jq '[.items[].track]' > /user/hadoop/spotify/track_data/raw/playlist_tracks.json
    """,  
    dag=dag 
)

问题原因 #

• Airflow 1.X BashOperator的XCom捕获规则：Airflow 1.X中，BashOperator默认仅将任务执行时最后一行stdout输出作为XCom值存储。开启set -x后，脚本会输出大量调试信息（如变量赋值、curl命令执行细节），这些信息可能成为stdout的最后一行，覆盖了echo $ACCESS_TOKEN的输出，导致令牌未被捕获为XCom。
• DAG解析阶段的字符串拼接风险：Task1使用f-string直接拼接encoded_credentials到bash_command，该操作在DAG解析阶段完成。若encoded_credentials包含特殊字符（如引号、反斜杠），可能导致bash脚本执行时出现隐式错误，虽然日志显示令牌获取成功，但stdout输出可能被干扰，影响XCom的正确捕获。
• XCom存储后端异常：Docker容器内的Airflow worker可能存在权限不足，无法写入元数据库；或元数据库配置错误，导致XCom记录无法持久化，进而在UI中无法查看。

改进方向 #

• 控制脚本输出内容：关闭set -x，或将调试信息重定向到stderr（如set -x 2>&1），确保echo $ACCESS_TOKEN是脚本输出的最后一行，让BashOperator能正确捕获该值作为XCom。
• 规范敏感信息的传递方式：避免在DAG解析阶段直接拼接敏感信息，改用Airflow的Connections存储encoded_credentials，通过模板语法（如{{ conn.spotify.extra_dejson['encoded_credentials'] }}）在任务执行阶段渲染，符合Airflow安全最佳实践，同时避免特殊字符导致的脚本问题。
• 排查XCom存储配置：检查Docker容器内Airflow的元数据库连接配置是否正确，确认worker进程有足够权限写入数据库；查看scheduler和worker的日志，排查是否有XCom存储相关的错误日志。
• 显式控制XCom推送：Airflow 1.X中，可通过在Bash脚本中调用Airflow CLI命令（如airflow xcom push）显式推送令牌，或改用PythonOperator完成令牌获取和XCom推送，更精准地控制XCom的存储内容。

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