java - 如何使用数据集聚合 spark(java) 中的字符串计数

Question

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

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

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

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

我的第一步是

table.groupBy("topic");

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

我怎样才能进行这种聚合？

Answer 1

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

好的，测试数据:

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 函数为每个实体列创建许多行； entities = a,b,c 将创建 3 行