使用Spark从json字符串列表中提取数组

Question

我的数据框中有一列包含JSON列表，但类型是字符串。我需要在此列上运行分解，因此首先需要将其转换为列表。我找不到太多关于这个用例的参考。

示例数据：

列名称："[{"name"："a"，"info"：{"age"："1"，"等级"："b"}，"其他"： 7}，{"随机"："x"}，{...}]"

上面是数据的样子，字段是不固定的(索引0可能有JSON和一些字段，而索引1将有一些字段和一些其他字段)。列表中可以有更多嵌套的JSONs或一些额外的字段。我现在正在用这个-

＜code＞“分解（拆分（regexp_replace（regexp_replace（colName），（\\\}，），“}，”），“（\\[|\\\]），”，“，”，”，））为colName“”，＜/code＞我只是用“}”替换“}”，然后删除“[]”，然后在“}”上调用拆分，但这种方法不起作用，因为存在嵌套的JSON。

如何从字符串中提取数组？

Answer 1

您可以尝试以下方式：

// Initial DataFrame

df.show(false)

+----------------------------------------------------------------------+
|columnName                                                            |
+----------------------------------------------------------------------+
|[{"name":"a","info":{"age":"1","grade":"b"},"other":7},{"random":"x"}]|
+----------------------------------------------------------------------+

df.printSchema()

root
 |-- columnName: string (nullable = true)
 
// toArray is a user defined function that parses an array of json objects which is present as a string
     
import org.json.JSONArray

val toArray = udf { (data: String) => {
    val jsonArray = new JSONArray(data)
    var arr: Array[String] = Array()
    val objects = (0 until jsonArray.length).map(x => jsonArray.getJSONObject(x))
    objects.foreach { elem =>
        arr :+= elem.toString
    }
    arr
}
}

// Using the udf and exploding the resultant array

val df1 = df.withColumn("columnName",explode(toArray(col("columnName"))))

df1.show(false)

+-----------------------------------------------------+
|columnName                                           |
+-----------------------------------------------------+
|{"other":7,"name":"a","info":{"grade":"b","age":"1"}}|
|{"random":"x"}                                       |
+-----------------------------------------------------+

df1.printSchema()

root
 |-- columnName: string (nullable = true)
 
// Parsing the json string by obtaining the schema dynamically

val schema = spark.read.json(df1.select("columnName").rdd.map(x => x(0).toString)).schema
val df2 = df1.withColumn("columnName",from_json(col("columnName"),schema))

df2.show(false)

+---------------+
|columnName     |
+---------------+
|[[1, b], a, 7,]|
|[,,, x]        |
+---------------+

df2.printSchema()

root
 |-- columnName: struct (nullable = true)
 |    |-- info: struct (nullable = true)
 |    |    |-- age: string (nullable = true)
 |    |    |-- grade: string (nullable = true)
 |    |-- name: string (nullable = true)
 |    |-- other: long (nullable = true)
 |    |-- random: string (nullable = true)
 
// Extracting all the fields from the json

df2.select(col("columnName.*")).show(false)

+------+----+-----+------+
|info  |name|other|random|
+------+----+-----+------+
|[1, b]|a   |7    |null  |
|null  |null|null |x     |
+------+----+-----+------+

编辑：

如果您可以使用get_json_object功能，则可以尝试这种方式

// Get the list of columns dynamically

val columns = spark.read.json(df1.select("columnName").rdd.map(x => x(0).toString)).columns

// define an empty array of Column type and get_json_object function to extract the columns

var extract_columns: Array[Column] = Array()
    columns.foreach { column =>
    extract_columns :+= get_json_object(col("columnName"), "$." + column).as(column)
}

df1.select(extract_columns: _*).show(false)

+-----------------------+----+-----+------+
|info                   |name|other|random|
+-----------------------+----+-----+------+
|{"grade":"b","age":"1"}|a   |7    |null  |
|null                   |null|null |x     |
+-----------------------+----+-----+------+

请注意，< code>info列不是struct类型。您可能必须遵循类似的方法来提取嵌套json的列