如何在Spark中限流JDBC数据读取，避免MySQL CPU过载？

以下是针对并行读取导致数据库过载问题的实用调整方案：

1. 给单分区添加读取延迟 #

在读取后的数据集上，对每个分区添加短暂休眠，限制单分区的读取速度，从而降低整体并发压力：

spark.read()
     .options(options)
     .jdbc("URL", "table", dbConfig.getPartitioningColumn(), lowebound, upperBound, numbOfPartitions, new Properties())
     .mapPartitions(iter -> {
         // 每个分区读取前休眠500毫秒，可根据负载调整
         Thread.sleep(500);
         return iter;
     }, Encoders.bean(YourDataClass.class));

注意：休眠时间需要根据数据库实际负载微调，避免过度拖慢作业。

2. 限制JDBC连接数与单次拉取量 #

通过JDBC连接属性控制并发连接数，同时开启游标读取减少单次请求的数据量：

Properties props = new Properties();
// 限制连接池最大连接数
props.setProperty("maxPoolSize", "5");
// 开启游标式读取，避免一次性加载大量数据
props.setProperty("useCursorFetch", "true");
// 控制每次从数据库拉取的行数
props.setProperty("defaultFetchSize", "1000");

// 使用配置后的属性读取数据
spark.read()
     .options(options)
     .jdbc("URL", "table", dbConfig.getPartitioningColumn(), lowebound, upperBound, numbOfPartitions, props);

这种方式既能减少并发连接，又能降低单请求对数据库的资源消耗。

3. 分批分段读取数据 #

将数据按分区列拆分为多个小批次，分阶段读取，批次之间添加间隔，彻底避免并行读取的压力：

// 示例：按日期列拆分批次
List<String> dateSegments = Arrays.asList("2024-01-01", "2024-01-02", "2024-01-03");
for (String date : dateSegments) {
    // 构造单批次查询语句
    String batchQuery = "(SELECT * FROM table WHERE date_column = '" + date + "') AS batch_data";
    Dataset<Row> batchDF = spark.read()
                               .options(options)
                               .jdbc("URL", batchQuery, new Properties());
    // 处理当前批次数据
    batchDF.write().mode(SaveMode.Append).parquet("hdfs://path/to/save");
    // 批次间休眠1秒，给数据库喘息时间
    Thread.sleep(1000);
}

适合数据量极大、对读取时效性要求不高的场景。

4. 限制Spark全局并行度 #

通过Spark配置限制全局并行任务数，间接降低对数据库的并发请求：

// 设置全局默认并行度
spark.conf().set("spark.default.parallelism", "8");
// 设置SQL shuffle分区数（如果涉及shuffle操作）
spark.conf().set("spark.sql.shuffle.partitions", "8");

这会限制所有阶段的并行任务数量，即使分区数设置较大，实际并行执行的任务数也会被限制。

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