Spark 2.2.0 API：我应该更喜欢哪一个 Dataset 与 Groupby 结合聚合或 RDD 与 ReduceBykey答案

【问题标题】：Spark 2.2.0 API: Which one should i prefer Dataset with Groupby combined with aggregate or RDD with ReduceBykeySpark 2.2.0 API：我应该更喜欢哪一个 Dataset 与 Groupby 结合聚合或 RDD 与 ReduceBykey
【发布时间】：2018-05-17 05:25:00
【问题描述】：

大家好，首先，根据标题，有人可能会说问题已经得到解答，但我的意思是比较 ReduceBykey、GroupBykey 的性能，具体针对 Dataset 和 RDD API。我在许多帖子中看到，ReduceBykey 方法的性能比 GroupByKey 更有效，我当然同意这一点。不过，我有点困惑，如果我们使用 Dataset 或 RDD，我无法弄清楚这些方法的行为。每种情况应该使用哪一种？

我会尝试更具体一些，因此我会提供我的解决方案以及工作代码，我正在等待您尽早提出改进建议。

+---+------------------+-----+
|id |Text1             |Text2|
+---+------------------+-----+
|1  |one,two,three     |one  |
|2  |four,one,five     |six  |
|3  |seven,nine,one,two|eight|
|4  |two,three,five    |five |
|5  |six,five,one      |seven|
+---+------------------+-----+

这里的重点是检查第二个列的每一行是否包含第三个列，然后收集它们的所有ID。例如，第三列«one»的单词出现在ID为1,5,2,3的第二列句子中。

+-----+------------+
|Text2|Set         |
+-----+------------+
|seven|[3]         |
|one  |[1, 5, 2, 3]|
|six  |[5]         |
|five |[5, 2, 4]   |
+-----+------------+

这是我的工作代码

List<Row> data = Arrays.asList(
                RowFactory.create(1, "one,two,three", "one"),
                RowFactory.create(2, "four,one,five", "six"),
                RowFactory.create(3, "seven,nine,one,two", "eight"),
                RowFactory.create(4, "two,three,five", "five"),
                RowFactory.create(5, "six,five,one", "seven")
        );

        StructType schema = new StructType(new StructField[]{
                new StructField("id", DataTypes.IntegerType, false, Metadata.empty()),
                new StructField("Text1", DataTypes.StringType, false, Metadata.empty()),
                new StructField("Text2", DataTypes.StringType, false, Metadata.empty())
        });

        Dataset<Row> df = spark.createDataFrame(data, schema);
        df.show(false);
        Dataset<Row> df1 = df.select("id", "Text1")
                .crossJoin(df.select("Text2"))
                .filter(col("Text1").contains(col("Text2")))
                .orderBy(col("Text2"));

        df1.show(false);

        Dataset<Row> df2 = df1
                .groupBy("Text2")
                .agg(collect_set(col("id")).as("Set"));

        df2.show(false);

我的问题分为 3 个子序列：

为了提高性能，我是否需要在 RDD 中转换 Dataset 并使用 ReduceBykey 而不是 Dataset groupby？
我应该使用哪一个，为什么？数据集或 RDD
如果您能提供一个更有效的替代解决方案（如果我的方法存在），我将不胜感激

【问题讨论】：

标签： java performance apache-spark dataset apache-spark-sql

【解决方案1】：

TL;DR 两者都不好，但如果您使用的是Dataset，请继续使用Dataset。

Dataset.groupBy 的行为类似于reduceByKey 如果与合适的函数一起使用。不幸的是，collect_set 的行为与groupByKey 非常相似，如果重复的数量很少的话。用reduceByKeywon't change a thing重写它。

如果您能提供一个更有效的替代解决方案（如果我的方法存在），我将不胜感激

你能做的最好的就是删除crossJoin：

val df = Seq((1, "one,two,three", "one"),
  (2, "four,one,five", "six"),
  (3, "seven,nine,one,two", "eight"),
  (4, "two,three,five", "five"),
  (5, "six,five,one", "seven")).toDF("id", "text1", "text2")

df.select(col("id"), explode(split(col("Text1"), ",")).alias("w"))
  .join(df.select(col("Text2").alias("w")), Seq("w"))
  .groupBy("w")
  .agg(collect_set(col("id")).as("Set")).show

+-----+------------+
|    w|         Set|
+-----+------------+
|seven|         [3]|
|  one|[1, 5, 2, 3]|
|  six|         [5]|
| five|   [5, 2, 4]|
+-----+------------+

【讨论】：