你可以用 SparkSession 读取csv文件，然后按列对数据进行分区，并对数据进行批处理。例如，将数据写入不需要太多处理的外部数据库。

dataFrame
    .write
    .mode(saveMode)
    .option("batchsize", 100)
    .jdbc(url, "tablename", new java.util.Properties())

如果您的业务逻辑要求您处理数据集/Dataframe的每一行，那么您可以使用 df.map() . 如果你的逻辑可以同时在多个rdd上工作，你可以使用 df.mapPartition() 。每记录开销高的任务在 mapPartition 而不是用 map 转变。
考虑初始化数据库的情况。如果我们使用 map() 或者 foreach() ，我们需要初始化的次数将等于rdd中的元素数。如果我们使用 mapPartitions() ，我们需要初始化的次数将等于分区数

您只需将带有检查结果的列附加到原始Dataframe中，然后使用一组规则UDF来执行实际验证，如下所示：

object Rules {
      val rule1UDF = udf(
        (col1: String, col2: String) => {
         // your validation code goes here
         true // the result of validation
      }
    }
    // ...
    val nonAggregatedChecksDf = df
       .withColumn("rule1_result", Rules.rule1UDF("col1", "col2"))
       .withColumn("rule2_result", Rules.rule2UDF("col1", "col3"))
       .select("id", "rule1_result", "rule2_result", <all the columns relevant for the aggregation checks>)

    val aggregatedChecksDf = nonAggregatedChecksDf
       .agg(<...>)
       .withColumn("rule3_result", Rules.rule3UDF("sum1", "avg2"))
       .withColumn("rule4_result", Rules.rule4UDF("count1", "count3"))
       .select("id", "rule1_result", "rule2_result", "rule3_result", "rule4_result")

第二个选项是使用dataset.collect
我建议不要这样做，而是从原始Dataframe和所有检查结果列中选择一个键字段，并将它们以列格式保存为parquet。

aggregatedChecksDf
    .select("id", "rule1_result", "rule2_result", "rule3_result", "rule4_result")
    .write
    .mode(saveMode)
    .parquet(path)

这会更快，因为写入是由所有执行器并行完成的，并且驱动程序不会成为瓶颈。它还很可能有助于避免oom问题，因为内存使用量分布在所有执行器上。