如何使用pysp将具有多个可能值的Json数组列表转换为dataframe中的列

Question

我通过Databricks（Spark+python3.5）中的pythonsdk使用Google管理报告API。在

它以以下格式返回数据（Databricks-pyspark代码）：

dbutils.fs.put("/tmp/test.json", '''{
    "userEmail": "rod@test.com", 
    "parameters": [
        {
            "intValue": "0",
            "name": "classroom:num_courses_created"
        },
        {
            "boolValue": true,
            "name": "accounts:is_disabled"
        },
        {
            "name": "classroom:role",
            "stringValue": "student"
        }
    ]
}''', True)

有188个参数，每个参数可以是int、bool、date或string。根据字段类型，Api返回适当值中的值（例如int字段的intValue和boolean的boolValue）。在

我将这个JSON原封不动地写到我的datalake中，然后通过将其加载到spark dataframe中进行处理：

^{pr2}$

这将生成具有以下架构的数据帧：

• 在用户电子邮件：字符串在
• p参数：数组
  • 在元素：结构
    • 在布尔值：布尔值在
    • 在intValue:字符串在
    • 在名称：字符串在
    • stringValue:字符串在

如果我显示数据帧，它显示为

{“boolValue”：null，“intValue”：“0”，“name”：课堂：创建num_课程“，”stringValue“：null} {“boolValue”：真，“intValue”：null，“name”：帐户：禁用了吗“，”stringValue“：null}
{“boolValue”：null，“intValue”：null，“name”：课堂：角色“，”stringValue“：”student“}

如您所见，它为不存在的typevalue推断了null。在

我想要的结束状态是数据帧中的列，如：

并且透视列的类型将正确（例如课堂：创建num_课程类型为int-参见上面的黄色列）

以下是我目前所做的尝试：

from pyspark.sql.functions import explode
tempDf = testJsonData.select("userEmail", explode("parameters").alias("parameters_exploded"))
explodedColsDf = tempDf.select("userEmail", "parameters_exploded.*")

这将生成具有以下架构的数据帧：

• 在用户电子邮件：字符串在
• 在布尔值：布尔值在
• 在intValue:字符串在
• 在名称：字符串在
• stringValue:字符串在

然后根据Name字段将行透视到列中（即“”课堂：创建num_课程", "课堂：角色“etc（有188个名称/值参数对）：

#turn intValue into an Int column
explodedColsDf = explodedColsDf.withColumn("intValue", explodedColsDf.intValue.cast(IntegerType()))
pivotedDf = explodedColsDf.groupBy("userEmail").pivot("name").sum("intValue")

这将导致以下数据帧：

• 在用户电子邮件：字符串在
• 在帐户：禁用了吗：长
• 在课堂：创建num_课程：长
• 在课堂：角色：长

这是不正确的，因为列的类型是错误的。在

我需要做的是以某种方式查看一个参数列的所有typeValue（除了在原始Json中，它只返回相关的typeValue之外，没有任何方法可以从名称中知道类型或推断它），并且无论哪一个不为null，都是该列的类型。每个参数只出现一次，因此只需为email键输出字符串、bool、int和date值，而不是聚合。在

这超出了我目前所知的范围，但是我认为一个更简单的解决方案可能是回到开头，在我写出Json之前将列旋转到，这样当我将它加载回Spark时，它将是我想要的格式，但是我根本不愿意转换原始数据。我也不希望手工编写188个字段的模式，因为我想动态地选择我想要的字段，所以它需要能够处理这些字段。在

Answer 1

下面的代码将提供的示例JSON转换为dataframe（不使用PySpark）。在

导入库

import numpy as np
import pandas as pd

分配变量

^{pr2}$

将JSON分配给变量

data = [{
"userEmail": "rod@test.com", 
"parameters": [
    {
        "intValue": "0",
        "name": "classroom:num_courses_created"
    },
    {
        "boolValue": true,
        "name": "accounts:is_disabled"
    },
    {
        "name": "classroom:role",
        "stringValue": "student"
    }
    ]
},
{
"userEmail": "EMAIL2@test.com", 
"parameters": [
    {
        "intValue": "1",
        "name": "classroom:num_courses_created"
    },
    {
        "boolValue": false,
        "name": "accounts:is_disabled"
    },
    {
        "name": "classroom:role",
        "stringValue": "student2"
    }
    ]
}

]

将字典转换为列的函数

def get_col(x):
    y = pd.DataFrame(x, index=[0])
    col_name = y.iloc[0]['name']
    y = y.drop(columns=['name'])
    y.columns = [col_name]
    return y

迭代JSON列表

df = pd.DataFrame()

for item in range(len(data)):

    # Initialize empty dataframe
    trow = pd.DataFrame()
    temp = pd.DataFrame(data[item])

    for i in range(temp.shape[0]):

        # Read each row
        x = temp.iloc[i]['parameters']
        trow = pd.concat([trow,get_col(x)], axis=1)
        trow['userEmail'] = temp.iloc[i]['userEmail']


    df = df.append(trow) 

# Rearrange columns, drop those that are not needed
df = df[['userEmail', 'classroom:num_courses_created', 'accounts:is_disabled', 'classroom:role']]

输出：

。。。。。。。。。。。。。。。。。。。。。。。。。上一次编辑。。。。。。。。。。。。。。。。。。。。。在

将JSON/嵌套字典转换为数据帧

temp = pd.DataFrame(data)

# Initialize empty dataframe
df = pd.DataFrame()
for i in range(temp.shape[0]):
    # Read each row
    x = temp.iloc[i]['parameters']
    temp1 = pd.DataFrame([x], columns=x.keys())
    temp1['userEmail'] = temp.iloc[i]['userEmail']

    # Convert nested key:value pairs
    y = x['name'].split(sep=':')
    temp1['name_' + y[0]] = y[1]

    # Combine to dataframe
    df = df.append(temp1, sort=False)

# Rearrange columns, drop those that are not needed
df = df[['userEmail', 'intValue', 'stringValue', 'boolValue', 'name_classroom', 'name_accounts']]

输出

编辑-1 根据更新后的问题截图，下面的代码应该可以工作。在

分配变量