从 PySpark 数据框创建嵌套 JSON答案

【问题标题】：Create Nested JSON out of PySpark Dataframe从 PySpark 数据框创建嵌套 JSON
【发布时间】：2020-01-21 20:12:18
【问题描述】：

这个平面 json 到 pyspark 中的嵌套 json。

{
    'event_type': 'click', 
    'id': '223',
    'person_id': 201031940, 
    'category': 'Chronicles', 
    'approved_content': 1
}

到

{
    'event_type': 'click', 
    user: {
        'id': '223',
        'person_id': 201031940
    },
    event: {
        'category': 'Chronicles', 
        'approved_content': 1
    }
}

【问题讨论】：

添加用户密钥。添加事件键。在事件类型之后调用每个值。删除那些键值对。将调用的键值对插入到添加的键中。

标签： json dataframe apache-spark pyspark nested

【解决方案1】：

您也可以不使用更有效的 udfs 来完成此操作，并且在处理大量记录时会产生显着影响：

import pyspark.sql.fuctions as f
events_schema = StructType([
    StructField('event_type', StringType(), True),
    StructField('id', StringType(), True),
    StructField('person_id', StringType(), True),
    StructField('category', StringType(), True),
    StructField('approved_content', StringType(), True),
])

events = [{
    'event_type': 'click',
    'id': '223',
    'person_id': 201031940,
    'category': 'Chronicles',
    'approved_content': 1
}]
df = spark.createDataFrame(events, schema=events_schema)
newDf = (df
          .withColumn('user', f.struct(df.id, df.person_id))
          .withColumn('event', f.struct(df.category, df.approved_content))
          .withColumn('nestedEvent', f.struct(f.col('user'), f.col('event')))
          .select('nestedEvent'))

【讨论】：

【解决方案2】：

你可以这样做：

定义架构，并使用架构将平面 json 转换为数据框。
注册几个 UDF 以构建用户和事件映射。
使用 #2 中的 UDF 寄存器在数据框中添加新列（用户和事件）
删除多余的列

完整代码如下：

from pyspark.sql.types import (
    StringType,
    StructField,
    StructType,
    MapType
)
from pyspark.sql.functions import udf

events_schema = StructType([
    StructField('event_type', StringType(), True),
    StructField('id', StringType(), True),
    StructField('person_id', StringType(), True),
    StructField('category', StringType(), True),
    StructField('approved_content', StringType(), True),
])

events = [{
    'event_type': 'click',
    'id': '223',
    'person_id': 201031940,
    'category': 'Chronicles',
    'approved_content': 1
}]
df = spark.createDataFrame(events, schema=events_schema)

build_user_udf = udf(lambda id, person_id: {
    'id': id,
    'person_id': person_id
}, MapType(StringType(), StringType()))

build_event_udf = udf(lambda category, approved_content: {
    'category': category,
    'approved_content': approved_content
}, MapType(StringType(), StringType()))

nested_event_df = (
    df
    .withColumn('user', build_user_udf(df['id'], df['person_id']))
    .withColumn('event', build_event_udf(df['category'], df['approved_content']))
    .drop('id')
    .drop('person_id')
    .drop('category')
    .drop('approved_content')
)

nested_event_df.toJSON().first()

'{"event_type":"click","user":{"id":"223","person_id":"201031940"},"event":{"approved_content":"1","category ":"编年史"}}'

nested_event_df.take(1)

[Row(event_type='click', user={'id': '223', 'person_id': '201031940'}, event={'approved_content': '1', 'category': 'Chronicles' })]

这是一个非常基础的版本，但如果你愿意，你可以做更多的优化。

【讨论】：

太棒了，这就是我要找的。次要的事情，我可以让嵌套列中的字段在它们各自的引号中，而不是在值中
我们需要在注册 UDF 时应用类型 - 我进行了必要的更改并更新了代码 sn-p。