scala – 处理记录大小超过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)

您的文件中可能包含一个包含数组的大行.在这里你得到一个例外,因为你正在尝试构建一个太大的CharBuffer(很可能是一个在超出界限后变为负数的整数). java中的最大数组/字符串大小为2 ^ 31-1(Integer.MAX_VALUE -1)(参见 this thread).你说你有一个3GB的记录,每个字符有1B,有30亿个字符超过2 ^ 31,大约相当于20亿.

你可以做的就是有点hacky但是因为你只有一个带有大阵列的键,所以它可能有效.您的json文件可能如下所示：

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

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

{
  "key" : [
      "v0","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("].*$",""))

请注意,如果您的记录可以包含字符“[”和“]”,则必须更加小心,但这是一个想法.