Airflow中通过task_ids拉取XCom值失败问题排查

你的pull_pubsub_message_task处于recommender_feed这个Task Group中，Airflow会自动为Task Group内任务的XCom条目添加Task Group名称作为前缀，直接用pull_pubsub_message作为task_id拉取XCom，自然找不到对应值。

修改extract_datetime函数里的xcom_pull调用，带上Task Group的完整前缀：

def extract_datetime(**kwargs):
    # 加上Task Group名称作为前缀
    message_data = kwargs['ti'].xcom_pull(task_ids='recommender_feed.pull_pubsub_message')
    print("MESSAGE FROM PUB/SUB", message_data)

    decoded_data = [base64.b64decode(message['message']['data']) for message in message_data]
    print("pubsub data received: ", decoded_data)
    for d in decoded_data:
        decoded_d = json.loads(str(d.decode("utf-8")))
        payload_datetime = datetime.strptime(decoded_d['run_datetime'], '%Y-%m-%d %H:%M:%S')
        extracted_date = payload_datetime.date()

        kwargs['ti'].xcom_push(key='extracted_date', value=extracted_date)

• 你可以在Airflow UI的XCom页面确认：对应条目的task_id字段会显示为recommender_feed.pull_pubsub_message，这就是需要指定的完整task_id
• 如果存在多层嵌套的Task Group，需要按照外层TaskGroup.内层TaskGroup.task_id的格式指定完整路径
• 若担心跨DAG或历史数据干扰，可以在xcom_pull中添加include_prior_dates=False参数，确保只拉取当前DAGRun的XCom值

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