Spark RDD 使用列表加入操作答案

【问题标题】：Spark RDDs join operation with listsSpark RDD 使用列表加入操作
【发布时间】：2020-10-13 12:49:01
【问题描述】：

我有以下 RDD：

JavaPairRDD<List<String>, String> firstRDD = ...
firstRDD.foreach(row -> System.out.println(row._1() + ", " + row._2()));
// [Man, Parent], Father

JavaPairRDD<List<String>, String> secondRDD = ...
secondRDD.foreach(row -> System.out.println(row._1() + ", " + row._2()));
// [Man, Parent, Father], Person

我想执行一个内连接，这样一行就等于另一行如果左键是IN（即，子列表）右键（在前面的例子中，[Man, Parent] 在[Man, Parent, Father])。

有什么建议吗？

谢谢！

【问题讨论】：

继续使用 scala 和数据框是我的建议。生活变得轻松多了

标签： java apache-spark java-8 rdd

【解决方案1】：

对于 RDD（以及 JavaPairRDD），join 操作只能检查完全匹配的键。

因此我们必须将 RDD 转换为 Dataframe：

public static Dataset<Row> toDataframe(SparkSession spark, JavaPairRDD<List<String>, String> rdd) {
    JavaRDD<Row> rowRDD1 = rdd.map(tuple -> {
        Seq<String> key = JavaConverters.asScalaIteratorConverter(tuple._1().iterator()).asScala().toSeq();
        return RowFactory.create(key, tuple._2());
    });
    StructType st = new StructType()
            .add(new StructField("key", DataTypes.createArrayType(DataTypes.StringType), true, new MetadataBuilder().build()))
            .add(new StructField("value", DataTypes.StringType, true, new MetadataBuilder().build()));
    return spark.createDataFrame(rowRDD1, st);
}

对于连接条件，我们需要一个 UDF 来检查一个数组是否是另一个数组的一部分。如果元素的顺序不重要，也可以使用array_intersect。

UserDefinedFunction contains = functions.udf((Seq<String> a, Seq<String> b) -> b.containsSlice(a), DataTypes.BooleanType);

将这两个元素放在一起，我们得到

Dataset<Row> df1 = toDataframe(spark, firstRDD);
Dataset<Row> df2 = toDataframe(spark, secondRDD);
Dataset<Row> result = df1.join(df2,contains.apply(df1.col("key"), df2.col("key")));

输入数据

firstRDD        secondRDD
+------+-----+  +------------+-----+
|   key|value|  |         key|value|
+------+-----+  +------------+-----+
|[a, b]|    A|  |   [a, b, c]|    C|
|[b, a]|    B|  |[a, b, c, d]|    D|
+------+-----+  +------------+-----+

我们得到

+------+-----+------------+-----+
|   key|value|         key|value|
+------+-----+------------+-----+
|[a, b]|    A|   [a, b, c]|    C|
|[a, b]|    A|[a, b, c, d]|    D|
+------+-----+------------+-----+

请注意，使用 UDF 作为连接条件可能不是 fastest option。

【讨论】：