scala - 如何跨 Array[DataFrame] 组合(加入)信息

Question

我有一个 Array[DataFrame]，我想检查每个数据框的每一行，按列的值是否有任何变化。假设我有第一行的三个数据框，例如:

 (0,1.0,0.4,0.1)
 (0,3.0,0.2,0.1)
 (0,5.0,0.4,0.1)

第一列是 ID，我对这个 ID 的理想输出是:

 (0, 1, 1, 0)

表示第二列和第三列发生了变化，而第三列没有变化。我在此处附上一些数据以复制我的设置

val rdd = sc.parallelize(Array((0,1.0,0.4,0.1),
                               (1,0.9,0.3,0.3),
                               (2,0.2,0.9,0.2),
                               (3,0.9,0.2,0.2),
                               (4,0.3,0.5,0.5)))
val rdd2 = sc.parallelize(Array((0,3.0,0.2,0.1),
                                (1,0.9,0.3,0.3),
                                (2,0.2,0.5,0.2),
                                (3,0.8,0.1,0.1),
                                (4,0.3,0.5,0.5)))
val rdd3 = sc.parallelize(Array((0,5.0,0.4,0.1),
                                (1,0.5,0.3,0.3),
                                (2,0.3,0.3,0.5),
                                (3,0.3,0.3,0.1),
                                (4,0.3,0.5,0.5)))
val df = rdd.toDF("id", "prop1", "prop2", "prop3")
val df2 = rdd2.toDF("id", "prop1", "prop2", "prop3")
val df3 = rdd3.toDF("id", "prop1", "prop2", "prop3")
val result:Array[DataFrame] = new Array[DataFrame](3)
result.update(0, df)
result.update(1,df2)
result.update(2,df3)

如何映射数组并获取输出？

Answer 1

您可以将 countDistinct 与 groupBy 一起使用:

import org.apache.spark.sql.functions.{countDistinct}

val exprs = Seq("prop1", "prop2", "prop3")
  .map(c => (countDistinct(c) > 1).cast("integer").alias(c))

val combined = result.reduce(_ unionAll _)

val aggregatedViaGroupBy = combined
  .groupBy($"id")
  .agg(exprs.head, exprs.tail: _*)

aggregatedViaGroupBy.show
// +---+-----+-----+-----+
// | id|prop1|prop2|prop3|
// +---+-----+-----+-----+
// |  0|    1|    1|    0|
// |  1|    1|    0|    0|
// |  2|    1|    1|    1|
// |  3|    1|    1|    1|
// |  4|    0|    0|    0|
// +---+-----+-----+-----+