使用scala的布尔运算在火花数据帧中折叠列

Question

我们如何使用scala使用OR操作将布尔列折叠成一行？

第1部分：

A  true
A  false
B  false
B  false
C  true
B  false
A  true
C  true

期望输出

B  false
A  true
C  true

我能想到的一个解决方案是按第一列条目对它们进行分组，filter true

这个解决方案相当混乱。此外，不知道这是否适用于所有边缘情况。有什么聪明的方法可以做到这一点吗？

编辑：给定的答案适用于上述给定的场景，但不适用于此场景。有什么方法可以实现所需的输出？

第2部分：

A  true    "Apple"
A  false   ""
B  false   ""
B  false   ""
C  true    "Cat"
C  true    "Cotton"
C  false   ""

期望输出

B  false []
A  true  ["Apple"]
C  true  ["Cat","Cotton"]

我试图通过col1和col2分组，然后用collect_set折叠col3来实现这一点，然后

1. 按第一列分组
2. 将第二列收集为布尔值集
3. 检查是否只有一个true如果是，则OR表达式的计算结果将始终为true

但这会导致col3_集合的丢失。

Answer 1

试试这个：

1. col1上的分组

将步骤2中收集的布尔值集传递给UDF，UDF将对所有元素执行一个简单的递归操作。

scala> val df = List(
     | ("A",  true),
     | ("A",  false),
     | ("B",  false),
     | ("B",  false),
     | ("C",  true),
     | ("B",  false),
     | ("A",  true),
     | ("C",  true)
     | ).toDF("col1","col2")
df: org.apache.spark.sql.DataFrame = [col1: string, col2: boolean]

scala> df.show
+----+-----+
|col1| col2|
+----+-----+
|   A| true|
|   A|false|
|   B|false|
|   B|false|
|   C| true|
|   B|false|
|   A| true|
|   C| true|
+----+-----+


scala> val aggOr = udf((a:Seq[Boolean])=>{a.reduceLeft(_||_)})
aggOr: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function1>,BooleanType,Some(List(ArrayType(BooleanType,false))))

scala> df.groupBy("col1").agg(aggOr(collect_set("col2")).as("col2Or")).show
+----+------+
|col1|col2Or|
+----+------+
|   B| false|
|   C|  true|
|   A|  true|
+----+------+

Answer 2

1. 按第1列分组
2. 收集第二列作为布尔集合
3. 收集第3列作为字符串集
4. 检查是否有一个true如果是，那么OR表达式将始终求值为true。
5. 删除空字符串"从col3_set

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

object GroupByAgg {

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

    val spark = Constant.getSparkSess

    import spark.implicits._

    val df = List(("A", true,"Apple"), ("A", false,""),
      ("B", false,""),
      ("B", false,""),
      ("C", true,"Cat"),
      ("C", true,"Cotton"),
      ("C", true,"")).toDF("Col1", "Col2","Col3")

    //Group by 1st column
    df.groupBy("Col1")
      // Collect unique values
      .agg(collect_set("Col2").as("Col2_set"),collect_set("Col3").as("Col3_set"))
      //check if the array contains single true
      .withColumn("OutputCol2", when(array_contains(col("Col2_set"), true), true)
        .otherwise(false))
    .withColumn("OutputCol3",array_remove(col("Col3_set"),lit("")))
//.withColumn("OutputCol3",expr("filter(Col3_set, x -> x != '')"))
      .drop("Col2_set")
      .drop("Col3_set")
      .show()
  }

}

输出：

+----+----------+-------------+
|Col1|OutputCol2|   OutputCol3|
+----+----------+-------------+
|   B|     false|           []|
|   C|      true|[Cat, Cotton]|
|   A|      true|      [Apple]|
+----+----------+-------------+