代码

(
    df
    .groupBy('x')
    .agg(F.expr("struct(x, count(x)) as counts"))
    .agg(F.expr('map_from_entries(collect_list(counts)) as counts'))
)

结果

+--------------------+
|              counts|
+--------------------+
|{new -> 2, old -> 1}|
+--------------------+

详情

• 对 Dataframe 进行分组并聚合，以创建组键和相应计数的结构，例如{new, 2}
• 再次聚合dataframe，并将所有此类结构体收集到一个列表中，然后使用map_from_entries从结构体列表中创建一个Map

可以在透视聚合之后创建Map列。

agg_sdf = data_sdf. \
    groupBy(func.lit(1).alias('dummy')). \
    pivot("x"). \
    agg(func.count("*").alias("count"))

agg_sdf. \
    select(func.create_map(*reduce(lambda x, y: x+y, [(func.lit(c), func.col(c)) for c in agg_sdf.drop('dummy').columns])).alias('counts')). \
    show(truncate=False)

# +--------------------+
# |counts              |
# +--------------------+
# |{new -> 2, old -> 1}|
# +--------------------+

我迟到了，但这里是另一个版本使用create_map
create_map需要key1,val1,key2,val2,ley3,val3...来创建Map，我们使用列表解析来生成列表

%python
from pyspark.sql import functions as F

df = spark.createDataFrame([('new',), ('new',), ('old',)],["x"])
agg_df = df.groupBy("x").agg(F.count("*").alias("count"))

map_col = F.create_map([F.lit(column) for row in agg_df.collect() for column in row])
agg_df.select('x','count',map_col).show()

输出：

+---+-----+--------------------+
|  x|count| map(new, 2, old, 1)|
+---+-----+--------------------+
|new|    2|{new -> 2, old -> 1}|
|old|    1|{new -> 2, old -> 1}|
+---+-----+--------------------+