java - 将数据集中的映射类型映射到列

Question

我在 Spark 中有一个 UDF，它返回 Map 输出。

Dataset<Row> dataSet = sql.sql("select *, address(col1,col2) as udfoutput from input");

我想将映射中返回的值附加到列中。

例如 - 如果输入表有 2 列并且 UDF 映射返回 2 个键值对，则应使用数据集创建总共 4 列。

Answer 1

怎么样

select 
    *, 
    address(col1,col2)['key1'] as key1,
    address(col1,col2)['key2'] as key2
    from input

或者使用 with 仅调用您的 UDF 一次:

with
    raw as (select *, address(col1,col2) address from input)
select 
    *, 
    address['key1'], 
    address['key2'] 
from raw

这就是 hive 方式。

在 Spark 中，您可以通过 Dataset API 使用所有命令式转换(而不是声明式 SQL)。在 Scala 中它可能看起来像这样。我相信，在 Java 中，它有点冗长:

// First your schemas as case classes (POJOs)
case class MyModelClass(col1: String, col2: String)
case class MyModelClassWithAddress(col1: String, col2: String, address: Map[String, String])

// in spark any function is a udf    
def address(col1: String, col2: String): Map[String, String] = ??? 

// Now imperative Spark code 
import spark.implicits._
val dataSet: Dataset[Row] = ??? // you can read table from Hive Metastore, or using spark.read ...
dataSet
  .as[MyModelClass]
  .map(myModel => MyModelWithAddress(myModel.col1, myModel.col1, address(myModel.col1, myModel.col2))
  .save(...) //wherever needs to be done later