Scala&Spark:一次投射多个列

Question

由于VectorAssembler崩溃,如果传递的列具有除NumericTypeor 之外的任何其他类型BooleanType并且我正在处理许多TimestampType列,我想知道:

有一种简单的方法,一次投下多个列吗？

根据这个答案,我已经有了一个方便的方法来构建一个列:

def castColumnTo(df: DataFrame, 
    columnName: String, 
    targetType: DataType ) : DataFrame = {
      df.withColumn( columnName, df(columnName).cast(targetType) )
}

我想过castColumnTo递归调用,但我强烈怀疑这是(高效)的方式.

Answer 1

在scala中使用惯用法绘制所有列

def castAllTypedColumnsTo(df: DataFrame, sourceType: DataType, targetType: DataType) = {
df.schema.filter(_.dataType == sourceType).foldLeft(df) {
    case (acc, col) => acc.withColumn(col.name, df(col.name).cast(targetType))
 }
}

Answer 2

根据评论(谢谢!)我想出了以下代码(没有实现错误处理):

def castAllTypedColumnsTo(df: DataFrame, 
   sourceType: DataType, targetType: DataType) : DataFrame = {

      val columnsToBeCasted = df.schema
         .filter(s => s.dataType == sourceType)

      //if(columnsToBeCasted.length > 0) {
      //   println(s"Found ${columnsToBeCasted.length} columns " +
      //      s"(${columnsToBeCasted.map(s => s.name).mkString(",")})" +
      //      s" - casting to ${targetType.typeName.capitalize}Type")
      //}

      columnsToBeCasted.foldLeft(df){(foldedDf, col) => 
         castColumnTo(foldedDf, col.name, LongType)}
}

感谢鼓舞人心的评论.foldLeft(在此处和此处解释)保存for循环以迭代var数据帧.