在pyspark中读取嵌套的JSON文件，其中structType列的值为string

Question

{“traffic\u fource”：“{name'：'intgreints'，medium'：'（none）'，source'：'（direct）}}”

这是一个拼花文件，它有json格式的数据，但值部分在双引号中，这使它成为一个字符串而不是StructType，我想不知道它在诸如traffic\u fource\u name和value等列之间有“_”将是int，然后traffic\u fource\u medium，值将是（none）。我已经有了一个函数，只有当它识别出数据类型是structType或ArrayType时，才会按我的需要取消对其进行测试，但在这里，我用双引号括起了值。因此，我已经完成了以下步骤，以将其识别为结构类型：

        schema_traffic = StructType([StructField("name",StringType()),StructField("medium",StringType()),StructField("source",StringType())])
        df = df.withColumn("traffic_source", from_json(inp_df.traffic_source, schema_traffic))

这很好用。但我面临的问题是，当我试图通过参数文件传递模式时。模式始终为字符串格式，其中from_json失败。从param文件中读取模式后，看起来如下所示：

“StructType（[StructField（“name”，StringType（）），StructField（“medium”，StringType（）），StructField（“source”，StringType（））]）”

Answer 1

读取文件时，内容将仅推断为字符串。

您可以使用如下所示内容的配置文件。添加列名和相应类型：

name:string
medium:string
source:string

读取此文件并动态创建StructType架构。我使用简单的文件读取器来读取和解析配置，如果需要，您可以使用任何其他配置解析器库。

type_to_spark = {"int": IntegerType(), "string": StringType()}

with open("config.conf") as f:
    contents = f.read().splitlines()

def prepare_struct(config_lines):
    configs = config_lines.split(":")
    return StructField(configs[0], type_to_spark[configs[1]])

schema = StructType(list(map(lambda s: prepare_struct(s), contents)))

# Now pass this schema to from_json

您可以将DDL字符串形式的架构传递给from_json。示例代码：

>>> df = spark.createDataFrame([('{"id": 1,"name": {"fn": "jk","ln": "jh"}}',)]).toDF("json")
>>> df.show(truncate=False)
+-----------------------------------------+
|json                                     |
+-----------------------------------------+
|{"id": 1,"name": {"fn": "jk","ln": "jh"}}|
+-----------------------------------------+

>>> outDF = df.select(from_json("json","id int, name struct<fn:string,ln:string>"))
>>> outDF.show()
+---------------+
|from_json(json)|
+---------------+
|  {1, {jk, jh}}|
+---------------+

>>> outDF.printSchema()
root
 |-- from_json(json): struct (nullable = true)
 |    |-- id: integer (nullable = true)
 |    |-- name: struct (nullable = true)
 |    |    |-- fn: string (nullable = true)
 |    |    |-- ln: string (nullable = true)

您可以将ddl模式放在一个文件中，然后读取该文件，将模式存储在一个变量中，并将模式变量传递给from_json。

Answer 2

太棒了。我确实有嵌套的列，所以如果我必须再次使用结构类型，那么我可以怎么做：“FireAnalyse": "[{'值”：“firebase_previous_screen”，“键”：{'string_value'：“NewUser”，“int_value”：“无”，“float_value”：“无”，“double_value”：“无'}}, {'值”：“firebase_previous_class”，“键”：{'string_value'：“用户详细信息”，“int_value”：“无”，“float_value”：无，“double_value”：无}]”

所以首先我删除了数组括号：inp_df=inp_df.with列（"FireAnalyse"，func.split（inp_df['FireAnalyse'], ']')[0]）inp_df=inp_df.with列（"FireAnalyse"，func.split（inp_df['FireAnalyse'], '[')[1]）

        schema = StructType([StructField("key",StringType()),StructField("value",StructType([StructField("string_value",StringType()),StructField("int_value",IntegerType()),StructField("float_value",FloatType()),StructField("double_value",DoubleType())]))])
        inp_df = inp_df.withColumn("event_params", from_json(inp_df.event_params, schema))