【问题标题】：How to aggregate string counts in spark(java) with datasets如何使用数据集聚合 spark(java) 中的字符串计数
【发布时间】：2017-04-27 18:06:31
【问题描述】：

我有一个具有以下结构的数据集

+--------------------+--------------------+--------------------+-----+ | content | entities | id|topic| +--------------------+--------------------+--------------------+-----+

entities 是一个字符串列表，命名实体，topic 是一个整数。

我想总结一个主题中每个不同实体的外观。

我的第一步是

table.groupBy("topic");

然后我想聚合，但我该怎么做，我很想实现一个自定义聚合器，但是我需要一个 Map<String,Integer> 的编码器，我不知道如何创建。

如何进行这种聚合？

【问题讨论】：

确定：entities 是一个列表，所以 id=1,entities=a,b,c; id=2，实体：b、e、f、r？
是的，正是这个

标签： java apache-spark dataset

【解决方案1】：

如果您只是计算每个主题的实体总数，那么您可以使用计数功能而不必编写自定义聚合器。直接来自 Spark 教程：http://spark.apache.org/docs/latest/sql-programming-guide.html:

table.groupBy("topic").count().show();

【讨论】：

不，我想总结一个主题中每个实体的计数。稍微澄清一下问题

【解决方案2】：

我不确定您究竟要计算什么，但我看到的两个选项都需要使用相同的函数。

好的，测试数据：

import org.apache.spark.sql.types._
import org.apache.spark.sql._
import sqlContext.implicits._

val list = Seq((1, "topic1", Array("a", "b", "c", "b")), (2, "topic1", Array("b", "c", "r")), (3, "topic2", Array("e", "b", "c", "e")), (4, "topic2", Array("b", "c", "e", "r")))
val df = sc.parallelize(list).toDF("id", "topic", "entities");

计算每个主题中每种类型的实体数量：

df
    .withColumn("entity", explode('entities))
    .groupBy('topic, 'entity)
    .count()
    .show();

结果将是：

+------+------+-----+
| topic|entity|count|
+------+------+-----+
|topic1|     a|    1|
|topic1|     b|    3|
|topic1|     c|    2|
|topic1|     r|    1|
|topic2|     b|    2|
|topic2|     c|    2|
|topic2|     e|    3|
|topic2|     r|    1|
+------+------+-----+

每个主题有多少不同的实体：

df
    .withColumn("entity", explode('entities))
    .groupBy('topic)
    .agg(countDistinct('entity))
    .show();

结果将是：

+------+-------------+
| topic|count(entity)|
+------+-------------+
|topic1|            4|
|topic2|            4|
+------+-------------+

答案的关键

您应该使用explode 函数为每个实体列创建许多行； entity = a,b,c 将创建 3 行

【讨论】：