处理 Spark 数据框列中的 json 字符串

Question

当我寻找在数据框的字符串列中解析 json 的方法时，我不断遇到更简单地读取 json 文件源的结果。我的源实际上是一个配置单元 ORC 表，其中一个列中有一些字符串，它是 json 格式。我真的很想把它转换成像地图一样解析的东西。

我很难找到一种方法来做到这一点：

import java.util.Date
import org.apache.spark.sql.Row
import scala.util.parsing.json.JSON

val items = sql("select * from db.items limit 10")
//items.printSchema
val internal = items.map {
  case Row(externalid: Long, itemid: Long, locale: String,
           internalitem: String, version: Long,
           createdat: Date, modifiedat: Date)
       => JSON.parseFull(internalitem)
}

我认为这应该可行，但也许有一种更 Spark 的方式来代替，因为我收到以下错误：

java.lang.ClassNotFoundException: scala.Any
 at scala.reflect.internal.util.AbstractFileClassLoader.findClass(AbstractFileClassLoader.scala:62)

具体来说，我的输入数据大概是这样的：

externalid, itemid, locale, internalitem, version, createdat, modifiedat
       123,    321, "en_us", "{'name':'thing','attr':{
                               21:{'attrname':'size','attrval':'big'},
                               42:{'attrname':'color','attrval':'red'}
                              }}",              1, 2017-05-05…, 2017-05-06…

是的，它不完全是 RFC 7158。

attr 键可以是任何 80,000 个值中的 5 到 30 个，所以我想改为这样：

externalid, itemid, locale, internalitem, version, createdat, modifiedat
       123,    321, "en_us", "{"name':"thing","attr":[
                               {"id":21,"attrname':"size","attrval":"big"},
                               {"id":42,"attrname":"color","attrval":"red"}
                              ]}",              1, 2017-05-05…, 2017-05-06…

然后将internalitem 展平为字段并分解attr 数组：

externalid, itemid, locale, name, attrid, attrname attrval, version, createdat, modifiedat
       123,    321, "en_us", "thing", 21,   "size",  "big",       1, 2017-05-05…, 2017-05-06…
       123,    321, "en_us", "thing", 21,  "color",  "red",       1, 2017-05-05…, 2017-05-06…

Answer 1

我从来没有使用过这样的计算，但我有一个建议给你：

在您自己对列进行任何操作之前，只需检查 sql.functions 包，其中包含一大堆有用的函数来处理列，例如日期提取和格式化、字符串连接和拆分， ...它还提供了一些函数来处理 json 对象，例如：from_json 和 json_tuple。

要使用这些方法，您只需导入它们并在 select 方法中调用它们，如下所示：

import spark.implicits._
import org.apache.spark.sql.functions._
val transofrmedDf = df.select($"externalid", $"itemid", … from_json($"internalitem", schema), $"version" …)

首先，您必须为您的 json 列创建一个架构并将其放入 schema 变量中

希望对你有帮助。