从Sparko的DynamoDB JSON字符串中提取嵌套的Json字段？

Question

我正在阅读Spark的dynamodb表，该表在一个字段中包含一个JSON字符串，在其他字段中包含字符串。我能够读取JSON字段，但不能读取嵌套的JSON字段。这不是query Json Column using dataframes的复制。这个问题解释了如何从JSON字符串中提取列，但不是嵌套JSON列。

import com.github.traviscrawford.spark.dynamodb._ 
val users = sqlContext.read.dynamodb("Dynamodb_table") 

users.show（1）

试样数据集合

|col1              | ID | field2|field3| 
------------------------------------------------------------------------------------- 
|{"a":[{"b":"value1","x":23},{"b":value2,"x":52}],"c":"valC"}|A1 | X1 |Y1 | 

我需要从COL1（JSON结构）和ID字段提取几个字段。我能够弄清楚如何解析JSON字段（col1）并从col1获取字段'c'，如here所解释的，但无法提取嵌套字段。

我的代码：

val users = sqlContext.read.dynamodb("Dynamodb_table") 
val data = users.selectExpr("get_json_object(col1, '$.c')","get_json_object(col1, '$.a')","ID") 

data.show(1,false) 
|a            |c |ID| 
--------------------------------------------------------- 
|[{"b":"value1","x":23},{"b":value2","x":52}...]|valC|A1| 

现在，当我尝试运用上面的数据帧在同get_json_object，我得到的所有空值。

val nestedData = data.selectExpr("get_json_object(a, '$.b')","c","ID") 
nestedData.show(false) 

|get_json_object(a, '$.b')| c | ID| 
------------------------------------ 
|null      |valC|A1 |  

我试过爆炸以及col'a'有数组和结构。但是，这并不奏效，因为数据框'data'将col/field'a'作为字符串而不是数组返回。任何想法如何解决这个问题？

更新：我也尝试使用JSON4s和net.liftweb.json.parse解析。这也没有帮助

case class aInfo(b: String) 
case class col1(a: Option[aInfo]), c: String) 

import net.liftweb.json.parse 
val parseJson = udf((data: String) => { 
implicit val formats = net.liftweb.json.DefaultFormats 
parse(data).extract[Data] 
}) 

val parsed = users.withColumn("parsedJSON", parseJson($"data")) 
parsed.show(1) 

当我使用这些解析器时，所有值都为空。

我预期的结果：我想从数据集

|b  |x |c | ID| 
-------------------- 
|value1|23|valC|A1 | 
|value2|52|valC|A1 | 

Answer 1

我相信所有所需的拼图碎片已经在这里获得了扁平结构让我们遵循一步一步来。您的数据等同于：

val df = Seq((
    """{"a":[{"b":"value1"},{"b": "value2"}],"c":"valC"}""", "A1", "X1", "Y1" 
)).toDF("col1", "ID", "field2", "field3") 

星火提供了实现相同的查询API作为提升json4s：

import org.json4s._ 
import org.json4s.jackson.JsonMethods._ 

，我们可以使用例如LINQ风格的API来定义UDF：

val getBs = udf((s: String) => for { 
    JString(b) <- parse(s) \ "a" \ "b" 
} yield b) 

如果你想提取多个字段，你当然可以扩展它。例如，如果JSON字符串有多个领域

{"a":[{"b":"value1","d":1},{"b":"value2","d":2}],"c":"valC"} 

，您可以：

for { 
    JObject(a) <- parse(s) \ "a" 
    JField("b", JString(b)) <- a 
    JField("d", JInt(d)) <- a 
} yield (b, d) 

这是假定这两个领域都存在，否则也不会有比赛。为了处理丢失的字段，你可能更喜欢XPath-like表达式或提取：

case class A(b: Option[String], d: Option[Int]) 

(parse(s) \ "a").extract(Seq[A]) 

UDF这样可以与explode使用提取领域：

val withBs = df.withColumn("b", explode(getBs($"col1"))) 

与结果：

+--------------------+---+------+------+------+ 
|    col1| ID|field2|field3|  b| 
+--------------------+---+------+------+------+ 
|{"a":[{"b":"value...| A1| X1| Y1|value1| 
|{"a":[{"b":"value...| A1| X1| Y1|value2| 
+--------------------+---+------+------+------+ 

你尝试使用电梯是不正确的，因为您预计a是aInfo的序列，但将其仅定义为Option[aInfo]。它应该是Option[Seq[aInfo]]：

case class col1(a: Option[Seq[aInfo]], c: String) 

随着类中定义这样的分析应该没有问题的工作。

如果使用当前生成（火花2.1.0）存在由SPARK-17699其中引入了from_json方法需要模式：

import org.apache.spark.sql.types._ 

val bSchema = StructType(Seq(StructField("b", StringType, true))) 
val aSchema = StructField("a", ArrayType(bSchema), true) 
val cSchema = StructField("c", StringType, true) 

val schema = StructType(Seq(aSchema, cSchema)) 

，并且可以被应用为：

import org.apache.spark.sql.functions.from_json 

val parsed = df.withColumn("col1", from_json($"col1", schema)) 

后您可以使用常用符号选择字段：

parsed.select($"col1.a.b")