PySpark处理2亿条Oracle数据耗时过长，求性能优化方案

一、数据源读取优化 #

• 推计算到Oracle端预处理：不要全量拉取2亿条数据到Spark，直接通过JDBC查询筛选异常候选集，利用Oracle的索引（若指定字段和ID已有索引）缩小数据集：

  query = """
  SELECT * FROM original_table 
  WHERE 指定字段 IS NULL 
  OR EXISTS (
      SELECT 1 FROM original_table t2 
      WHERE t2.id = original_table.id 
      GROUP BY t2.id 
      HAVING COUNT(*) > 1
  )
  """
  df = spark.read.jdbc(url=oracle_url, table=f"({query}) t", properties=conn_props)
  

• 并行化JDBC读取：用ID作为分区列，配合numPartitions、fetchsize参数拆分数据，让多个executor并行读取：

  df = spark.read.jdbc(
      url=oracle_url,
      table="original_table",
      partitionColumn="id",
      lowerBound=1,
      upperBound=200000000,
      numPartitions=100,
      fetchsize=10000,
      properties=conn_props
  )
  

二、Spark作业资源与执行调优 #

• 资源配置适配：根据集群规模调整executor-memory（如16G）、executor-cores（如4）、num-executors，开启动态资源分配，提升并行处理能力。
• 减少Shuffle开销：若原函数用groupBy("id")检测重复，改用窗口函数替代，避免全量Shuffle：

  from pyspark.sql.window import Window
  from pyspark.sql.functions import count, col
  
  window_spec = Window.partitionBy("id")
  df = df.withColumn("id_count", count("id").over(window_spec))
  duplicate_df = df.filter(col("id_count") > 1)
  

• 启用高效序列化：设置spark.serializer为org.apache.spark.serializer.KryoSerializer，减少数据序列化/反序列化开销。

三、错误数据写入优化 #

• 批量写入Oracle：配置JDBC写入的batchsize（如10000），避免单条插入，减少数据库交互次数：

  write_props = {"batchsize": "10000", ...}
  error_df.write.jdbc(url=oracle_url, table="Error_table", mode="append", properties=write_props)
  

• 合并异常数据后统一写入：先将NULL异常和重复ID异常的数据合并，再执行一次写入操作，避免多次连接数据库：

  null_df = df.filter(col("指定字段").isNull())
  duplicate_df = df.filter(col("id_count") > 1)
  error_df = null_df.union(duplicate_df)
  error_df.write.jdbc(...)
  

四、DAG与数据倾斜优化 #

• 定位耗时节点：通过Spark UI查看执行计划，定位Shuffle量大、任务耗时久的stage。若存在ID数据倾斜，可对ID加盐后分组统计，再合并结果消除倾斜。
• 优先过滤无效数据：先执行filter筛选出NULL异常数据，再处理重复ID检测，减少后续步骤的数据处理量。

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