如何在两个不同的spark rdd中获取所有不同的记录

Question

对 spark 和 RDD 来说非常新，所以我希望我能很好地解释我所追求的东西，以便有人理解和帮助:)

我有两组非常大的数据，比如说 300 万行和 50 列存储在 hadoop hdfs 中。 我想要做的是将这两个读入 RDD，以便它使用并行性 & 我想返回包含所有不匹配的记录（来自任一 RDD）的第三个 RDD。

希望下面有助于显示我想要做什么... 只是试图以最快最有效的方式找到所有不同的记录......

数据的顺序不一定相同 - rdd1 的第 1 行可能是 rdd2 的第 4 行。

提前非常感谢！！

所以...这似乎在做我想做的事，但似乎很容易正确...

%spark

import org.apache.spark.sql.DataFrame
import org.apache.spark.rdd.RDD
import sqlContext.implicits._
import org.apache.spark.sql._

//create the tab1 rdd.
val rdd1 = sqlContext.sql("select * FROM table1").withColumn("source",lit("tab1"))

//create the tab2 rdd.
val rdd2 = sqlContext.sql("select * FROM table2").withColumn("source",lit("tab2"))

//create the rdd of all misaligned records between table1 and the table2.
val rdd3 = rdd1.except(rdd2).unionAll(rdd2.except(rdd1))

//rdd3.printSchema()    
//val rdd3 = rdd1.except(rdd2)

//drop the temporary table that was used to create a hive compatible table from the last run.
sqlContext.dropTempTable("table3")

//register the new temporary table.
rdd3.toDF().registerTempTable("table3")

//drop the old compare table.
sqlContext.sql("drop table if exists data_base.compare_table")

//create the new version of the s_asset compare table.
sqlContext.sql("create table data_base.compare_table as select * from table3")

这是我到目前为止完成的最后一段代码，它似乎正在完成这项工作 - 不确定完整数据集的性能，让我的手指交叉......

非常感谢所有花时间帮助这个可怜的人 :)

附言如果有人有性能更高的解决方案，我很想听听！ 或者如果您可以看到一些问题，这可能意味着它会返回错误的结果。

Answer 1

1. 将两个数据框加载为df1,df2
2. 添加source列，默认值分别为rdd1和rdd2
3. 联合 df1 和 df2
4. 按"rowid", "name", "status", "lastupdated" 分组并按集合收集其来源
5. 过滤所有具有单一来源的行

import org.apache.spark.sql.functions._

object OuterJoin {

  def main(args: Array[String]): Unit = {

    val spark = Constant.getSparkSess

    import spark.implicits._

    val cols = Array("rowid", "name", "status", "lastupdated")

    val df1 = List(
      ("1-za23f0", "product1", "active", "30-12-2019"),
      ("1-za23f1", "product2", "inactive", "31-12-2019"),
      ("1-za23f2", "product3", "inactive", "01-01-2020"),
      ("1-za23f3", "product4", "inactive", "02-01-2020"),
      ("1-za23f4", "product5", "inactive", "03-01-2020"))
      .toDF(cols:_ *)
      .withColumn("source",lit("rdd1"))

    val df2 = List(
      ("1-za23f0", "product1", "active", "30-12-2019"),
      ("1-za23f1", "product2", "active", "31-12-2019"),
      ("1-za23f2", "product3", "active", "01-01-2020"),
      ("1-za23f3", "product1", "inactive", "02-01-2020"),
      ("1-za23f4", "product5", "inactive", "03-01-2020"))
      .toDF(cols:_ *)
        .withColumn("source",lit("rdd2"))

    df1.union(df2)
      .groupBy(cols.map(col):_ *)
      .agg(collect_set("source").as("sources"))
      .filter(size(col("sources")) === 1)
      .withColumn("from_rdd", explode(col("sources") ))
      .drop("sources")
      .show()
  }

}

Answer 2

您可以将数据读入数据帧而不是 Rdds，然后使用 union 和 group by 来实现结果

Answer 3

两者都可以与“full_outer”连接，然后应用过滤器，其中字段值在两者中进行比较：

val filterCondition = cols
  .map(c => (col(s"l.$c") =!= col(s"r.$c") || col(s"l.$c").isNull || col(s"r.$c").isNull))
  .reduce((acc, c) => acc || c)

df1.alias("l")
  .join(df2.alias("r"), $"l.rowid" === $"r.rowid", "full_outer")
  .where(filterCondition)

输出：

+--------+--------+--------+-----------+------+--------+--------+--------+-----------+------+
|rowid   |name    |status  |lastupdated|source|rowid   |name    |status  |lastupdated|source|
+--------+--------+--------+-----------+------+--------+--------+--------+-----------+------+
|1-za23f1|product2|inactive|31-12-2019 |rdd1  |1-za23f1|product2|active  |31-12-2019 |rdd2  |
|1-za23f2|product3|inactive|01-01-2020 |rdd1  |1-za23f2|product3|active  |01-01-2020 |rdd2  |
|1-za23f3|product4|inactive|02-01-2020 |rdd1  |1-za23f3|product1|inactive|02-01-2020 |rdd2  |
+--------+--------+--------+-----------+------+--------+--------+--------+-----------+------+