Spark应用中多Spark Session的追踪与UI可观测性问题

针对多Spark Session场景下无法在UI区分作业归属的问题，提供以下几种可行方案：

方案一：Spark 3.x+ 官方作业标签特性（推荐） #

Spark 3.0及以上版本支持为作业添加自定义标签，UI的作业列表会专门展示标签列，无需依赖反射修改作业名称，稳定性更高。

实现步骤：

• 为每个新建的Spark Session分配唯一标识（如session_001、session_002）
• 在执行该Session的SQL/Job前，通过SparkContext设置专属标签，执行完成后清理标签：

// 创建Session
SparkSession session1 = sparkSession.newSession();
// 执行作业前设置标签
session1.sparkContext().setJobTag("session_id", "session_001");
try {
    // 执行Session内的SQL查询或操作
    session1.sql("SELECT * FROM table_a JOIN table_b ON table_a.id = table_b.id").show();
} finally {
    // 清理标签，避免影响其他Session的作业
    session1.sparkContext().clearJobTag("session_id");
}

完成后在Spark UI的「Jobs」页面，即可通过「Tags」列直接查看每个作业所属的Session标识。

方案二：自定义QueryExecutionListener + 作业名称修改 #

通过为每个Session注册专属的查询监听器，在查询执行时动态修改作业名称，添加Session标识。

1. 实现自定义QueryExecutionListener：

public class SessionQueryListener implements QueryExecutionListener {
    private final String sessionId;

    public SessionQueryListener(String sessionId) {
        this.sessionId = sessionId;
    }

    @Override
    public void onSuccess(String query, QueryExecution qe, long duration) {
        // 注册临时SparkListener修改作业名称
        qe.sparkSession().sparkContext().addSparkListener(new SparkListener() {
            @Override
            public void onJobStart(SparkListenerJobStart jobStart) {
                JobExecutionInfo jobInfo = jobStart.jobInfo();
                // 为作业名称添加Session前缀
                String newJobName = String.format("[%s] %s", sessionId, jobInfo.jobName());
                // 通过反射修改作业名称（注意Spark版本兼容性）
                try {
                    Field jobNameField = JobExecutionInfo.class.getDeclaredField("jobName");
                    jobNameField.setAccessible(true);
                    jobNameField.set(jobInfo, newJobName);
                } catch (Exception e) {
                    // 异常捕获处理
                }
            }
        });
    }

    @Override
    public void onFailure(String query, QueryExecution qe, Exception exception) {
        // 可选：查询失败时的处理逻辑
    }
}

2. 为每个Session注册监听器：

SparkSession session2 = sparkSession.newSession();
session2.listenerManager().register(new SessionQueryListener("session_002"));
// 后续该Session执行的所有SQL作业名称都会带上[session_002]前缀

方案三：ThreadLocal存储Session标识 + 全局SparkListener #

利用线程本地变量存储当前执行作业的Session标识，通过全局SparkListener统一修改作业名称。

1. 定义ThreadLocal工具类：

public class SessionContextHolder {
    private static final ThreadLocal<String> SESSION_ID_HOLDER = new ThreadLocal<>();

    public static void setSessionId(String sessionId) {
        SESSION_ID_HOLDER.set(sessionId);
    }

    public static String getSessionId() {
        return SESSION_ID_HOLDER.get();
    }

    public static void clearSessionId() {
        SESSION_ID_HOLDER.remove();
    }
}

2. 实现全局SparkListener：

public class SessionJobListener extends SparkListener {
    @Override
    public void onJobStart(SparkListenerJobStart jobStart) {
        String sessionId = SessionContextHolder.getSessionId();
        if (sessionId != null) {
            JobExecutionInfo jobInfo = jobStart.jobInfo();
            String newJobName = String.format("[%s] %s", sessionId, jobInfo.jobName());
            try {
                Field jobNameField = JobExecutionInfo.class.getDeclaredField("jobName");
                jobNameField.setAccessible(true);
                jobNameField.set(jobInfo, newJobName);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

3. 注册全局监听器并绑定Session标识：

// 给共享的SparkContext注册全局监听器
sparkSession.sparkContext().addSparkListener(new SessionJobListener());

// 执行Session作业时绑定标识
SparkSession session3 = sparkSession.newSession();
SessionContextHolder.setSessionId("session_003");
try {
    session3.sql("SELECT count(*) FROM table_c").show();
} finally {
    SessionContextHolder.clearSessionId();
}

注意事项 #

• 若使用反射修改作业名称，需注意Spark版本兼容性，不同版本的JobExecutionInfo类结构可能存在差异
• 多线程环境下务必确保Session标识的设置与清理在同一线程内执行，避免出现标识串用的情况
• 优先使用Spark 3.x+的作业标签方案，无需依赖反射，更符合官方规范

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