如何使用 Spark 比较两个表的列？

Question

我正在尝试通过读取为 DataFrame 来比较两个表（）。对于这些表中的每个公共列，使用主键连接说 order_id 与 order_date、order_name、order_event 等其他列。

我正在使用的 Scala 代码

val primary_key=order_id
for (i <- commonColumnsList){
      val column_name = i
      val tempDataFrameForNew = newDataFrame.selectExpr(s"concat($primaryKey,$i) as concatenated")
      val tempDataFrameOld = oldDataFrame.selectExpr(s"concat($primaryKey,$i) as concatenated")

      //Get those records which aren common in both old/new tables
      matchCountCalculated = tempDataFrameForNew.intersect(tempDataFrameOld)
      //Get those records which aren't common in both old/new tables
      nonMatchCountCalculated = tempDataFrameOld.unionAll(tempDataFrameForNew).except(matchCountCalculated)

      //Total Null/Non-Null Counts in both old and new tables.
      nullsCountInNewDataFrame = newDataFrame.select(s"$i").filter(x => x.isNullAt(0)).count().toInt
      nullsCountInOldDataFrame = oldDataFrame.select(s"$i").filter(x => x.isNullAt(0)).count().toInt
      nonNullsCountInNewDataFrame = newDFCount - nullsCountInNewDataFrame
      nonNullsCountInOldDataFrame = oldDFCount - nullsCountInOldDataFrame

      //Put the result for a given column in a Seq variable, later convert it to Dataframe.
      tempSeq = tempSeq :+ Row(column_name, matchCountCalculated.toString, nonMatchCountCalculated.toString, (nullsCountInNewDataFrame - nullsCountInOldDataFrame).toString,
       (nonNullsCountInNewDataFrame - nonNullsCountInOldDataFrame).toString)
}
     // Final Step: Create DataFrame using Seq and some Schema.
     spark.createDataFrame(spark.sparkContext.parallelize(tempSeq), schema)

上述代码对于中等数据集运行良好，但随着我的新旧表中列和记录数量的增加，执行时间也在增加。任何形式的建议表示赞赏。 提前谢谢你。

Answer 1

您可以执行以下操作：
1. 在主键上外连接新旧数据框
joined_df = df_old.join(df_new, primary_key, "outer") 2. 尽可能缓存它。这将为您节省大量时间
3. 现在您可以使用 spark 函数遍历列并比较列（.isNull 表示不匹配，== 表示匹配等）

for (col <- df_new.columns){
  val matchCount = df_joined.filter(df_new[col].isNotNull && df_old[col].isNotNull).count()
  val nonMatchCount = ...
}

这应该会快很多，尤其是当您可以缓存数据帧时。如果你不能，这可能是个好主意，所以将加入的 df 保存到磁盘，以避免每次都洗牌