[pyspark入门]Dataframe groupby分组后统计词频WordCount的几种写法

每次写spark分组统计词频总要先到网上抄代码[捂脸]，索性在这里做个总结和记录，也为需要的小伙伴提供参考

wordcount在分布式当中的地位，大概跟初学编程语言时的hello world差不多。Pyspark基础 wordcount.py在此不再赘述了，有需要请参考：WordCount入门

已知一个dataframe，想按照某字段/某列(column)分组后，再对各分组中某String类型的字段统计词频，这里默认已经分好词，直接split即可。

那么，与入门wordcount唯一的区别也就找到了，我们需要对GroupedData进行map-reduce词频统计，而最终的结果也不再是Pair对，而是<(group_field, word), count>Pair对，至于后续是否需要分组排序，则可以再灵活操作了，我们这里只考虑统计词频这一步。

Talk is cheap。直接上代码了

## 首先我们先简单创建个df
group_names = [1,1,1,2,2,2]
texts = [
    'python is best',
    'spark is best',
    'dataframe is best',
    'how to use groupby',
    'how to use mapreduce',
    'how to tap wordcount'
]
data = [{'group_name': x, 'text': y} for x, y in zip(group_names, texts)]
df = spark.createDataframe(map(lambda x: Row(**x), data))
df.show()

输出的结果：
±---------±-------------------+
|group_name| text|
±---------±-------------------+
| 1| python is best|
| 1| spark is best|
| 1| dataframe is best|
| 2| how to use groupby|
| 2|how to use mapreduce|
| 2|how to tap wordcount|
±---------±-------------------+

接下来就是分组统计词频

word_counter = df.rdd.flatMap(lambda x: [(x.group_name, token) for token in x.text.split()])
                .map(lambda x: (x, 1))
                .reduceByKey(lambda x, y: x+y)
                .toDF()
word_counter.show()

输出：

如果后续需要分组查看或排序的话，可以将group_name提取出来，有很多种写法，下面写一个例子。

from pyspark.sql.functions import desc
group_extract_ = udf(lambda x: x[0], types.StringType())
sort_counter = word_counter.withColumn("group_name", group_extract_(col("_1")))
                            .orderBy("group_name", desc("_2"))
                            .withColumnRenamed('_2', 'count')
                            .withColumnRenamed('_1', 'Pair')
sort_counter.show()

输出：