Spark DataFrame - 区分缺少列的记录与坏值

Posted 2021-04-18

这个问题与这个主题有关：Spark 2.2 Scala DataFrame select from string array, catching errors

我需要区分缺少列的记录（在我的用例中不是错误）和具有不适用于列类型的垃圾值的记录。

在我执行selectExpr之后，这两种情况在生成的DataFrame中显示为null。我正在寻找一种快速的方法来包含缺少列​​的记录以及良好的结果，同时将具有垃圾值的记录丢弃到坏桶中。坏的将包括像空字符串作为值的int字段，或“abc”。

例如，假设我有这样的DataFrame：Col A - string，Col B - int，Col C - string，

A    B    C
"x", "",  ""  - Error, bad value for B
"",  null,""  - Good, missing value for B
"",  "a", ""  - Bad, bad value for B
"x", "1", "x" - Good, normal case

-----编辑-----

显示创建数据帧的代码。数据以json形式出现，所有字段都被引用，因此它最初认为一切都是字符串。我需要输入几个字段为int，boolean等。有关详细信息，请参阅顶部的链接。

 val cols = dfLower.columns
 val typedCols = cols.map( c => getTypeStmtForCol(c, qradarType) )
 val result = dfLower.selectExpr(typedCols: _*)

 // This puts both records with missing columns and bad values in bad.
 // Need way to distinguish between those 2 cases.     
 val bad = dfLower.where(typedCols.map(expr(_).isNull).reduce(_ || _))
 val good = result.na.drop()

----编辑2 ----

我想我可能有个主意。如果我可以计算每个记录之前和之后的空值数，那么只有那些在select之后具有更多空值的记录才会出错。不确定如何实现......

答案

快速和脏的类型，但创建一个测试您的条件的udf并根据条件的结果返回状态。

def checkIntData=udf((columnData: String) => {
  var status = "GOOD"
  try{
    columnData.toInt
  } catch {
    case ex: Exception => {
      if(columnData == null) {
        // Do nothing. This is fine
      } else if(columnData.length == 0) {
        status = "ERROR"
      } else {
        status = "BAD"
      }
    }
  }
  status
})

val seqData = Seq(("x","","","0"),("",null,"","3"),("","a","","z"),("x","1","x",""))
val df = seqData.toDF("col1","col2","col3","col4")
val colsToCheck = df.select("col2","col4").columns
var newdf = df
// Iterate over the columns you want to check inside the dataframe. Each checked column will add a new status column to newdf
colsToCheck.map{column =>
  newdf = newdf.withColumn(column+"Status", checkIntData(newdf(column)))
}
newdf.show()

这将返回以下内容：

+----+----+----+----+----------+----------+
|col1|col2|col3|col4|col2Status|col4Status|
+----+----+----+----+----------+----------+
|   x|    |    |   0|     ERROR|      GOOD|
|    |null|    |   3|      GOOD|      GOOD|
|    |   a|    |    |       BAD|     ERROR|
|   x|   1|   x|   z|      GOOD|       BAD|
+----+----+----+----+----------+----------+

然后，您可以通过基于状态列进行过滤来创建错误存储桶。

第1列到第3列来自您的示例。我添加了第4列，以展示如何将其应用于多个列而无需编写.withColumn()一百次。我通过创建列colsToCheck数组然后迭代以将udf应用于所有选定列来实现此目的。

注意！因为我可能会因为这样做而大喊大叫，所以我想让你知道使用try / catch作为流控制被认为是反模式（又称糟糕的编程）。 Read more to find out why.