使用 Spark 展平嵌套的 ORC 文件 - 性能问题答案

【问题标题】：Flattening a nested ORC file with Spark - Performance issue使用 Spark 展平嵌套的 ORC 文件 - 性能问题
【发布时间】：2017-10-02 18:40:03
【问题描述】：

我们在读取嵌套 ORC 文件时面临严重的性能问题。

这是我们的 ORC 架构：

|-- uploader: string (nullable = true)
|-- email: string (nullable = true)
|-- data: array (nullable = true)
|    |-- element: struct (containsNull = true) 
|    |    |-- startTime: string (nullable = true)
|    |    |-- endTime: string (nullable = true)
|    |    |-- val1: string (nullable = true)
|    |    |-- val2: string (nullable = true)
|    |    |-- val3: integer (nullable = true)
|    |    |-- val4: integer (nullable = true)
|    |    |-- val5: integer (nullable = true)
|    |    |-- val6: integer (nullable = true)

“数据”数组可能包含 75K 对象。

在我们的 spark 应用程序中，我们将这个 ORC 展平，如下所示：

val dataFrame = spark.read.orc(files: _*)
val withData = dataFrame.withColumn("data", explode(dataFrame.col("data")))
val withUploader = withData.select($"uploader", $"data")
val allData = withUploader
  .withColumn("val_1", $"data.val1")
  .withColumn("val_2", $"data.val2")
  .withColumn("val_3", $"data.val3")
  .withColumn("val_4", $"data.val4")
  .withColumn("val_5", $"data.val5")
  .withColumn("val_6", $"data.val6")
  .withColumn("utc_start_time", timestampUdf($"data.startTime"))
  .withColumn("utc_end_time", timestampUdf($"data.endTime"))

allData.drop("data")

扁平化过程似乎是一个非常繁重的操作：读取一个包含 20 条记录的 2MB ORC 文件，每条记录都包含一个包含 75K 对象的数据数组，需要数小时的处理时间。读取文件并在不展平的情况下收集它，需要 22 秒。

有没有办法让 Spark 更快地处理数据？

【问题讨论】：

标签： scala apache-spark nested orc

【解决方案1】：

我会尽量避免使用大的explodes。数组中有 75K 个元素：

每个 Row 创建 75K Row 对象。这是一项巨大的分配工作。
您复制了 uploaded 和 email 75K 次。短期内它将引用相同的数据，但一旦数据以内部格式序列化和反序列化，它们就会指向不同的对象，从而有效地增加内存需求。

根据您要应用的转换，使用 UDF 处理整个数组可能会更有效。

【讨论】：

【解决方案2】：

如果这对某人有帮助，我发现使用 flatmap 展平数据比使用爆炸要快得多：

dataFrame.as[InputFormat].flatMap(r => r.data.map(v => OutputFormat(v, r.tenant)))

性能的提升非常显着。

处理一个包含 20 条记录的文件，每条记录都包含一个 250K 行的数组 - 展开实现需要 8 小时，而平面图实现需要 7 分钟 (!)

【讨论】：