Scala+Spark集群下JDBC写入PostgreSQL的分区配置方法

针对你的Scala 2.13.9 + Spark 3.3.0集群环境（多Worker、每个Executor分配1核），要充分利用所有核心并行写入PostgreSQL，核心是让Spark任务分区数匹配集群总可用核心数，同时通过JDBC配置实现分布式写入，具体步骤如下：

1. 对齐集群总核心数与数据分区数 #

先计算集群总可用逻辑核心数：比如3个Worker节点，每个节点运行5个1核Executor，总核心数即为15。后续所有分区配置都要向这个数值对齐，确保每个核心对应一个写入任务。

2. 读取源文件时的预分区处理 #

如果源文件无天然分区（比如非分区存储的CSV/Parquet），读取后通过repartition将DataFrame分区数调整为总核心数，避免后续写入时额外shuffle：

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder().appName("PGWrite").getOrCreate()
val sourceDF = spark.read.format("parquet").load("hdfs://path/to/source/data")

// 总核心数设为15，调整分区数
val partitionedDF = sourceDF.repartition(15)

如果源文件是按目录分区的（比如Hive分区表），直接利用天然分区即可，无需额外repartition。

3. JDBC写入的关键分区配置 #

使用Spark JDBC的partitionColumn、lowerBound、upperBound、numPartitions参数实现分布式写入：

• partitionColumn：选分布均匀的数值/日期型列（比如自增ID、时间戳）作为分区键，避免数据倾斜
• lowerBound/upperBound：分区键的最小/最大值，用来计算每个分区的数据范围
• numPartitions：设置为集群总核心数，确保每个核心处理一个写入任务

示例代码：

val pgUrl = "jdbc:postgresql://your-pg-host:5432/db_name"
val pgProps = new java.util.Properties()
pgProps.setProperty("user", "username")
pgProps.setProperty("password", "password")
pgProps.setProperty("driver", "org.postgresql.Driver")

partitionedDF.write
  .format("jdbc")
  .option("url", pgUrl)
  .option("dbtable", "target_table")
  .option("partitionColumn", "id") // 假设用自增ID作为分区键
  .option("lowerBound", "1")
  .option("upperBound", "1000000") // 假设ID最大值是100万
  .option("numPartitions", "15") // 匹配总核心数
  .mode("append")
  .properties(pgProps)
  .save()

4. 无合适分区键的替代方案 #

如果源数据没有分布均匀的列可用，可生成临时递增ID作为分区键：

import org.apache.spark.sql.functions.monotonically_increasing_id

// 添加临时分区列
val dfWithTempId = sourceDF.withColumn("temp_part_id", monotonically_increasing_id())
val totalCount = dfWithTempId.count()

dfWithTempId.write
  .format("jdbc")
  .option("url", pgUrl)
  .option("dbtable", "target_table")
  .option("partitionColumn", "temp_part_id")
  .option("lowerBound", "0")
  .option("upperBound", totalCount.toString)
  .option("numPartitions", "15")
  .mode("append")
  .properties(pgProps)
  .save()

5. 集群提交参数优化 #

提交Spark任务时，确保Executor配置匹配集群环境：

spark-submit \
  --class com.your.package.MainClass \
  --master yarn \ # 假设使用YARN集群
  --num-executors 15 \ # 总核心数=num-executors*executor-cores=15*1
  --executor-cores 1 \
  --executor-memory 8G \ # 根据单分区数据量调整，避免OOM
  --jars postgresql-42.5.4.jar \ # 适配Spark 3.3.0的PG驱动版本
  your-project.jar

注意事项 #

• 避免数据倾斜：分区键取值要尽可能均匀，不要选性别、状态这类重复值多的列
• 调整shuffle分区数：将spark.sql.shuffle.partitions设置为总核心数（默认200），减少不必要的shuffle开销，可在代码中配置：

  spark.conf.set("spark.sql.shuffle.partitions", "15")
  

• 驱动版本匹配：使用与Spark 3.3.0兼容的PostgreSQL驱动（推荐42.2.x至42.5.x版本）

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