Airflow中SparkSubmitOperator的XCom Push失效问题求助

核心原因 #

Airflow 2.0.0版本的SparkSubmitOperator，即便设置do_xcom_push=True，默认推送的并非Spark作业的日志内容，而是SparkSubmit命令的执行返回码或空值（取决于作业是否同步执行）。该版本的operator设计逻辑里，没有将作业日志捕获并推送到XCom的机制。

解决方案 #

1. 自定义SparkSubmitOperator子类，捕获日志并推送XCom #

重写operator的execute方法，手动捕获Spark作业的输出日志并推送到XCom：

from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
from airflow.utils.decorators import apply_defaults
import subprocess

class CustomSparkSubmitOperator(SparkSubmitOperator):
    @apply_defaults
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def execute(self, context):
        # 构建SparkSubmit执行命令
        cmd = self._build_spark_submit_command()
        # 执行命令并捕获标准输出、错误输出
        result = subprocess.run(cmd, capture_output=True, text=True)
        # 将完整日志推送到XCom，指定自定义key
        context['ti'].xcom_push(key='spark_job_log', value=result.stdout + result.stderr)
        # 校验命令执行状态
        if result.returncode != 0:
            raise Exception(f"Spark作业启动失败: {result.stderr}")
        return result.returncode

替换原有SparkSubmitOperator为自定义类：

spark_submit_task = CustomSparkSubmitOperator(
        name=f"{job_name}",
        task_id="submit_spark_job",
        conn_id="spark3",
        conf=conf,
        java_class="Application",
        application=f"{jar_path}{jar_file}",
        application_args=application_args,
        execution_timeout=timedelta(minutes=5) 
    )

2. 调整extract_app_id函数，拉取自定义XCom内容 #

修改xcom_pull的参数，指定自定义的key来获取日志：

import re
import logging

logger = logging.getLogger(__name__)

def extract_app_id(**kwargs):
    ti = kwargs['ti']
    # 拉取自定义key对应的XCom日志内容
    log = ti.xcom_pull(task_ids='submit_spark_job', key='spark_job_log')
    if not log:
        logger.error("未获取到Spark作业日志")
        return None
    
    log_str = str(log)
    logger.info("拉取到的Spark日志片段: %s", log_str)
    # 匹配Application ID格式
    app_id_match = re.search(r'application_\d+_\d+', log_str)
    if app_id_match:
        app_id = app_id_match.group()
        logger.info("提取到的Application ID: %s", app_id)
        # 将提取结果推送到XCom供后续任务使用
        ti.xcom_push(key='spark_app_id', value=app_id)
        return app_id
    else:
        logger.error("未在日志中匹配到Application ID")
        return None

3. 替代方案：通过Spark REST API获取Application ID #

如果无法修改operator，可直接调用Spark集群的REST API查询作业信息：

import requests

def extract_app_id(**kwargs):
    ti = kwargs['ti']
    # 替换为你的Spark Master REST API地址
    spark_api_url = "http://your-spark-master:6060/v1/submissions"
    response = requests.get(spark_api_url)
    
    if response.status_code == 200:
        submissions = response.json().get('submissions', [])
        # 按提交时间倒序，匹配当前作业名称
        for submission in reversed(submissions):
            if submission['name'] == job_name:
                app_id = submission['appId']
                logger.info("从API获取到的Application ID: %s", app_id)
                ti.xcom_push(key='spark_app_id', value=app_id)
                return app_id
    logger.error("无法从Spark API获取作业信息")
    return None

注意事项 #

• Airflow 2.x版本中，PythonOperator的provide_context=True可替换为op_kwargs={"ti": "{{ ti }}"}，更贴合新版本规范，但2.0.0版本仍支持provide_context。
• 确保Airflow Worker节点有访问Spark集群日志或REST API的权限。
• 若Spark作业为异步执行（如集群模式），subprocess.run会在作业启动后立即返回，需保证捕获的启动日志中包含Application ID。

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