如何将 Scala 函数应用于两个不同的列数据框 Scala

Question

我目前正在用 Scala 编写一个脚本，为此，我预先编写了一个如下所示的函数：

def reducer_scala(node: (Int, List[Int])): (List[(Int, Int)], Int) = {
var counter = 0
var minimum = node._1
var value_List = List.empty[Int]
var joined = List.empty[(Int, Int)]

for(value <- node._2){
    if(value < minimum){
      minimum = value
    }
    value_List :+= value
  }

if(minimum<node._1){
    joined = joined :+ (node._1, minimum)
    for(value <- value_List){
  
      if(minimum != value){
        counter = counter + 1
        joined = joined :+ (value, minimum)
      }
    }
  }
  return (joined, counter)
}

如您所见，它将一个 Int 和一个 List[Int] 的元组作为输入。

我现在想将它应用到我的 Dataframe 中的两列，如下所示：

  _1,      collect_list(_2)

   0|             [1]|
|  1|       [2, 3, 0]|
|  3|          [4, 1]|
|  5|             [6]|
|  6|          [7, 5]|
|  2|             [1]|
|  4|             [3]|
|   7|             [6]

如您所见，第一列是 Int，第二列是 List。

我想知道： -如何转换在同一个元组中重新组合两列的每个元素 -如何将我的 reducer_scala 函数应用到每个元素（元组（element_column_1，element_column_2））

最好的问候

Answer 1

select() 函数使您可以将 udf 应用于数据框的每一行，使用 col("colname") 选择所需的列 你只需要稍微调整一下你的函数，让它接受两个单独的参数而不是一个元组。

确保您的函数已注册为 udf，例如：

val reducer = udf {(x: Int, y: List[Int]) => reducer_scala(x, y)}

然后用df = df.select(reducer(col("colname1"), col("colname2"))) 调用它

希望这会有所帮助！