Pyspark:解析一列json字符串

如果你的JSON字符串是JSON数组而不是对象（我没有代表，所以无法评论），这个答案是为了增加背景。如果你使用 马丁-塔普的坚实答案 它将为你的列返回空值。

tl;dr

如果你的JSON字符串是像这样的数组对象。

[{"a":1, "b":1.0}]
替换代码1】将返回一个数据框架，该框架包含了elements而不是包括数组本身。from_json对这一点并不满意，所以为了达到它想要的具体效果，你可以将由spark.read.json推断出的模式包裹在ArrayType中，它将正确解析（而不是为所有东西返回空值）。
from pyspark.sql import functions as F
from pyspark.sql.types import ArrayType
array_item_schema = \
  spark.read.json(df.rdd.map(lambda row: row['json_string_column'])).schema
json_array_schema = ArrayType(array_item_schema, True)
arrays_df = df.select(F.from_json('json_string_column', json_array_schema).alias('json_arrays'))
objects_df = arrays_df.select(F.explode('json_arrays').alias('objects'))
Intro
作为对Nolan Conaway的补充，似乎当你的JSON的形式为
    "a": 1.0,
    "b": 1
    "a": 0.0,
    "b": 2
在顶层对象是一个数组（而不是一个对象）的情况下，pyspark的spark.read.json()将数组作为一个对象的集合来处理，将其转换为行，而不是单一的行。
参见在PySpark 3.3.0 shell中运行的例子。
>>> myjson        = """[{"a": 1.0,"b": 1},{"a": 2.0,"b": 2}]"""
>>> myotherjson   = """[{"a": 3.0,"b": 3}]"""
>>> rawobjectjson = """{"a": 4.0,"b": 4}"""
>>> spark_read_df = spark.read.json(sc.parallelize([myjson,myotherjson,rawobjectjson]))
>>> spark_read_df.show()
+---+---+
|  a|  b|
+---+---+
|1.0|  1|
|2.0|  2|
|3.0|  3|
|4.0|  4|
+---+---+
>>> spark_read_df.printSchema()
 |-- a: double (nullable = true)
 |-- b: long (nullable = true)
我们可以看到，myjson和myotherjson是JSON对象的JSON数组，它们被扩展为每个对象都有一行。当其中一个JSON字符串rawobjectjson只是一个原始对象时，它也能顺利处理。我认为文档在这里有一点不足，因为我找不到关于数组对象的处理方法。
现在让我们创建一个带有JSON字符串列的数据框架。我们将放弃rawobjectjson，因为我们将看到from_json要求每个字符串都有相同的模式（而这个includes如果有的话，就是顶层阵列）。
>>> from pyspark.sql.types import StructType, StructField, StringType, ArrayType
>>> json_string_data = [
...     (myjson,),
...     (myotherjson,),
... ]
>>> json_df_schema = StructType([
...     StructField('json_strings', StringType(), True),
... ])
>>> raw_json_df = spark.createDataFrame(data=json_string_data, schema=json_df_schema)
>>> raw_json_df.show()
+--------------------+
|        json_strings|
+--------------------+
|[{"a": 1.0,"b": 1...|
| [{"a": 3.0,"b": 3}]|
+--------------------+
现在我试图用spark.read.json推断出的模式传递给from_json，把JSON列读成对象。但它一直返回完全null的列。.正如Nolan Conaway提到的，当传递给from_json的模式无法应用于给定的字符串时，就会发生这种情况。
问题在于，在这些字符串中，它将顶层视为一个数组，但正如spark_read_df.printSchema()所显示的，由spark.read.json()推断出的模式忽略了数组层。
The Solution
因此，我最终采用的解决方案是，在进行读取时，只考虑模式中的顶层数组。
from pyspark.sql import functions as F
# This one won't work for directly passing to from_json as it ignores top-level arrays in json strings
# (if any)!
# json_object_schema = spark_read_df.schema()
# from_json is a bit more "simple", it directly applies the schema to the string. In this case
# the top level type is actually an array, so a simple fix is to just wrap the schema that
# spark.read.json returned in an ArrayType to match the true JSON string
json_array_schema = ArrayType(spark_read_df.schema, True)
json_extracted_df = raw_json_df.select(
    F.from_json('json_strings', json_array_schema)
        .alias('json_arrays')
>>> json_extracted_df.show()
+--------------------+
|         json_arrays|
+--------------------+
|[{1.0, 1}, {2.0, 2}]|
|          [{3.0, 3}]|
+--------------------+
>>> json_extracted_df.printSchema()
 |-- json_arrays: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- a: double (nullable = true)
 |    |    |-- b: long (nullable = true)
从那里可以用pyspark.sql.functions.explode将对象从阵列中拉出来。
>>> exploded_df = json_extracted_df.select(F.explode('json_arrays').alias('objects'))
>>> exploded_df.show()
+--------+
| objects|
+--------+
|{1.0, 1}|
|{2.0, 2}|
|{3.0, 3}|
+--------+
>>> exploded_df.printSchema()