Spark JDBC读取Oracle数据随机丢行问题求助

AI 小北

2026-6-16

排查思路

分区数据分布校验：检查分区列的数据分布是否均匀，若存在大量重复值或数据倾斜，会导致部分分区数据量过大，触发JDBC驱动的隐性异常。可通过Spark临时统计分区列的分组行数：
```
spark.read.jdbc(...)
  .groupBy("partition_column")
  .count()
  .orderBy(desc("count"))
  .show(10)
```
隐性错误排查：查看Glue作业的CloudWatch日志，搜索oracle.jdbc、fetch、abort等关键词，确认是否存在未被Spark捕获的JDBC读取异常——部分驱动异常可能不会触发作业失败，但会导致当前分区数据丢失。
事务隔离级别验证：Oracle默认的READ COMMITTED隔离级别下，跨分区读取时可能因源表数据变更导致快照不一致，引发丢行。检查是否显式设置了事务隔离参数，尝试调整为更严格或宽松的级别。
驱动参数有效性验证：确认Spark的fetchsize参数是否被Oracle驱动正确接收，部分驱动版本会忽略该参数，需通过驱动专属参数oracle.jdbc.defaultRowPrefetch覆盖。

解决方案

精准分区配置：

使用主键或唯一索引列作为分区键，通过lowerBound、upperBound、numPartitions手动控制分区范围，确保每个分区数据量均匀（7000万行32分区，单分区约218万行）：

val df = spark.read
  .format("jdbc")
  .option("url", jdbcUrl)
  .option("dbtable", "source_table")
  .option("user", username)
  .option("password", password)
  .option("driver", "oracle.jdbc.OracleDriver")
  .option("numPartitions", "32")
  .option("partitionColumn", "primary_key")
  .option("lowerBound", "1")
  .option("upperBound", "70000000")
  .load()

若无法使用数值型主键，可通过predicates参数手动生成每个分区的查询条件，避免自动分区的逻辑漏洞：

val predicates = Array(
  "create_time >= '2023-01-01' AND create_time < '2023-02-01'",
  "create_time >= '2023-02-01' AND create_time < '2023-03-01'",
  // 其他分区条件...
)
val df = spark.read.jdbc(jdbcUrl, "source_table", predicates, connectionProps)

JDBC参数调优：

降低fetchsize至10000-20000，减少单次读取的数据量，降低驱动端内存压力：
```
.option("fetchsize", "20000")
```

添加驱动专属参数确保fetchsize生效：

.option("oracle.jdbc.enableFetchSizeOnFirstRead", "false")
.option("oracle.jdbc.defaultRowPrefetch", "20000")

开启读取重试与错误终止，避免静默丢行：

.option("spark.sql.jdbc.read.retries", "3")
.option("spark.sql.jdbc.failOnError", "true")

数据校验与兜底：
- 按分区统计行数，写入S3后对比每个分区的预期行数（可在读取时实时统计，或提前通过JDBC预查询分区行数），不匹配则抛出异常：
```
val partitionCounts = df.groupBy("partition_column").count().collectAsMap()
// 遍历对比预期值，不一致则触发失败
```
- 保留源表主键，写入S3后执行主键去重，确保最终数据无遗漏：
```
df.dropDuplicates("primary_key")
  .write
  .parquet("s3://target-path")
```

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