在 Spark 中导入 parquet 文件时出现内存问题答案

【问题标题】：Memory issue when importing parquet files in Spark在 Spark 中导入 parquet 文件时出现内存问题
【发布时间】：2016-07-08 18:04:17
【问题描述】：

我正在尝试从 Scala Spark (1.5) 中的 parquet 文件中查询数据，包括 200 万行的查询（以下代码中的“变体”）。

val sqlContext = new org.apache.spark.sql.SQLContext(sc)  
sqlContext.sql("SET spark.sql.parquet.binaryAsString=true")

val parquetFile = sqlContext.read.parquet(<path>)

parquetFile.registerTempTable("tmpTable")
sqlContext.cacheTable("tmpTable")

val patients = sqlContext.sql("SELECT DISTINCT patient FROM tmpTable ...)

val variants = sqlContext.sql("SELECT DISTINCT ... FROM tmpTable ... )

当提取的行数较少时，这运行良好，但在请求大量数据时失败并出现“大小超过 Integer.MAX_VALUE”错误。错误如下：

User class threw exception: org.apache.spark.SparkException:
Job aborted due to stage failure: Task 43 in stage 1.0 failed 4 times,
most recent failure: Lost task 43.3 in stage 1.0 (TID 123, node009):
java.lang.RuntimeException: java.lang.IllegalArgumentException:
Size exceeds Integer.MAX_VALUE at
sun.nio.ch.FileChannelImpl.map(FileChannelImpl.java:828) at
org.apache.spark.storage.DiskStore$$anonfun$getBytes$2.apply(DiskStore.scala:125) at
org.apache.spark.storage.DiskStore$$anonfun$getBytes$2.apply(DiskStore.scala:113) at ...

我可以做些什么来完成这项工作？

这看起来像是一个内存问题，但我尝试过使用多达 100 个执行器，没有区别（无论涉及的执行器数量如何，失败所需的时间都保持不变）。感觉数据没有跨节点分区？

我试图通过天真地替换这条线来强制更高的并行化，但无济于事：

val variants = sqlContext.sql("SELECT DISTINCT ... FROM tmpTable ... ).repartition(sc.defaultParallelism*10)

【问题讨论】：

提前对数据进行重新分区，就像在读入数据时一样。Parquet 是一种列数据格式，而 distinct 是面向行的任务。 Spark 需要采用镶木地板格式并将其转换为行，我相信这就是导致您的问题的原因。基本上一个分区会占用太多内存，尽管您会认为 Spark 可能会溢出到磁盘。

标签： scala apache-spark apache-spark-sql parquet

【解决方案1】：

我不认为这个问题是镶木地板特有的。您正在“达到” Spark 中分区最大大小的限制。

大小超过 Integer.MAX_VALUE 在 sun.nio.ch.FileChannelImpl.map(FileChannelImpl.java:828) at ...

Integer.MAX_VALUE 检测到您的分区大小（我相信）超过 2GB（需要超过 int32 才能对其进行索引）。

Joe Widen 的评论很到位。您需要对数据进行更多重新分区。尝试 1000 或更多。

例如，

val data = sqlContext.read.parquet("data.parquet").rdd.repartition(1000).toDF

【讨论】：

将重新分区移动到read.parquet 行似乎已经解决了这个问题。谢谢！