你的问题核心在于误用了collect()把全量数据拉到Driver节点，这对于2000万条记录来说完全是灾难——不仅会耗尽Driver内存，还把分布式计算变成了单线程本地操作，速度慢到无法接受。下面给你一套Scala Spark 2.4下的高效替代方案，完全利用Spark的分布式计算能力：

问题根源分析 #

原代码的致命问题：

• df1.collect()：把2000万条记录全部加载到Driver内存，极易触发OOM，且完全放弃Spark的分布式优势
• 本地ListBuffer操作：单线程处理2000万+5-8亿条数据，性能瓶颈无法突破
• 最后parallelize再分发：相当于把本地数据重新推回集群，做了完全无用的往返操作

高效解决方案：分布式转换+拆分输出 #

我们可以直接在Spark的分布式RDD/DataFrame上完成计算，把任务分散到所有Executor节点并行执行，同时拆分生成两个目标DataFrame。

方案1：用RDD API（灵活适配复杂计算逻辑） #

如果你的属性计算逻辑比较复杂（比如有大量分支、自定义计算），用RDD的map+flatMap是最灵活的选择：

步骤1：定义数据结构（可选，比手动创建Row更安全）

先定义对应df2和df3的case class（如果必须用手动Schema，后面会给出替代方式）：

// df2的4列结构，根据实际字段调整
case class EMIOutput2(
  AS_OF_DATE: String,
  EST_END_DT: String,
  EFF_DT: String,
  NET_PRCPL_AMT: BigDecimal
)

// df3的20列结构，根据实际字段补充
case class EMIOutput3(
  AS_OF_DATE: String,
  EST_END_DT: String,
  EFF_DT: String,
  NET_PRCPL_AMT: BigDecimal,
  attr5: String,
  attr6: Int,
  // ... 剩余16列依次定义
)

步骤2：分布式转换并拆分

// 对df1的RDD进行分布式处理，每条记录生成(df2记录, df3记录列表)的元组
val combinedRDD = df1.rdd.map { row =>
  // --- 计算df2的属性（对应原代码中rwList的逻辑）---
  val asOfDate = row.getAs[String]("AS_OF_DATE")
  val estEndDt = row.getAs[String]("EST_END_DT")
  val effDt = row.getAs[String]("EFF_DT")
  val netPrcplAmt = row.getAs[BigDecimal]("NET_PRCPL_AMT")
  // 其他df2属性计算...
  val emi2Record = EMIOutput2(asOfDate, estEndDt, effDt, netPrcplAmt)

  // --- 计算df3的n条记录（对应原代码中for循环的逻辑）---
  // 这里的n可以是固定值，也可以从row的某列获取（比如row.getAs[Int]("LOOP_COUNT")）
  val emi3Records = (1 to n).map { i =>
    // 根据row和i计算df3的属性
    val effDt3 = s"$effDt-$i" // 示例：基于原EFF_DT加索引
    val netPrcplAmt3 = netPrcplAmt * i // 示例：金额乘以循环次数
    // 其他df3属性计算...
    EMIOutput3(asOfDate, estEndDt, effDt3, netPrcplAmt3, "xxx", i, ...)
  }.toList

  // 返回元组：(df2单条记录, df3多条记录列表)
  (emi2Record, emi3Records)
}

// 拆分得到两个独立的RDD
val df2RDD = combinedRDD.map(_._1)
val df3RDD = combinedRDD.flatMap(_._2) // 扁平化df3的列表为单条记录

// 转换为DataFrame（Spark自动推断case class的Schema）
val df2 = spark.createDataFrame(df2RDD)
val df3 = spark.createDataFrame(df3RDD)

如果你必须使用手动定义的Schema：

把case class换成手动创建Row即可：

val combinedRDD = df1.rdd.map { row =>
  // 生成df2的Row
  val emi2Row = Row(
    row.getAs[String]("AS_OF_DATE"),
    row.getAs[String]("EST_END_DT"),
    row.getAs[String]("EFF_DT"),
    row.getAs[BigDecimal]("NET_PRCPL_AMT")
    // 其他df2列...
  )

  // 生成df3的Row列表
  val emi3RowList = (1 to n).map { i =>
    Row(
      row.getAs[String]("AS_OF_DATE"),
      row.getAs[String]("EST_END_DT"),
      s"${row.getAs[String]("EFF_DT")}-$i",
      row.getAs[BigDecimal]("NET_PRCPL_AMT") * i,
      // 其他df3列...
    )
  }.toList

  (emi2Row, emi3RowList)
}

// 用指定Schema创建DataFrame
val df2 = spark.createDataFrame(df2RDD, emiDFSchema)
val df3 = spark.createDataFrame(df3RDD, emiDFSchema1)

方案2：用DataFrame API（更简洁，自带Catalyst优化） #

如果你的计算逻辑可以用Spark内置函数实现，推荐用DataFrame API，它会自动触发Catalyst优化器，性能可能更好：

// 生成df2：直接选择/计算需要的4列
val df2 = df1.select(
  col("AS_OF_DATE"),
  col("EST_END_DT"),
  col("EFF_DT"),
  col("NET_PRCPL_AMT")
  // 其他需要计算的列用expr或UDF实现，比如：
  // expr("NET_PRCPL_AMT * 0.8").alias("DISCOUNTED_AMT")
)

// 生成df3：先通过explode生成n行，再计算各列
// 假设n是固定值，比如10；如果n是每行不同，用col("N_COL")代替lit(10)
val df3 = df1
  // 生成包含1到n的数组，再explode成n行
  .withColumn("loop_idx", explode(array((1 to n).map(lit(_)): _*)))
  // 计算df3的各列
  .select(
    col("AS_OF_DATE"),
    col("EST_END_DT"),
    concat(col("EFF_DT"), lit("-"), col("loop_idx")).alias("EFF_DT"),
    (col("NET_PRCPL_AMT") * col("loop_idx")).alias("NET_PRCPL_AMT"),
    // 其他列的计算逻辑，比如基于loop_idx调整
    // ...
  )

额外优化建议 #

1. 调整分区数：确保df1的分区数是Executor核心数的2-3倍（比如100核心对应200-300分区），避免数据倾斜或任务并行度不足
2. 资源配置：增大Executor内存（比如--executor-memory 16G）和核心数（--executor-cores 4），让集群能承载大计算量
3. 缓存复用：如果df1需要多次读取，先执行df1.cache()缓存到内存
4. 避免UDF滥用：能用内置函数实现的计算就不用自定义UDF，内置函数的性能更好

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