代码之家 › 专栏 › 技术社区 › Shankar Panda

PySpark:如何创建json结构?

pyspark-sql apache-spark-sql pyspark apache-spark python-3.x

Shankar Panda · 技术社区 · 7 年前

样本数据:

Country|SegmentID|total_cnt|max_value|
+---------+---------+---------+---------+
|     Pune|        1|     10.0|       15|
|    Delhi|        1|     10.0|       15|
|Bangalore|        1|     10.0|       15|
|     Pune|        2|     10.0|       16|
|    Delhi|        2|     10.0|       16|
|Bangalore|        2|     10.0|       16|
|     Pune|        3|     15.0|       16|
|    Delhi|        3|     10.0|       16|
|Bangalore|        3|     15.0|       16|
+---------+---------+---------+---------+

这是我的密码:

[{
        "NAME": "SEG1",
        "VAL": 15,
        "CITIES": {
            "Bangalore": 10,
            "Delhi": 10,
            "Pune": 10
        }
    },
    {
        "NAME": "SEG2",
        "VAL": 16,
        "CITIES": {
            "Bangalore": 10,
            "Delhi": 10,
            "Pune": 10
        }
    },
    {
        "NAME": "SEG3",
        "VAL": 16,
        "CITIES": {
            "Bangalore": 15,
            "Delhi": 10,
            "Pune": 15
        }
    }
]

我可以创建一个一级层次结构,但这也不能满足我的要求。

join_df=join_df.toPandas()
j = (join_df.groupby(['SegmentID','max_value'], as_index=False)
                .apply(lambda x: x[['Country','total_cnt']].to_dict('r'))
                .reset_index().rename(columns={0:'CITIES'})
                .to_json(orient='records'))

[{"SegmentID":1,"max_value":15,"Cities":[{"Country":"Pune","total_cnt":10.0},{"Country":"Delhi","total_cnt":10.0},{"Country":"Bangalore","total_cnt":10.0}]},{"SegmentID":2,"max_value":16,"Cities":[{"Country":"Pune","total_cnt":10.0},{"Country":"Delhi","total_cnt":10.0},{"Country":"Bangalore","total_cnt":10.0}]},{"SegmentID":3,"max_value":16,"Cities":[{"Country":"Pune","total_cnt":15.0},{"Country":"Delhi","total_cnt":10.0},{"Country":"Bangalore","total_cnt":15.0}]}]

1 回复 | 直到 7 年前

Soheil Pourbafrani 7 年前

您可以将Dataframe转换为RDD并应用转换:

from pyspark.sql.types import *
import json

NewSchema = StructType([StructField("Name", StringType())
                           , StructField("VAL", IntegerType())
                           , StructField("CITIES", StringType())
                        ])

def reduceKeys(row1, row2):
        row1[0].update(row2[0])
        return row1

res_df = join_df.rdd.map(lambda row: ("SEG" + str(row[1]), ({row[0]: row[2]}, row[3])))\
    .reduceByKey(lambda x, y: reduceKeys(x, y))\
    .map(lambda row: (row[0], row[1][1], json.dumps(row[1][0])))\
    .toDF(NewSchema)

结果如下:

res_df.show(20, False)


+----+---+------------------------------------------------+
|Name|VAL|CITIES                                          |
+----+---+------------------------------------------------+
|SEG1|15 |{"Pune": 10.0, "Delhi": 10.0, "Bangalore": 10.0}|
|SEG3|16 |{"Pune": 15.0, "Delhi": 10.0, "Bangalore": 15.0}|
|SEG2|16 |{"Pune": 10.0, "Delhi": 10.0, "Bangalore": 10.0}|
+----+---+------------------------------------------------+

res_df.coalesce(1).write.format('json').save('output.json')

推荐文章

ÎÎÎ½Î· ÎÎ®Î¹Î½Î¿Ï · Python lxml.html语法错误:使用lxml find时XPATH的谓词无效

1 年前

Cam · Pandas列表日期到日期时间

1 年前

RASEL MAHMUD · 为什么以及如何在is_even()函数内的IF条件中递归X变量在满足0后递增?[副本]

1 年前

jjkennedy · Pandas文本文件导入:当每个文件中存在多个表时,自动选择1个表

1 年前

LMC · Numpy数组布尔索引以获取包含元素

1 年前

vr8ce · 非成对标记中特定字符的正则表达式

1 年前

Kernel · 如果指定了crs参数,shapefile的geopandas.read_file将出错

1 年前

ShaAnder · 为什么sqllachemy返回的是类而不是字符串

1 年前

sixtytrees · detectron2软件包未安装(没有名为“torch”的模块),但我安装了torch

1 年前

Pernoctador · Python映射可以复制吗?我需要参考地图

1 年前