星星之火:将JSON文件转换为正确的格式

Question

我已经用以下的100+结构将百万条记录存储在文件中(真正的数据有更多的列、行和嵌套)

{"id":"2-2-3","key":"value"}{"id":"2-2-3","key":"value"}{"id":"2-2-3","key":"value"}{"id":"2-2-3","key":"value"}{"id":"2-2-3","key":"value"}

sqlContext.read.json函数无法解析这一点，因为记录不是在多行上，而是在一条大行上。下面的解决方案解决了这个问题，但却是一个很大的性能杀手。在Apache中处理这个问题的最佳方法是什么？

val rdd = sc.wholeTextFiles("s3://some-bucket/**/*")
val validJSON = rdd.flatMap(_._2.replace("}{", "}\n{").split("\n"))

val df = sqlContext.read.json(validJSON)

df.count()
df.select("id").show()

Answer 1

这是Antot's answer的一个片段，它应该处理嵌套的JSON

input.toVector
    .foldLeft((false, Vector.empty[Char], Vector.empty[String])) {
      case ((true, charAccum, strAccum), '{') => (false, Vector('{'), strAccum :+ charAccum.mkString);
      case ((_, charAccum, strAccum), '}')    => (true, charAccum :+ '}', strAccum);
      case ((_, charAccum, strAccum), char)   => (false, charAccum :+ char, strAccum)
    }
    ._3

它所做的基本上是将数据拆分为一个Vector[Char]，并使用foldLeft将输入聚合到子字符串中。诀窍是跟踪有关前一个字符的足够信息，以确定{是否标志着新对象的开始。

我使用这个输入来测试它(基本上是OP的示例输入，其中包含一个嵌套的对象)：

val input = """{"id":"2-2-3","key":{ "test": "value"}}{"id":"2-2-3","key":"value"}{"id":"2-2-3","key":"value"}{"id":"2-2-3","key":"value"}{"id":"2-2-3","key":"value"}"""

得到了这样的结果，看起来很好：

Vector({"id":"2-2-3","key":{ "test": "value"}}, 
       {"id":"2-2-3","key":"value"}, 
       {"id":"2-2-3","key":"value"}, 
       {"id":"2-2-3","key":"value"})

Answer 2

原始方法的问题是调用_._2.replace("}{", "}\n{"，它从输入字符串中创建另一个巨大的字符串，插入新的行字符，然后再将其分割成一个数组。

通过最小化中间字符串的创建并尽快检索目标字符串，可以进行改进。为此，我们可以使用一些子字符串：

val validJson = rdd.flatMap(rawJson => {

  // functions extracted to make it more readable.
  def nextObjectStartIndex(fromIndex: Int):Int = rawJson._2.indexOf('{', fromIndex)
  def currObjectEndIndex(fromIndex: Int): Int = rawJson._2.indexOf('}', fromIndex)
  def extractObject(fromIndex: Int, toIndex: Int): String = rawJson._2.substring(fromIndex, toIndex + 1)

  // the resulting strings are put in a local buffer
  val buffer = new ListBuffer[String]()

  // init the scanning of the input string
  var posStartNextObject = nextObjectStartIndex(0)

  // main loop terminates when there are no more '{' chars
  while (posStartNextObject != -1) {
    val posEndObject = currObjectEndIndex(posStartNextObject)
    val extractedObject = extractObject(posStartNextObject, posEndObject)
    posStartNextObject = nextObjectStartIndex(posEndObject)
    buffer += extractedObject
  }

  buffer
})

请注意，只有当输入JSON中的对象不是嵌套的，假设所有大括号都是相同级别的单独对象时，这种方法才能工作。