Spark读取多集群JDBC时如何控制并发连接数与单查询数据量？

要实现每个集群最多15个并发连接，同时保证单查询最多获取10条数据，核心是按集群隔离查询任务，而非全局设置并发数。具体步骤如下：

1. 按集群分组谓词 #

由于已明确每个条目所属集群，先将你已划分好的10条一组的谓词，按所属集群归类。比如构建一个集群ID -> 该集群对应的谓词数组的映射。

2. 针对每个集群独立读取并控制并发 #

对每个集群单独执行JDBC读取任务，设置numPartitions=15，这样每个集群的查询任务最多同时建立15个数据库连接，严格符合单集群的并发限制。每个谓词对应一个分区的查询，保证单查询只拉取10条数据。

3. 合并所有集群的结果 #

将每个集群读取到的DataFrame合并为最终结果。

代码示例 #

import org.apache.spark.sql.DataFrame

// 假设你已经按集群分组好了谓词，key是集群标识，value是该集群下的10条一组的谓词数组
val clusterPredicatesMap: Map[String, Array[String]] = Map(
  "cluster_01" -> Array("id IN (1,2,3,4,5,6,7,8,9,10)", "id IN (11,12,...,20)", ...),
  "cluster_02" -> Array("id IN (21,22,...,30)", "id IN (31,32,...,40)", ...),
  // 其他集群的谓词映射
)

// 初始化空DataFrame用于合并结果
var finalItemsDF: DataFrame = spark.emptyDataFrame

// 遍历每个集群执行查询
clusterPredicatesMap.foreach { case (clusterId, predicates) =>
  // 针对当前集群创建JDBC连接，限制并发为15
  val clusterDF = spark.read
    .option("driver", drivername)
    .option("numPartitions", 15) // 单集群最多15个并发连接
    .jdbc(getJdbcUrlForCluster(clusterId), "items", predicates, connectionProperties)
  
  // 合并到最终DataFrame
  finalItemsDF = finalItemsDF.union(clusterDF)
}

// 可选：调整合并后的分区数，避免过多小分区
finalItemsDF = finalItemsDF.repartition(20)

finalItemsDF.show()

关键细节说明 #

• 为什么全局设置numPartitions无效？：全局的numPartitions是所有集群查询任务共享的总分区数，可能导致单个集群的并发连接数超过15（比如多个集群的谓词总数远大于15时），或者并发数不足（谓词总数少于15时）。按集群隔离后，每个集群的并发数独立受控。
• numPartitions的作用：单个JDBC读取任务中，numPartitions决定了该任务最多同时创建的数据库连接数，每个分区对应一个连接执行一个谓词查询。设置为15后，同一时间该集群最多有15个查询在执行。
• 谓词的有效性：你已将条目划分为每组10个的谓词，这保证了单条查询的数据量不会过大，避免长时查询的问题。

额外注意事项 #

• 如果不同集群对应不同的JDBC地址，要实现getJdbcUrlForCluster方法，根据集群ID返回对应的JDBC URL。
• 确保连接属性connectionProperties中没有设置与numPartitions冲突的连接池参数（比如maxPoolSize），以免干扰并发控制。
• 合并DataFrame时，可根据实际数据量调整repartition的分区数，优化后续计算的性能。

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