在 spark 中处理超过 3GB 的记录答案

【问题标题】：Handling record size more than 3GB in spark在 spark 中处理超过 3GB 的记录
【发布时间】：2018-04-21 17:44:08
【问题描述】：

当单个记录大小超过 3GB 时，我遇到了异常 `

java.lang.IllegalArgumentException
App > at java.nio.CharBuffer.allocate(CharBuffer.java:330)
App > at java.nio.charset.CharsetDecoder.decode(CharsetDecoder.java:792)
App > at org.apache.hadoop.io.Text.decode(Text.java:412)
App > at org.apache.hadoop.io.Text.decode(Text.java:389)
App > at org.apache.hadoop.io.Text.toString(Text.java:280)
App > at org.apache.spark.sql.execution.datasources.json.JsonFileFormat$$anonfun$createBaseRdd$1.apply(JsonFileFormat.scala:135)
App > at org.apache.spark.sql.execution.datasources.json.JsonFileFormat$$anonfun$createBaseRdd$1.apply(JsonFileFormat.scala:135)

如何增加单个记录的缓冲区大小？

【问题讨论】：

可能没有帮助，但值得注意的是：即使可以这样做（不确定），也可能不是正确的方法（会很慢而且有风险）。你能避免如此巨大的记录大小吗？它是怎么来的？例如，如果它是 RDD.groupByKey 的结果，您可能希望将其替换为 reduceByKey 或其他一些聚合。
它是一个json文件，将所有记录作为json数组放在一个键下。我正在尝试将其弄平。但我无法对其执行任何操作。甚至不打印 json 数组的模式。
如果您有能力改变 JSON 文件的结构，我们能否在进一步处理之前使用一些实用程序将“单个大数组”拆分为“数组数组”？

标签： scala hadoop apache-spark memory-management spark-dataframe

【解决方案1】：

您的文件中可能有一大行包含该数组。在这里您会遇到一个异常，因为您正在尝试构建一个太大的 CharBuffer（很可能是一个在超出范围后变为负数的整数）。 Java 中的最大数组/字符串大小为 2^31-1 (Integer.MAX_VALUE -1)（请参阅 this thread）。你说你有一个 3GB 的记录，每个字符 1B，产生 30 亿个字符，大于 2^31，大约等于 20 亿。

您可以做的有点笨拙，但是由于您只有一个带有大数组的键，因此它可能会起作用。您的 json 文件可能如下所示：

{
  "key" : ["v0", "v1", "v2"... ]
}

或者像这样，但我认为在你的情况下是前者：

{
  "key" : [
      "v0", 
      "v1", 
      "v2",
      ... 
   ]
}

因此，您可以尝试将 hadoop 使用的行分隔符更改为“，”，如 here。基本上，他们是这样做的：

import org.apache.hadoop.io.LongWritable
import org.apache.hadoop.io.Text
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat

def nlFile(path: String) = {
    val conf = new Configuration
    conf.set("textinputformat.record.delimiter", ",")
    sc.newAPIHadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text], conf)
          .map(_._2.toString)
}

然后您可以读取您的数组，并且只需要自己删除 JSON 括号，如下所示：

nlFile("...")
  .map(_.replaceAll("^.*\\[", "").replaceAll("\\].*$",""))

请注意，如果您的记录可以包含字符“[”和“]”，则您必须更加小心，但这里是我们的想法。

【讨论】：