基于给定的操作列创建一个新的数据集答案

【问题标题】：Create a new dataset based given operation column基于给定的操作列创建一个新的数据集
【发布时间】：2020-05-23 11:24:28
【问题描述】：

我正在使用 spark-sql-2.3.1v 并且有以下场景：

给定一个数据集：

val ds = Seq(
  (1, "x1", "y1", "0.1992019"),
  (2, null, "y2", "2.2500000"),
  (3, "x3", null, "15.34567"),
  (4, null, "y4", null),
  (5, "x4", "y4", "0")
   ).toDF("id","col_x", "col_y","value")

即

+---+-----+-----+---------+
| id|col_x|col_y|    value|
+---+-----+-----+---------+
|  1|   x1|   y1|0.1992019|
|  2| null|   y2|2.2500000|
|  3|   x3| null| 15.34567|
|  4| null|   y4|     null|
|  5|   x4|   y4|        0|
+---+-----+-----+---------+

要求：

我得到了需要从外部执行一些计算的操作列（即operationCol）。

在对“col_x”列执行某些操作时，我需要通过过滤掉所有具有“col_x”空值的记录并返回该新数据集来创建一个新数据集。

同样，在对“col_y”列执行一些操作时，我需要通过过滤掉所有具有“col_y”空值的记录并返回该新数据集来创建一个新数据集。

示例：

val operationCol ="col_x";

if(operationCol === "col_x"){
  //filter out all rows which has "col_x" null and return that new dataset.
}

if(operationCol === "col_y"){
  //filter out all rows which has "col_y" null and return that new dataset.
}

当 operationCol === "col_x" 预期输出时：

+---+-----+-----+---------+
| id|col_x|col_y|    value|
+---+-----+-----+---------+
|  1|   x1|   y1|0.1992019|
|  3|   x3| null| 15.34567|
|  5|   x4|   y4|        0|
+---+-----+-----+---------+

当 operationCol === "col_y" 预期输出时：

+---+-----+-----+---------+
| id|col_x|col_y|    value|
+---+-----+-----+---------+
|  1|   x1|   y1|0.1992019|
|  2| null|   y2|2.2500000|
|  4| null|   y4|     null|
|  5|   x4|   y4|        0|
+---+-----+-----+---------+

如何实现这个预期的输出？换句话说，如何进行数据帧的分支？如何在流程中间创建新的数据框/数据集？

【问题讨论】：

标签： apache-spark apache-spark-sql spark-streaming

【解决方案1】：

您可以使用df.na.drop() 删除包含空值的行。 drop 函数可以将你想要考虑的列的列表作为输入，所以在这种情况下，你可以这样写：

val newDf = df.na.drop(Seq(operationCol))

这将创建一个新的数据框newDf，其中operationCol 中的所有行都已被删除。

【讨论】：

【解决方案2】：

您还可以使用filter 过滤掉空值。


scala> val operationCol = "col_x" // for one column
operationCol: String = col_x

scala> ds.filter(col(operationCol).isNotNull).show(false)
+---+-----+-----+---------+
|id |col_x|col_y|value    |
+---+-----+-----+---------+
|1  |x1   |y1   |0.1992019|
|3  |x3   |null |15.34567 |
|5  |x4   |y4   |0        |
+---+-----+-----+---------+


scala> val operationCol = Seq("col_x","col_y") // For multiple Columns
operationCol: Seq[String] = List(col_x, col_y)

scala> ds.filter(operationCol.map(col(_).isNotNull).reduce(_ && _)).show
+---+-----+-----+---------+
| id|col_x|col_y|    value|
+---+-----+-----+---------+
|  1|   x1|   y1|0.1992019|
|  5|   x4|   y4|        0|
+---+-----+-----+---------+

【讨论】：