Spark DataFrame 与 CreateDataFrame

Question

全部，

我们有很多离子格式的文件，我们必须在分组和过滤之前对数据进行一些计算（它们的总大小约为 1.3 TB，每个文件约为 200 - 300 MB）。 我尝试过这两种不同的方式。

首先是并行化我需要的来自 S3 的文件列表，将其转换为 RDD[Row]。然后从中创建数据框。

val rdd = ss.sparkContext.parallelize(suspendedList, suspendedList.size).flatMap(chunk => {
  Ranger.consumeStreamToRow(chunk, dfSchema.value)
})
val df = ss.createDataFrame(rdd, schema)

第二种方法是并行化文件列表，并使用 AvroParquetWriter 将其写入 S3 中的 parquet 文件。将其加载为 dataFrame

val rdd = ss.sparkContext.parallelize(suspendedList, suspendedList.size).foreach( chunk => {
  Ranger.writeParquetFile(chunk, avroSchema.value, TaskContext.getPartitionId())
})

第二种方式最终比第一种方式快得多，尽管它必须从 S3 进行读写。 我注意到的一件事是，与下一种方法相比，第一种方法的垃圾收集时间似乎很长，我认为这导致实际处理数据所需的时间更长。

有人可以解释第一种和第二种方法发生了什么吗？为什么即使我正在读写 S3，第二种方法最终还是要快得多？ （我将执行程序内存设置为 10GB，驱动程序设置为 40GB，我使用的是 Spark EMR）

Answer 1

根据您的 cmets，我了解第一种方法读取文件并创建 List[Row]，第二种方法 spark 用于读取文件。基于这些（如果我错了，请纠正我）我会解释。

RDD 对象仅包含有关“如何以及在何处读取数据”的信息，而不是实际数据（有点像流或 python 生成器）。因此，要创建 RDD，spark 将列出 s3 目录中的所有文件，并（通常）为每个文件创建一个分区并将所有文件放入 RDD 对象中。到目前为止，实际数据甚至还没有被读取/加载。
在这个 RDD 上附加的transformation（例如过滤器、映射）只会创建更多的 RDD（以之前的 RDD 作为源）和应该执行的计算。还没有计算任何东西。

只有当最终的Action（sink）例如save，collect，count被执行时才会读取实际数据，并且每个分区由单独的worker读取（可能在不同的物理机器上）。
因此，在任何给定时间，一台机器（工作人员）只需要在内存中拥有一小部分数据。

当您读取文件时，文件被逐行读取以首先创建一个List[String]文件中的每一行都成为列表中的一个值，此列表中的map方法用于转换每个String对象到行对象。所有这些都在单个驱动程序（主）进程的内存中，几乎没有并行性。因此非常慢，需要大量内存（和垃圾收集）。

总而言之，了解您的代码的哪些部分在驱动程序进程中运行（每个应用程序只有一个）以及哪些部分在工作程序中运行（spark 应用程序可以在许多不同的机器上具有多个工作程序进程）。并将您的计算转移给工作人员。