在 pyspark 中聚合 json 数据

Question

我是 pyspark 的新手，谁能帮我汇总这些数据。

我在这样的文本文件中有 json 数据

{"a":1 , "b":"abc", "c":"abc2", "d":"abc3" , "e":"1234"}
{"a":1 , "b":"abc2", "c":"abc", "d":"abc" ,"e":"1234"}
{"a":1 , "b":"abc", "c":"abc2", "d":"abc3","e":"123"}

我想在 'b'、'c'、'd' 列上聚合数据，同时在 'a' 列中添加值，然后添加一个新列 'unique_e'，这将在基础上给出唯一的 'e' 列值列'e'。

这是我想要的最终输出

{"a":2 , "b":"abc", "c":"abc2", "d":"abc3" , "unique_e":"2"}
{"a":1 , "b":"abc2", "c":"abc", "d":"abc" , "unique_e":"1"}

Answer 1

试试这个：

import pyspark.sql.functions as f
 df = spark.read.json(pathToFile)
 df2 = df.groupby("b","c","d").agg(f.countDistinct(df["e"]).alias("unique_e"), f.sum(df["a"]).alias("sum_a"))

如果您想将其保存到单个 json 文件中，您可以这样做：

df2.coalesce(1).write(pathToDir)

json 将在一个看起来像这样的文件中：part-00000-e3421247-f8cd-4ecb-b8e6-fc26894f5282.json

如果您没有定义 spark（如果您使用 pyspark shell 或正确配置您的 notebook 将被定义），如果您从外部脚本运行代码就是这种情况，您可以执行以下操作：

    spark = SparkSession.builder.master("local").appName("example").getOrCreate()

创建它。