java - Spark Group By Key to (String, Iterable)

Question

我正在尝试按键对 urldata 进行分组，其中值是字符串

示例数据:

url_3 url_2
url_3 url_2
url_3 url_1
url_4 url_3
url_4 url_1

预期结果:

(url_3,(url_2,url_1))
(url_4,(url_3,url_1))

1)加载url数据:

Dataset<String> lines = spark.read()
    .textFile("C:/Users/91984/workspace/myApp/src/test/resources/in/urldata.txt");

2)使用空格分割数据集

Encoder<Tuple2<String, String>> encoder2 = 
    Encoders.tuple(Encoders.STRING(), Encoders.STRING());
Dataset<Tuple2<String, String>> tupleRDD = lines.map(f->{
    Tuple2<String, String> m = 
        new Tuple2<String, String>(f.split(" ")[0], f.split(" ")[1]);
    return m;
},encoder2);

3)使用groupbyKey对key上的tupleRDD数据库进行分组

KeyValueGroupedDataset<String, Tuple2<String, String>> keygrpDS = 
    tupleRDD.groupByKey(f->f._1, Encoders.STRING());

有人可以解释一下为什么第 3 步的 groupByKey 返回 KeyValueGroupedDataset<String, Tuple2<String, String>>而不是KeyValueGroupedDataset<String, Iterable<String>>以及需要进行哪些更改才能获得预期结果。

Answer 1

这就是它在 Spark 中处理数据集的方式。当您有 Dataset<T> 类型的数据集时，您可以通过某种映射函数对其进行分组，该映射函数采用 T 类型的对象并返回 K 类型的对象(键)。你得到的是 KeyValueGroupedDataset<K,T>您可以在其上调用聚合函数(请参阅 the javadoc )。在你的情况下，你可以使用 mapGroups您可以提供一个映射键 K 的函数。和一个可迭代的 Iterable<T>到您选择的新对象 R。如果有帮助的话，在您的代码中，T 是一个 Tuple2，K 是一个 URL。