Spark >= 2.0
您可以跳过 udf
,并使用 map
(Python 中的 create_map
)SQL 函数:
import org.apache.spark.sql.functions.map
df.select(
map(mapData.map(c => lit(c) :: col(c) :: Nil).flatten: _*).alias("a_map")
)
Spark < 2.0
据我所知,没有直接的方法可以完成此操作。您可以使用以下 UDF:
import org.apache.spark.sql.functions.{udf, array, lit, col}
val df = sc.parallelize(Seq(
(1L, "addr1", "addr2", "addr3")
)).toDF("user", "address1", "address2", "address3")
val asMap = udf((keys: Seq[String], values: Seq[String]) =>
keys.zip(values).filter{
case (k, null) => false
case _ => true
}.toMap)
val keys = array(mapData.map(lit): _*)
val values = array(mapData.map(col): _*)
val dfWithMap = df.withColumn("address", asMap(keys, values))
另一种不需要UDF的选择是使用结构体字段而不是映射:
val dfWithStruct = df.withColumn("address", struct(mapData.map(col): _*))
最大的优点是它可以轻松处理不同类型的值。