将 Scala 可变数组转换为 spark 数据框

Question

我定义了三个可变数组：

import scala.collection.mutable.ArrayBuffer
var quartile_1 = ArrayBuffer[Double]()
var quartile_3 = ArrayBuffer[Double]()
var id = ArrayBuffer[String]()

quartile_1 和 quartile_3 是 id 级别的信息，我目前将它们计算为：

def func1(x: org.apache.spark.sql.Row) {
  val apQuantile = df_auth_for_qnt.where($"id" === x(0).toString).stat.approxQuantile("tran_amt", Array(0.25, 0.75), 0.001)
  quartile_1 += apQuantile(0)
  quartile_3 += apQuantile(1)
  id += x(0).toString()
}

val cardNumList = df_auth_for_qnt_gb.where($"tran_cnt" > 8).select("card_num_1").collect.foreach(func1)

有没有比将它们附加到可变数组更好的方法？我的目标是将分位数数据 id 作为数据框提供 - 这样我就可以进行进一步的连接。

Answer 1

像 ArrayBuffer 这样的可变结构是邪恶的，尤其是在可并行化的上下文中。在这里可以很容易地避免它们。

func1可以return一个(String, Array[Double])的元组，其中第一个元素对应id（以前的id缓冲区），第二个元素是四分位数return来自 approxQuantile:

def func1(x: Row): (String, Array[Double]) = {
  val cardNum1 = x(0).toString
  val quartiles = df_auth_for_qnt.where($"id" === cardNum1).stat.approxQuantile("tran_amt", Array(0.25, 0.75), 0.001)
  (cardNum1, quartiles)
}

现在，使用函数转换我们可以获得一个不可变的结果结构。

作为数据框：

val resultDf = df_auth_for_qnt_gb.where($"tran_cnt" > 8).select("card_num_1").map(func1).toDF("id", "quartiles")

或者作为 Map[String, Array[Double]] 具有与元组相同的关联 return 来自 func1:

val resultMap = df_auth_for_qnt_gb.where($"tran_cnt" > 8).select("card_num_1").map(func1).collect().toMap

将 Scala 可变数组转换为 spark 数据框

Converting Scala mutable arrays to a spark dataframe

scala

arraybuffer

spark-dataframe