如何让Spark各Worker节点单独加载数据避免主节点载入海量数据

问题核心原因 #

默认Spark JDBC为单分区读取，所有数据会先拉取到Driver（你所说的Master节点）再执行重分区，大数据量下必然触发Driver内存溢出，读取完成后再执行重分区的逻辑完全无法解决前置的单节点读取瓶颈。

实现方案：JDBC并行分区读取 #

Spark JDBC原生支持多worker并行拉取数据，只需要配置四个分区参数即可实现每个worker独立查询对应范围的数据，无需先集中到Driver节点。

必须配置的核心参数 #

• partitionColumn：用于切分数据的分区列，必须为整数、长整数等数值类型，建议选择有索引的列（比如你用到的ID_Col如果符合要求可直接使用）
• lowerBound：分区列的最小值
• upperBound：分区列的最大值
• numPartitions：读取的分区总数，对应并行拉取的任务数
  Spark会自动按照上述参数将查询切分为numPartitions个范围查询，每个worker节点执行对应范围的SQL，拉取专属数据段生成DataFrame分区。

Scala实现代码 #

val spark = SparkSession
  .builder
  // 生产环境移除master("local[4]")配置，提交到集群时由spark-submit指定
  .config("spark.executor.memory", "8g")
  .config("spark.executor.cores", 4)
  .config("spark.task.cpus",1)
  .appName("Oracle Parallel Read")
  .getOrCreate()

// 第一步：先查询分区列的上下界，避免硬编码
val boundary = spark.read
  .format("jdbc")
  .option("url", "jdbc:oracle:thin:@x.x.x.x:1521:orcldb")
  .option("dbtable", "(select min(ID_Col) as min_id, max(ID_Col) as max_id from table) t")
  .option("user", "orcl")
  .option("password", "********")
  .load()
  .first()
val lowerBound = boundary.getAs[Long]("min_id")
val upperBound = boundary.getAs[Long]("max_id")
val numPartitions = 8 // 可根据集群资源、Oracle负载调整

// 第二步：并行分区读取，数据直接分散到各个worker
val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:oracle:thin:@x.x.x.x:1521:orcldb")
  .option("dbtable", "table")
  .option("user", "orcl")
  .option("password", "********")
  .option("partitionColumn", "ID_Col")
  .option("lowerBound", lowerBound)
  .option("upperBound", upperBound)
  .option("numPartitions", numPartitions)
  .load()

// 无需额外repartition，读取完成后DataFrame已经是分布式存储在各个worker的多分区结构
jdbcDF.explain()

Python实现代码 #

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .config("spark.executor.memory", "8g") \
    .config("spark.executor.cores", 4) \
    .config("spark.task.cpus", 1) \
    .appName("Oracle Parallel Read") \
    .getOrCreate()

# 第一步：查询分区列上下界
boundary = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:oracle:thin:@x.x.x.x:1521:orcldb") \
    .option("dbtable", "(select min(ID_Col) as min_id, max(ID_Col) as max_id from table) t") \
    .option("user", "orcl") \
    .option("password", "********") \
    .load() \
    .first()
lower_bound = boundary["min_id"]
upper_bound = boundary["max_id"]
num_partitions = 8

# 第二步：并行分区读取
jdbcDF = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:oracle:thin:@x.x.x.x:1521:orcldb") \
    .option("dbtable", "table") \
    .option("user", "orcl") \
    .option("password", "********") \
    .option("partitionColumn", "ID_Col") \
    .option("lowerBound", lower_bound) \
    .option("upperBound", upper_bound) \
    .option("numPartitions", num_partitions) \
    .load()

jdbcDF.explain()

可选方案：非数值类型分区列适配 #

如果分区列不是数值类型，可以使用predicates参数自定义每个分区的查询条件，示例如下（Scala为例）：

val predicates = Array(
  "CREATE_TIME between date'2023-01-01' and date'2023-03-31'",
  "CREATE_TIME between date'2023-04-01' and date'2023-06-30'",
  "CREATE_TIME between date'2023-07-01' and date'2023-09-30'",
  "CREATE_TIME between date'2023-10-01' and date'2023-12-31'"
)
val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:oracle:thin:@x.x.x.x:1521:orcldb")
  .option("dbtable", "table")
  .option("user", "orcl")
  .option("password", "********")
  .option("predicates", predicates)
  .load()

注意事项 #

• 生产环境提交任务时移除代码中的master("local[4]")配置，由spark-submit参数指定集群运行模式
• numPartitions不要设置过大，避免并发查询过多压垮Oracle数据库，建议和executor总核心数保持一致
• 分区列建议选择有索引的字段，避免每个分区查询全表导致性能下降

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