如何将Spark集群模式日志同步至Airflow SparkSubmitOperator任务日志？

针对集群模式下SparkSubmitOperator无法直接在Airflow任务日志中查看YARN日志的问题，以下是几种实用的实现方式：

一、自定义Operator继承SparkSubmitOperator，自动拉取日志 #

通过继承原Operator，在任务提交完成后自动调用yarn logs命令并将日志输出到Airflow任务日志中：

from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator
import subprocess
import re

class SparkSubmitWithYarnLogs(SparkSubmitOperator):
    def execute(self, context):
        # 执行原生Spark提交逻辑
        submission_result = super().execute(context)
        
        # 提取YARN应用ID（从提交结果或日志输出中获取）
        app_id = None
        if hasattr(submission_result, 'applicationId'):
            app_id = submission_result.applicationId
        else:
            # 若结果中无ID，从Spark提交的标准输出中匹配（适配不同Spark版本）
            spark_output = self._run_spark_submit()  # 重写run_spark_submit捕获输出
            app_id_match = re.search(r'Application ID: (application_\d+_\d+)', spark_output)
            if app_id_match:
                app_id = app_id_match.group(1)
        
        if app_id:
            # 拉取并打印YARN日志
            try:
                logs = subprocess.check_output(
                    ['yarn', 'logs', '-applicationId', app_id],
                    stderr=subprocess.STDOUT,
                    text=True
                )
                print("\n=== 开始输出YARN应用日志 ===")
                print(logs)
                print("=== YARN应用日志输出结束 ===\n")
            except subprocess.CalledProcessError as e:
                print(f"拉取YARN日志失败: {e.output}")
        return submission_result

使用时直接替换原SparkSubmitOperator为自定义的SparkSubmitWithYarnLogs即可。

二、通过回调函数实现日志拉取 #

无需修改Operator，给原SparkSubmitOperator添加成功/失败回调，在任务结束后触发日志拉取：

import subprocess

def pull_yarn_logs(context):
    task_instance = context['task_instance']
    # 从XCom中获取预存的应用ID（需在提交时存入）
    app_id = task_instance.xcom_pull(task_ids=task_instance.task_id, key='yarn_app_id')
    
    if not app_id:
        print("未获取到YARN应用ID，跳过日志拉取")
        return
    
    try:
        logs_output = subprocess.check_output(
            ['yarn', 'logs', '-applicationId', app_id],
            stderr=subprocess.STDOUT,
            text=True
        )
        print("\n=== YARN日志开始 ===")
        print(logs_output)
        print("=== YARN日志结束 ===\n")
    except Exception as e:
        print(f"拉取YARN日志出错: {str(e)}")

# 定义Spark任务时配置回调
spark_cluster_task = SparkSubmitOperator(
    task_id='spark_cluster_job',
    application='/path/to/your/spark_app.jar',
    conn_id='spark_yarn_conn',
    deploy_mode='cluster',
    on_success_callback=pull_yarn_logs,
    on_failure_callback=pull_yarn_logs  # 失败时也拉取日志
)

# 补充：在提交任务时将应用ID存入XCom（可通过修改Operator或捕获输出实现）
# 例如在SparkSubmitOperator的execute方法后添加：
# context['task_instance'].xcom_push(key='yarn_app_id', value=app_id)

注意事项 #

• 权限配置：Airflow的执行用户必须拥有YARN集群的操作权限，能够执行yarn logs命令访问应用日志。
• 日志大小：若YARN日志过大，直接打印可能导致Airflow任务日志膨胀，可添加日志过滤（比如只拉取最近N行，或过滤关键词）。
• 应用ID提取：不同Spark/YARN版本的应用ID输出格式可能略有差异，需根据实际情况调整正则匹配规则。

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