Scala Spark 中同一 Spark Dataframe 列上的顺序动态过滤器答案

【问题标题】：Sequential Dynamic filters on the same Spark Dataframe Column in Scala SparkScala Spark 中同一 Spark Dataframe 列上的顺序动态过滤器
【发布时间】：2019-09-12 17:41:12
【问题描述】：

我有一个名为 root 的列，需要根据根列的不同值过滤数据框。

假设我在 root 中有一个值是父、子或子子，我想通过变量动态应用这些过滤器。

val x = ("parent,child,sub-child").split(",")
x.map(eachvalue <- {

   var df1 = df.filter(col("root").contains(eachvalue))

}

但是当我这样做时，它总是会覆盖 DF1，我想应用所有 3 个过滤器并获得结果。

将来我可能会将列表扩展到任意数量的过滤器值，并且代码应该可以工作。

谢谢，呸

【问题讨论】：

根列的类型是什么？

标签： scala apache-spark

【解决方案1】：

您应该将后续过滤器应用于前一个过滤器的结果，而不是df：

val x = ("parent,child,sub-child").split(",")
var df1 = df
x.map(eachvalue <- {
    df1 = df1.filter(col("root").contains(eachvalue))
}

df1 映射操作后将应用所有过滤器。

【讨论】：

【解决方案2】：

让我们看一个 spark shell 的例子。希望对你有帮助。

scala> import spark.implicits._
import spark.implicits._

scala> val df0 = 
spark.sparkContext.parallelize(List(1,2,1,3,3,2,1)).toDF("number")
df0: org.apache.spark.sql.DataFrame = [number: int]

scala> val list = List(1,2,3)
list: List[Int] = List(1, 2, 3)


scala> val dfFiltered = for (number <- list) yield { df0.filter($"number" === number)}
dfFiltered: List[org.apache.spark.sql.Dataset[org.apache.spark.sql.Row]] = List([number: int], [number: int], [number: int])

scala> dfFiltered(0).show
+------+
|number|
+------+
|     1|
|     1|
|     1|
+------+


scala> dfFiltered(1).show
+------+
|number|
+------+
|     2|
|     2|
+------+


scala> dfFiltered(2).show
+------+
|number|
+------+
|     3|
|     3|
+------+

【讨论】：

【解决方案3】：

AFAIK isin 可以在这种情况下使用，下面是示例。

import spark.implicits._

val colorStringArr = "red,yellow,blue".split(",")
val colorDF =
  List(
    "red",
    "yellow",
    "purple"
  ).toDF("color")
// to derive a column using a list
colorDF.withColumn(
  "is_primary_color",
  col("color").isin(colorStringArr: _*)
).show()


     println( "if you don't want derived column and directly want to   filter using a list with isin then .. ")
    colorDF.filter(col("color").isin(colorStringArr: _*)).show

结果：

+------+----------------+
| color|is_primary_color|
+------+----------------+
|   red|            true|
|yellow|            true|
|purple|           false|
+------+----------------+

如果您不想要派生列并直接使用带有 isin 的列表进行过滤，那么 ....

+------+
| color|
+------+
|   red|
|yellow|
+------+

【讨论】：

【解决方案4】：

另一种使用 array_contains 和交换参数的方法。

scala> val x = ("parent,child,sub-child").split(",")
x: Array[String] = Array(parent, child, sub-child)

scala> val df = Seq(("parent"),("grand-parent"),("child"),("sub-child"),("cousin")).toDF("root")
df: org.apache.spark.sql.DataFrame = [root: string]

scala> df.show
+------------+
|        root|
+------------+
|      parent|
|grand-parent|
|       child|
|   sub-child|
|      cousin|
+------------+


scala> df.withColumn("check", array_contains(lit(x),'root)).show
+------------+-----+
|        root|check|
+------------+-----+
|      parent| true|
|grand-parent|false|
|       child| true|
|   sub-child| true|
|      cousin|false|
+------------+-----+


scala>

【讨论】：

【解决方案5】：

这是我的两分钱

val filters = List(1,2,3)
val data = List(5,1,2,1,3,3,2,1,4)
val colName = "number"
val df = spark.
  sparkContext.
  parallelize(data).
  toDF(colName).
  filter(
     r => filters.contains(r.getAs[Int](colName))
  )

df.show()

导致

+------+
|number|
+------+
|     1|
|     2|
|     1|
|     3|
|     3|
|     2|
|     1|
+------+

【讨论】：