PySpark从Oracle读取数据：并行读取行数与预期不符问题排查

核心原因 #

方法2出现总行数多于方法1的情况，主要有两种可能：

1. 分片边界重复读取：Spark JDBC自动分片时，会将[lowerBound, upperBound]划分为numPartitions个区间，相邻区间的边界值（如id=1000）可能被两个分片同时查询，导致同一条主键记录被重复加载，最终DataFrame总行数增加。
2. 参数取值错误：如果lowerBound和upperBound取的是全表id的极值，而非eff_dt between '01SEP2022' AND '30SEP2022'过滤后的数据的id范围，可能导致分片查询逻辑冲突（极端情况下可能因驱动解析问题读取到额外数据）。

验证与解决步骤 #

1. 确认是否存在重复行 #

先检查方法2生成的DataFrame是否有重复主键：

duplicate_ids = df.groupBy("id").count().filter("count > 1")
print(f"重复id数量: {duplicate_ids.count()}")

若结果大于0，即可确认是边界重复导致的问题。

2. 修正并行读取逻辑 #

方案一：手动指定分片条件（推荐）

放弃自动分片，改用predicates参数手动定义每个分片的查询条件，彻底避免边界重复：

# 根据实际过滤后的id范围，拆分为10个不重叠的条件
predicates = [
    "eff_dt between '01SEP2022' AND '30SEP2022' AND id <= 1000000",
    "eff_dt between '01SEP2022' AND '30SEP2022' AND id > 1000000 AND id <= 2000000",
    # 依次添加剩余8个分片条件，最后一个示例如下
    "eff_dt between '01SEP2022' AND '30SEP2022' AND id > 9000000"
]

df = spark.read.format('jdbc')\
     .option('driver', 'driver_name')\
     .option('dbtable', 'table_name')\
     .option('user', 'user_name')\
     .option('password', 'pwd')\
     .option('predicates', predicates)\
     .load()

方案二：修正自动分片参数

如果坚持用自动分片：

• 先在Oracle中获取过滤后的数据的id范围：

  SELECT MIN(id), MAX(id) FROM table_name WHERE eff_dt BETWEEN '01SEP2022' AND '30SEP2022';
  

• 将上述查询结果作为lowerBound和upperBound的取值，确保分片范围完全匹配过滤后的数据集。

3. 优化日期过滤逻辑 #

避免日期字符串解析歧义，改用Oracle标准日期函数：

query = "(select * from table_name where eff_dt between TO_DATE('2022-09-01','YYYY-MM-DD') AND TO_DATE('2022-09-30','YYYY-MM-DD')) myTable"

4. 备选方案：提升单线程读取效率 #

如果并行读取的必要性不高，可给方法1添加fetchsize参数提升读取速度，同时保证数据准确性：

df = spark.read.format('jdbc')\
     .option('driver', 'driver_name')\
     .option('dbtable', query)\
     .option('user', 'user_name')\
     .option('password', 'pwd')\
     .option('fetchsize', 10000)  # 加大批量读取行数，减少网络交互
     .load()

总结 #

方法2的行数异常大概率是分片边界重复或参数取值错误导致的。通过手动指定分片条件、修正参数取值，即可解决问题并保证并行读取的效率与准确性。

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