如何使用过滤器从 scala 中的数据帧获取包含空值的行集

Question

我是 Spark 的新手，对基于空条件过滤数据帧有疑问。 我已经经历了许多答案，这些答案都有解决方案，例如

df.filter(($"col2".isNotNULL) || ($"col2" !== "NULL")  || ($"col2" !== "null")  || ($"col2".trim !== "NULL"))

但就我而言，我无法编写硬编码的列名，因为我的架构不是固定的。我正在阅读csv文件，根据其中的列，我必须过滤数据帧中的空值，并希望它在另一个数据帧中。简而言之，任何具有 null 值的列，该完整行都应位于不同的数据帧下。

例如： 输入数据帧 ：

+----+----+---------+---------+
|name|  id|    email|  company|
+----+----+---------+---------+
|  n1|null|[e-mail]|[c1,1,d1]|
|  n2|   2|null     |[c1,1,d1]|
|  n3|   3|[e-mail]| null    |
|  n4|   4|[e-mail]|[c2,2,d2]|
|  n6|   6|[e-mail]|[c2,2,d2]|

输出：

+----+----+---------+---------+
|name|  id|    email|  company|
+----+----+---------+---------+
|  n1|null|[e-mail]|[c1,1,d1]|
|  n2|   2|null     |[c1,1,d1]|
|  n3|   3|[e-mail]| null    |

提前谢谢你。

Answer 1

试试这个-

val df1 = spark.sql("select col1, col2 from values (null, 1), (2, null), (null, null), (1,2) T(col1, col2)")
/**
* +----+----+
* |col1|col2|
* +----+----+
* |null|1   |
* |2   |null|
* |null|null|
* |1   |2   |
* +----+----+
*/
df1.show(false)
df1.filter(df1.columns.map(col(_).isNull).reduce(_ || _)).show(false)
/**
* +----+----+
* |col1|col2|
* +----+----+
* |null|1   |
* |2   |null|
* |null|null|
* +----+----+
*/

Answer 2

非常感谢您的回答。我尝试了以下逻辑，它对我有用。

var arrayColumn = df.columns;
val filterString = String.format(" %1$s is null or %1$s == '' "+ arrayColumn(0));
val x = new StringBuilder(filterString);
for(i <- 1 until arrayColumn.length){
if (x.toString() != ""){
x ++= String.format("or %1$s is null or %1$s == '' ", arrayColumn(i))
}
}
val dfWithNullRows =  df.filter(x.toString());

Answer 3

为了处理空值和数据帧，Spark 有一些有用的函数。

我将展示一些具有不同列数的数据帧示例。

val schema = StructType(List(StructField("id", IntegerType, true), StructField("obj",DoubleType, true)))
val schema1 = StructType(List(StructField("id", IntegerType, true), StructField("obj",StringType, true), StructField("obj",IntegerType, true)))
val t1 = sc.parallelize(Seq((1,null),(1,1.0),(8,3.0),(2,null),(3,1.4),(3,2.5),(null,3.7))).map(t => Row(t._1,t._2))
val t2 = sc.parallelize(Seq((1,"A",null),(2,"B",null),(3,"C",36),(null,"D",15),(5,"E",25),(6,null,7),(7,"G",null))).map(t => Row(t._1,t._2,t._3))
val tt1 = spark.createDataFrame(t1, schema)
val tt2 = spark.createDataFrame(t2, schema1)
tt1.show()
tt2.show()
// To clean all rows with null values
val dfWithoutNull = tt1.na.drop()
dfWithoutNull.show()
val df2WithoutNull = tt2.na.drop()
df2WithoutNull.show()
// To fill null values with another value
val df1 = tt1.na.fill(-1)
df1.show()
// to get new dataframes with the null values rows
val nullValues = tt1.filter(row => row.anyNull == true)
nullValues.show()
val nullValues2 = tt2.filter(row => row.anyNull == true)
nullValues2.show()

输出

// input dataframes
+----+----+
|  id| obj|
+----+----+
|   1|null|
|   1| 1.0|
|   8| 3.0|
|   2|null|
|   3| 1.4|
|   3| 2.5|
|null| 3.7|
+----+----+
+----+----+----+
|  id| obj| obj|
+----+----+----+
|   1|   A|null|
|   2|   B|null|
|   3|   C|  36|
|null|   D|  15|
|   5|   E|  25|
|   6|null|   7|
|   7|   G|null|
+----+----+----+
// Dataframes without null values
+---+---+
| id|obj|
+---+---+
|  1|1.0|
|  8|3.0|
|  3|1.4|
|  3|2.5|
+---+---+
+---+---+---+
| id|obj|obj|
+---+---+---+
|  3|  C| 36|
|  5|  E| 25|
+---+---+---+
// Dataframe with null values replaced
+---+----+
| id| obj|
+---+----+
|  1|-1.0|
|  1| 1.0|
|  8| 3.0|
|  2|-1.0|
|  3| 1.4|
|  3| 2.5|
| -1| 3.7|
+---+----+
// Dataframes which the rows have at least one null value
+----+----+
|  id| obj|
+----+----+
|   1|null|
|   2|null|
|null| 3.7|
+----+----+
+----+----+----+
|  id| obj| obj|
+----+----+----+
|   1|   A|null|
|   2|   B|null|
|null|   D|  15|
|   6|null|   7|
|   7|   G|null|
+----+----+----+