Spark加载数据到数据库时Executor失败终止Driver的实现咨询

你当前的代码无法实现需求，核心问题如下：

• 累加器countErrors的更新逻辑写在Driver侧的JDBC写入语句之后，完全无法捕获Executor端执行数据校验、写入时出现的坏记录、执行异常
• 没有新增针对特定列的坏记录校验逻辑
• 异常捕获块为空，既没有上报逻辑也没有作业终止逻辑
• Spark默认的任务重试机制会导致Executor失败后自动重试，无法做到立即终止

具体实现方案 #

1. 核心逻辑调整 #

• 新增数据校验算子：在写入JDBC之前通过mapPartitions遍历每一条数据，校验指定列是否为坏记录，匹配到坏记录就更新累加器，同时主动抛出运行时异常终止分区执行
• 关闭Spark任务重试：配置spark.task.maxFailures=1，避免Executor失败后自动重试，保证异常立即抛到Driver端
• Driver侧双维度校验：既捕获JDBC写入动作抛出的全局异常，也在写入完成后判断累加器数值，只要错误数大于0就触发上报、终止作业
• 补充异常捕获块的上报、作业终止逻辑

2. 修改后的完整代码 #

import org.apache.spark.api.java.function.MapPartitionsFunction;
import org.apache.spark.sql.*;
import org.apache.spark.util.LongAccumulator;
import java.util.Properties;
import java.util.Iterator;

public class LoadSqlData {
    public static void main(String[] args) {
        // 配置Spark关闭任务重试，保证出错立即终止
        SparkSession spark= SparkSession.builder()
                .appName("loadSqlData")
                .master("local[*]")
                .config("spark.task.maxFailures", 1)
                .getOrCreate();

        Properties connectionProperties= new Properties();
        connectionProperties.put("user","postgres");
        connectionProperties.put("password","root");

        // 注意Windows路径反斜杠需要转义
        Dataset<Row> personcsvdata = spark.read().option("header","true").csv("C:\\Users\\Manasa\\Documents\\nulldata.csv");
        personcsvdata.show();

        LongAccumulator countErrors = spark.sparkContext().longAccumulator("errorRecordCount");

        try {
            // 新增特定列坏记录校验逻辑，示例为校验id列不能为null/空值，可替换为实际规则
            Dataset<Row> validatedData = personcsvdata.mapPartitions((MapPartitionsFunction<Row, Row>) iterator -> {
                return new Iterator<Row>() {
                    @Override
                    public boolean hasNext() {
                        return iterator.hasNext();
                    }

                    @Override
                    public Row next() {
                        Row row = iterator.next();
                        String id = row.getAs("id");
                        if (id == null || id.trim().isEmpty()) {
                            countErrors.add(1);
                            // 检测到坏记录直接抛出异常终止当前分区执行，异常会上报到Driver
                            throw new RuntimeException("检测到坏记录，id列为空：" + row);
                        }
                        return row;
                    }
                };
            }, Encoders.row(personcsvdata.schema()));

            // 写入JDBC，该动作触发所有计算逻辑
            validatedData.write().mode(SaveMode.Append).jdbc("jdbc:postgresql://localhost:5432/postgres", "public.employee", connectionProperties);

            // 写入完成后校验累加器数值，有错误就终止
            if (countErrors.value() > 0) {
                throw new RuntimeException("作业检测到" + countErrors.value() + "条坏记录，已终止");
            }

        } catch (Exception e) {
            // 这里替换为实际的上报逻辑，比如发告警、写错误日志
            System.err.println("作业执行失败：" + e.getMessage());
            // 终止Spark作业
            spark.stop();
            System.exit(1);
        }

        // 正常执行结束退出
        spark.stop();
    }
}

3. 注意事项 #

• 累加器的更新必须写在Executor端执行的算子内部，才能捕获到分布式计算过程中的错误
• 如果是JDBC写入过程中出现数据库连接失败、约束冲突等错误，也会直接抛到Driver的catch块中，触发终止逻辑
• 如果需要更细粒度的错误信息上报，可以自定义累加器存储错误详情，而不是只统计错误数量

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