我试图在databricks上读取带有json列的pysparkDataframe。
Dataframe：
year month json_col2010 09 [{"p_id":"vfdvtbe"}, {"p_id":"cdscs"}, {"p_id":"usdvwq"}]2010 09 [{"p_id":"ujhbe"}, {"p_id":"cdscs"}, {"p_id":"yjev"}]2007 10 [{"p_id":"ukerge"}, {"p_id":"ikrtw"}, {"p_id":"ikwca"}]2007 10 [{"p_id":"unvwq"}, {"p_id":"cqwcq"}, {"p_id":"ikwca"}]
我需要一个新的Dataframe与所有重复的“p\u id”被删除，并按年度和月份聚合
year month p_id (string)2010 09 ["vfdvtbe", "cdscs", "usdvwq", "ujhbe", "yjev"]2007 10 ["ukerge", "ikrtw", "ikwca", "unvwq", "cqwcq"]
新列“p\u id”是一个数组字符串。我想数一数什么是不同的“p\u id”以及每年和每月有多少个。并且，还要删除在同一年和同一个月出现的重复元素。
我的代码：
from pyspark.sql.types import *from pyspark.sql.functions import *schema = ArrayType(StructType([StructField('p_id', StringType(), True)]))schema = ArrayType(MapType(StringType(),StringType()))t = ff.withColumn("data",F.explode(F.from_json(F.col("json_col"),schema))).withColumn("data",F.when(F.col("data")["product_id"].cast("string").isNotNull(),F.col("data")["product_id"])).filter(F.col("data").isNotNull()).drop("json_col")display(t)
我不确定这是否可以删除重复？
谢谢
1条答案
按热度按时间zrfyljdw1#
使用
flatten, array_distinct与groupBy, collect_list本案例的功能。Example:```df.show(10,False)
+----+-----+---------------------------------------------------------+
|year|month|json_col |
+----+-----+---------------------------------------------------------+
|2010|09 |[{"p_id":"vfdvtbe"}, {"p_id":"cdscs"}, {"p_id":"usdvwq"}]|
|2010|09 |[{"p_id":"ujhbe"}, {"p_id":"cdscs"}, {"p_id":"yjev"}] |
|2007|10 |[{"p_id":"ukerge"}, {"p_id":"ikrtw"}, {"p_id":"ikwca"}] |
|2007|10 |[{"p_id":"unvwq"}, {"p_id":"cqwcq"}, {"p_id":"ikwca"}] |
+----+-----+---------------------------------------------------------+
from pyspark.sql.types import *
from pyspark.sql.functions import *
schema = ArrayType(StructType(
[
StructField('p_id', StringType(), True)
]
))
df1=df.withColumn("ff",from_json(col("json_col"),schema)).
select("year","month",expr('transform(ff , f -> f.p_id)').alias("tmp"))
df1.groupBy("year","month").
agg(to_json(array_distinct(flatten(collect_list(col("tmp"))))).alias("p_id")).
show(10,False)
+----+-----+-------------------------------------------+
|year|month|p_id |
+----+-----+-------------------------------------------+
|2010|09 |["vfdvtbe","cdscs","usdvwq","ujhbe","yjev"]|
|2007|10 |["ukerge","ikrtw","ikwca","unvwq","cqwcq"] |
+----+-----+-------------------------------------------+