spark dataframes：读取具有重复列名但数据类型不同的 json

Question

我有下面这样的 json 数据，其中版本字段是差异化因素 -

file_1 = {"version": 1, "stats": {"hits":20}}

file_2 = {"version": 2, "stats": [{"hour":1,"hits":10},{"hour":2,"hits":12}]}

在新格式中，stats 列现在是Arraytype(StructType)。

之前只需要file_1，所以我使用

spark.read.schema(schema_def_v1).json(path)

现在我需要同时读取这两种类型的多个 json 文件。我无法在 schema_def 中将 stats 定义为字符串，因为这会影响 corruptrecord 功能（用于 stats 列），该功能检查所有字段的格式错误的 json 和架构合规性。

1 个只读文件中需要的示例 df 输出 -

version | hour | hits
1       | null | 20
2       | 1    | 10
2       | 2    | 12

我尝试使用mergeSchema 选项读取，但这会使统计字段成为字符串类型。

另外，我尝试通过过滤版本字段并应用spark.read.schema(schema_def_v1).json(df_v1.toJSON) 来制作两个数据帧。这里的 stats 列也变成了String 类型。

我在想，如果在阅读时，我可以根据数据类型将 df 列标题解析为 stats_v1 和 stats_v2 可以解决问题。请提供任何可能的解决方案。

Answer 1

UDF检查字符串或数组，如果是字符串则将字符串转换为数组。

import org.apache.spark.sql.functions.udf
import org.json4s.{DefaultFormats, JObject}
import org.json4s.jackson.JsonMethods.parse
import org.json4s.jackson.Serialization.write
import scala.util.{Failure, Success, Try}

object Parse {
    implicit val formats = DefaultFormats
    def toArray(data:String) = {
      val json_data = (parse(data))
      if(json_data.isInstanceOf[JObject]) write(List(json_data)) else data
    }
}

val toJsonArray = udf(Parse.toArray _)

scala> "ls -ltr /tmp/data".!
total 16
-rw-r--r--  1 srinivas  root  37 Jun 26 17:49 file_1.json
-rw-r--r--  1 srinivas  root  69 Jun 26 17:49 file_2.json
res4: Int = 0

scala> val df = spark.read.json("/tmp/data").select("stats","version")
df: org.apache.spark.sql.DataFrame = [stats: string, version: bigint]

scala> df.printSchema
root
 |-- stats: string (nullable = true)
 |-- version: long (nullable = true)

scala> df.show(false)
+-------+-------------------------------------------+
|version|stats                                      |
+-------+-------------------------------------------+
|1      |{"hits":20}                                |
|2      |[{"hour":1,"hits":10},{"hour":2,"hits":12}]|
+-------+-------------------------------------------+

输出

scala> 

import org.apache.spark.sql.types._
val schema = ArrayType(MapType(StringType,IntegerType))

df
.withColumn("json_stats",explode(from_json(toJsonArray($"stats"),schema)))
.select(
    $"version",
    $"stats",
    $"json_stats".getItem("hour").as("hour"),
    $"json_stats".getItem("hits").as("hits")
).show(false)

+-------+-------------------------------------------+----+----+
|version|stats                                      |hour|hits|
+-------+-------------------------------------------+----+----+
|1      |{"hits":20}                                |null|20  |
|2      |[{"hour":1,"hits":10},{"hour":2,"hits":12}]|1   |10  |
|2      |[{"hour":1,"hits":10},{"hour":2,"hits":12}]|2   |12  |
+-------+-------------------------------------------+----+----+

没有 UDF

scala> val schema = ArrayType(MapType(StringType,IntegerType))

scala> val expr = when(!$"stats".contains("[{"),concat(lit("["),$"stats",lit("]"))).otherwise($"stats")

df
.withColumn("stats",expr)
.withColumn("stats",explode(from_json($"stats",schema)))
.select(
    $"version",
    $"stats",
    $"stats".getItem("hour").as("hour"),
    $"stats".getItem("hits").as("hits")
)
.show(false)

+-------+-----------------------+----+----+
|version|stats                  |hour|hits|
+-------+-----------------------+----+----+
|1      |[hits -> 20]           |null|20  |
|2      |[hour -> 1, hits -> 10]|1   |10  |
|2      |[hour -> 2, hits -> 12]|2   |12  |
+-------+-----------------------+----+----+

Answer 2

先读取第二个文件，爆破stats，使用schema读取第一个文件。

from pyspark.sql import SparkSession
from pyspark.sql.functions import  explode

spark = SparkSession.builder.getOrCreate()
sc = spark.sparkContext

file_1 = {"version": 1, "stats": {"hits": 20}}

file_2 = {"version": 2, "stats": [{"hour": 1, "hits": 10}, {"hour": 2, "hits": 12}]}

df1 = spark.read.json(sc.parallelize([file_2])).withColumn('stats', explode('stats'))
schema = df1.schema

spark.read.schema(schema).json(sc.parallelize([file_1])).printSchema()

output >> root
 |-- stats: struct (nullable = true)
 |    |-- hits: long (nullable = true)
 |    |-- hour: long (nullable = true)
 |-- version: long (nullable = true)

Answer 3

IIUC，您可以使用spark.read.text 读取JSON 文件，然后使用json_tuple、from_json 解析value。注意stats 字段我们使用coalesce 来解析基于两个或多个模式的字段。 （如果每个文件包含跨多行的单个 JSON 文档，则添加 wholetext=True 作为 spark.read.text 的参数）

from pyspark.sql.functions import json_tuple, coalesce, from_json, array

df = spark.read.text("/path/to/all/jsons/")

schema_1 = "array<struct<hour:int,hits:int>>"
schema_2 = "struct<hour:int,hits:int>"

df.select(json_tuple('value', 'version', 'stats').alias('version', 'stats')) \
    .withColumn('status', coalesce(from_json('stats', schema_1), array(from_json('stats', schema_2)))) \
    .selectExpr('version', 'inline_outer(status)') \
    .show()
+-------+----+----+
|version|hour|hits|
+-------+----+----+
|      2|   1|  10|
|      2|   2|  12|
|      1|null|  20|
+-------+----+----+