数据帧Spark scala爆炸json数组

Question

假设我有一个如下所示的数据框:

+--------------------+--------------------+--------------------------------------------------------------+
|                id  |           Name     |                                                       Payment|
+--------------------+--------------------+--------------------------------------------------------------+
|                1   |           James    |[ {"@id": 1, "currency":"GBP"},{"@id": 2, "currency": "USD"} ]|
+--------------------+--------------------+--------------------------------------------------------------+

架构是:

根

|-- id: integer (nullable = true)
|-- Name: string (nullable = true)   
|-- Payment: string (nullable = true)

我怎样才能将上面的JSON数组分解为:

+--------------------+--------------------+-------------------------------+
|                id  |           Name     |                        Payment|
+--------------------+--------------------+-------------------------------+
|                1   |           James    |   {"@id":1, "currency":"GBP"} |
+--------------------+--------------------+-------------------------------+
|                1   |           James    |   {"@id":2, "currency":"USD"} |
+--------------------+--------------------+-------------------------------+

我一直在尝试使用如下所示的爆炸功能,但它不起作用.它给出了一个关于无法分解字符串类型的错误,并且它需要一个映射或数组.这是有道理的,因为架构表示它是一个字符串,而不是数组/映射,但我不知道如何将其转换为适当的格式.

val newDF = dataframe.withColumn("nestedPayment", explode(dataframe.col("Payment")))

任何帮助是极大的赞赏!

Answer 1

您必须将JSON字符串解析为JSON 数组,然后explode对结果使用(explode需要一个数组).

为此(假设Spark 2.0.*):

• 如果您知道所有Payment值都包含表示具有相同大小的数组的json(在本例中为2),则可以对第一个和第二个元素进行硬编码,将它们包装在一个数组中并进行爆炸:

  val newDF = dataframe.withColumn("Payment", explode(array(
    get_json_object($"Payment", "$[0]"),
    get_json_object($"Payment", "$[1]")
  )))
  

• 如果你不能保证所有记录都有一个带有2个元素数组的JSON,但你可以保证这些数组的最大长度,你可以使用这个技巧来解析最大大小的元素,然后过滤掉生成null的结果:

  val maxJsonParts = 3 // whatever that number is...
  val jsonElements = (0 until maxJsonParts)
                       .map(i => get_json_object($"Payment", s"$$[$i]"))
  
  val newDF = dataframe
    .withColumn("Payment", explode(array(jsonElements: _*)))
    .where(!isnull($"Payment"))