带 3 个参数的 zip 函数

Posted 2023-03-23

【中文标题】带 3 个参数的 zip 函数

【英文标题】：zip function with 3 parameter

【发布时间】：2019-07-08 09:58:09

【问题描述】：

我想转置 Spark SQL 表中的多个列 我发现这个解决方案只有两列，我想知道如何使用三列的 zip 函数varA, varB and varC.

import org.apache.spark.sql.functions.udf, explode

val zip = udf((xs: Seq[Long], ys: Seq[Long]) => xs.zip(ys))

df.withColumn("vars", explode(zip($"varA", $"varB"))).select(
   $"userId", $"someString",
   $"vars._1".alias("varA"), $"vars._2".alias("varB")).show

这是我的数据框架构：

`root
 |-- owningcustomerid: string (nullable = true)
 |-- event_stoptime: string (nullable = true)
 |-- balancename: string (nullable = false)
 |-- chargedvalue: string (nullable = false)
 |-- newbalance: string (nullable = false)
`

我试过这段代码：

    val zip = udf((xs: Seq[String], ys: Seq[String], zs: Seq[String]) => (xs, ys, zs).zipped.toSeq)

df.printSchema

val df4=df.withColumn("vars", explode(zip($"balancename", $"chargedvalue",$"newbalance"))).select(
   $"owningcustomerid", $"event_stoptime",
   $"vars._1".alias("balancename"), $"vars._2".alias("chargedvalue"),$"vars._2".alias("newbalance"))

我收到了这个错误：

cannot resolve 'UDF(balancename, chargedvalue, newbalance)' due to data type mismatch: argument 1 requires array<string> type, however, '`balancename`' is of string type. argument 2 requires array<string> type, however, '`chargedvalue`' is of string type. argument 3 requires array<string> type, however, '`newbalance`' is of string type.;;

'项目 [owningcustomerid#1085，event_stoptime#1086，balancename#1159，chargedvalue#1160，newbalance#1161，explode(UDF(balancename#1159，chargedvalue#1160，newbalance#1161))AS vars#1167]

【参考方案1】：

在一般的 Scala 中，您可以使用 Tuple3.zipped

val zip = udf((xs: Seq[Long], ys: Seq[Long], zs: Seq[Long]) => 
  (xs, ys, zs).zipped.toSeq)

zip($"varA", $"varB", $"varC")

特别是在 Spark SQL (>= 2.4) 中，您可以使用arrays_zip 函数：

import org.apache.spark.sql.functions.arrays_zip

arrays_zip($"varA", $"varB", $"varC")

但是您必须注意，您的数据不包含 array<string>，而是纯 strings - 因此 Spark arrays_zip 或爆炸是不允许的，您应该先解析您的数据。

【讨论】：

架构清楚地表明您没有数组列 - 在这种情况下，压缩没有意义。

【参考方案2】：

val zip = udf((a: Seq[String], b: Seq[String], c: Seq[String], d: Seq[String]) => a.indices.map(i=> (a(i), b(i), c(i), d(i))))