Pyspark：将 CSV 转换为嵌套 JSON

Question

我是 pyspark 的新手。我有一个要求，我需要将 hdfs 位置的大 CSV 文件转换为基于不同 primaryId 的多个嵌套 JSON 文件。

示例输入： data.csv

**PrimaryId,FirstName,LastName,City,CarName,DogName**
100,John,Smith,NewYork,Toyota,Spike
100,John,Smith,NewYork,BMW,Spike
100,John,Smith,NewYork,Toyota,Rusty
100,John,Smith,NewYork,BMW,Rusty
101,Ben,Swan,Sydney,Volkswagen,Buddy
101,Ben,Swan,Sydney,Ford,Buddy
101,Ben,Swan,Sydney,Audi,Buddy
101,Ben,Swan,Sydney,Volkswagen,Max
101,Ben,Swan,Sydney,Ford,Max
101,Ben,Swan,Sydney,Audi,Max
102,Julia,Brown,London,Mini,Lucy

示例输出文件：

文件1： Output_100.json

{
    "100": [
        {
            "City": "NewYork", 
            "FirstName": "John", 
            "LastName": "Smith", 
            "CarName": [
                "Toyota", 
                "BMW"
            ], 
            "DogName": [
                "Spike", 
                "Rusty"
            ]
        }
}

文件2： Output_101.json

{
    "101": [
        {
            "City": "Sydney", 
            "FirstName": "Ben", 
            "LastName": "Swan", 
            "CarName": [
                "Volkswagen", 
                "Ford", 
                "Audi"
            ], 
            "DogName": [
                "Buddy", 
                "Max"
            ]
        }
}

文件3： Output_102.json

{
    "102": [
        {
            "City": "London", 
            "FirstName": "Julia", 
            "LastName": "Brown", 
            "CarName": [
                "Mini"
            ], 
            "DogName": [
                "Lucy"
            ]
        }
    ]
}

我们将不胜感激。

Answer 1

看来您需要对 Id 进行分组，并将 Cars 和 Dogs 收集为一组。

从 pyspark.sql.functions 导入 collect_set

df = spark.read.format("csv").option("header", "true").load("cars.csv")
df2 = (
    df
    .groupBy("PrimaryId","FirstName","LastName")
    .agg(collect_set('CarName').alias('CarName'), collect_set('DogName').alias('DogName'))
)
df2.write.format("json").save("cars.json", mode="overwrite")

生成的文件：

{"PrimaryId":"100","FirstName":"John","LastName":"Smith","CarName":["Toyota","BMW"],"DogName":["Spike","Rusty"]}

{"PrimaryId":"101","FirstName":"Ben","LastName":"Swan","CarName":["Ford","Volkswagen","Audi"],"DogName":["Max","Buddy"]}

{"PrimaryId":"102","FirstName":"Julia","LastName":"Brown","CarName":["Mini"],"DogName":["Lucy"]}

如果这就是你要找的，请告诉我。

Answer 2

您可以使用 pandas.groupby() 按 Id 分组，然后遍历 DataFrameGroupBy 对象创建对象并写入文件。

您需要通过$ pip install pandas 将pandas 安装到您的virtualenv 中。

# coding: utf-8
import json
import pandas as pd


def group_csv_columns(csv_file):
    df = pd.read_csv(csv_file)
    group_frame = df.groupby(['PrimaryId'])

    for i in group_frame:
        data_frame = i[1]
        data = {}
        data[i[0]] = [{
            "City": data_frame['City'].unique().tolist()[0],
            "FirstName": data_frame['FirstName'].unique().tolist()[0],
            "CarName": data_frame['CarName'].unique().tolist(),
            'DogName': data_frame['DogName'].unique().tolist(),
            'LastName': data_frame['LastName'].unique().tolist()[0],
        }]
        # Write to file
        file_name = 'Output_' + str(i[0]) + '.json'
        with open(file_name, 'w') as fh:
            contents = json.dumps(data)
            fh.write(contents)


group_csv_columns('/tmp/sample.csv')

调用group_csv_columns()，文件名包含csv内容。

查看pandas docs