在每个时间列传递时迭代Dataframe以执行转换

Question

我有一个包含100列和col名称的数据表，如col1、col2、col3....我想根据条件匹配对列的值应用某些转换。我可以将列名存储在字符串数组中。并在withColumn中传递数组的每个元素的值，并基于When条件i可以垂直转换列的值。但问题是，由于Dataframe是不可变的，所以每个更新的版本都需要存储在一个新的变量中，并且新的Dataframe需要传入withColumn以进行下一次迭代转换。是否有任何方法来创建数组的dataframe，以便新的dataframe可以存储为数组的一个元素，并可以根据迭代器的值进行迭代。或者有没有其他方法来处理同样的事情。

var arr_df : Array[DataFrame] = new Array[DataFrame](60)   

-->这将引发错误“未找到类型dataframe”

val df(0) = df1.union(df2)

for(i <- 1 to 99){
  val df(i) = df(i-1).withColumn(col(i), when(col(i)> 0, col(i) + 
   1).otherwise(col(i)))

scala> val original_df = Seq((1,2,3,4),(2,3,4,5),(3,4,5,6),(4,5,6,7),(5,6,7,8),(6,7,8,9)).toDF("col1","col2","col3","col4")
original_df: org.apache.spark.sql.DataFrame = [col1: int, col2: int ... 2 more fields]

scala> original_df.show()

+----+----+----+----+
|col1|col2|col3|col4|
+----+----+----+----+
|   1|   2|   3|   4|
|   2|   3|   4|   5|
|   3|   4|   5|   6|
|   4|   5|   6|   7|
|   5|   6|   7|   8|
|   6|   7|   8|   9|
+----+----+----+----+

Answer 1

检查下面的代码。

scala> df.show(false)
+----+----+----+----+
|col1|col2|col3|col4|
+----+----+----+----+
|1   |2   |3   |4   |
|2   |3   |4   |5   |
|3   |4   |5   |6   |
|4   |5   |6   |7   |
|5   |6   |7   |8   |
|6   |7   |8   |9   |
+----+----+----+----+

scala>  val requiredColumns = df.columns.zipWithIndex.filter(_._2 < 3).map(_._1).toSet
requiredColumns: scala.collection.immutable.Set[String] = Set(col1, col2, col3)

scala> val allColumns = df.columns
allColumns: Array[String] = Array(col1, col2, col3, col4)

scala> val columnExpr = allColumns.filterNot(requiredColumns(_)).map(col(_)) ++ requiredColumns.map(c => when(col(c) > 3, col(c) + 1).otherwise(col(c)).as(c))

scala> df.select(columnExpr:_*).show(false)
+----+----+----+----+
|col1|col2|col3|col4|
+----+----+----+----+
|1   |2   |3   |4   |
|2   |3   |5   |5   |
|3   |5   |6   |6   |
|5   |6   |7   |7   |
|6   |7   |8   |8   |
|7   |8   |9   |9   |
+----+----+----+----+