Spark 2.2 Scala DataFrame 从字符串数组中选择，捕获错误

我是 SparkSQL/Scala 的新手，正在努力完成一些看似简单的任务。

我正在尝试从 Scala 字符串数组构建一些动态 SQL。我正在尝试在 DataFrame 中重新输入一些列，但直到运行时我才能看到 DataFrame 中的列集时，我才会确切地知道需要重新输入哪些列。所以我正在尝试这样做：

val cols = df.columns
val typedCols = cols.map( c => getTypedColumn(c) )
df.select( ...)  or df.selectExpr(...) // how to invoke this with vals from my string array??

typedCols 最终将成为一个字符串数组，其值如下：

["a", "cast(b as int) b", "c"]

我是否需要首先从该数组创建一个大的逗号分隔字符串？

因此，假设这可行，我将调用该 select 语句，它会将我的 DataFrame 转换为具有我所需类型的新 DataFrame。但是 DataFrame 中的某些记录将出现错误，并且尝试重新输入将失败。

如何获得包含通过输入的所有良好记录的 DataFrame 结果，然后将所有不良记录放入某种错误存储桶中？在尝试 DataFrame 选择之前，我是否需要先进行验证？

您可以只使用可变参数：

val df = Seq(("a", "1", "c"), ("foo", "bar", "baz")).toDF("a", "b", "c")
val typedCols = Array("a", "cast(b as int) b", "c")
df.selectExpr(typedCols: _*).show

+---+----+---+
|  a|   b|  c|
+---+----+---+
|  a|   1|  c|
|foo|null|baz|
+---+----+---+

但我个人更喜欢专栏：

val typedCols = Array($"a", $"b" cast "int", $"c")
df.select(typedCols: _*).show

如何获得包含通过输入的所有良好记录的 DataFrame 结果，然后将所有不良记录放入某种错误存储桶中？

失败的数据cast is NULL。要找到好的记录，请使用na.drop:

val result = df.selectExpr(typedCols: _*)
val good = result.na.drop()

查找空头支票是否有NULL

import org.apache.spark.sql.functions.col

val bad = result.where(result.columns.map(col(_).isNull).reduce(_ || _))

要获取不匹配的数据：

• If typedCols are Seq[Column] you can

  df.where(typedCols.map(_.isNull).reduce(_ || _))  
  

• If typedCols are Seq[String]你可以：

  import org.apache.spark.sql.functions.expr
  
  df.where(typedCols.map(expr(_).isNull).reduce(_ || _))