如何在不指定每一列的情况下将整行作为参数传递给 Spark（Java）中的 UDF？

Posted 2023-02-15

【中文标题】如何在不指定每一列的情况下将整行作为参数传递给 Spark（Java）中的 UDF？

【英文标题】：How to pass the whole row as an argument to an UDF in Spark (Java) without specifying every column?

【发布时间】：2017-10-22 06:17:40

【问题描述】：

我有这个 java 代码，其中 spark UDF 将 Row 作为输入并返回 Row。还有一个广播变量，它是一个 HashMap。

UDF 所做的只是检查广播 HashMap 是否包含 rowKey，如果包含，则返回一个新行，其中包含来自输​​入行的一些现有值和来自广播 HashMap 的一些更新值。如果不是，则按原样返回输入行。我这样做是因为我想根据 HashMap 中的值更新行列值。代码如下：

广播变量

final Broadcast<HashMap<String, HashMap<String, String>>> broadcastVariable = jsc.broadcast(someHashMap);

UDF 定义

UDF1<Row, Row> myUDF = new UDF1<Row, Row> () 
    @Override
    public Row call(Row inputRow) 

        String myKey = inputRow.getString(3);

        if (broadcastVariable.value().containsKey(myKey))
            Map<String, String> valuesToUpdate = broadcastVariable.value().get(myKey);

            String col1 = inputRow.getString(0);
            String col2 = inputRow.getString(1);
            String col3 = inputRow.getString(2);

            for (Map.Entry<String, String> entry : valuesToUpdate.entrySet())
            
                String columnName = entry.getKey();

                switch(columnName) 
                case "col1" :
                    col1 = entry.getValue();
                    break;
                case "col2" :
                    col2 = entry.getValue();
                    break;
                case "col3" :
                    col3 = entry.getValue();
                    break;
                
            
            return RowFactory.create(col1,col2,col3,myKey);

        
        return inputRow;
    
;

UDF 注册

hiveContext.udf().register("myUDF", myUDF, DataTypes.createStructType(DF1.schema().fields()));

UDF 调用

DataFrame DF2 = DF1.select(org.apache.spark.sql.functions.callUDF
                ("myUDF", org.apache.spark.sql.functions.struct(DF1.col("col1"),
                        DF1.col("col2"),
                        DF1.col("col3"),
                        DF1.col("myKey"))));

我有以下问题，

如何在不一一列出的情况下将数据框中的所有列传递给 UDF？我问这个的原因是实际的 DataFrame 有超过 50 列。我看到了这个 example，但无法让它在 Java 中工作。

有没有一种方法可以在 UDF 中按名称访问行列？现在我正在使用 getString(int)。

UDF 输出是一个名为 myUDF(struct(col1,col2,col3,myKey)) 的结构。超过 50 列会变得很长。我怎样才能给它起别名？

感谢任何帮助！

【问题讨论】：

How to pass whole Row to UDF - Spark DataFrame filter的可能重复

@agsachin 这里的 OP 要求用 Java 提供解决方案，而链接的线程在 Scala 中，而不是完全重复的。

【参考方案1】：

您无需提前知道列名！

您可以将 Row 类型作为 udf 的参数之一。例如：

import org.apache.spark.sql.functions._

val myUdf = udf((row: Row) => <here comes the code inside your udf>)

你这样称呼那个udf：

df.withColumn(newColumnName, myUdf(struct(df.columns map col: _*)))

然后您可以访问 udf 中的数据框行（结构和数据）以获取所需的任何内容，例如 - 将该行转换为 (column_name -> column_value) 的映射：

val myUdf = udf((row: Row) => row.getValuesMap(row.schema.fieldNames))

【讨论】：

OP 要求 Java 解决方案。这是在 Scala 中。

【参考方案2】：

TL;DR 使用 Dataset.map（并将 UDF 替换为 map 函数）。

---

如何将数据框中的所有列传递给 UDF，而不一一列出？

dataframe.schema.fieldNames

请参阅Dataset API。

有没有办法可以在 UDF 中按名称访问行列？

引用Row.fieldIndex的scaladoc：

fieldIndex(name: String): Int 返回给定字段名称的索引。

并使用索引。

超过 50 列会变得很长。我怎样才能给它起别名？

看起来您的代码将受益于一些重构和组合。在单个管道中处理 50 个字段可能会有点笨拙。

【讨论】：

谢谢！我使用的是 Spark 1.6.0，所以对使用 DataSet 有点犹豫，认为它可能会导致与最新版本的一些兼容性问题。我会试一试。您提到“在单个管道中处理 50 个字段可能会有点笨拙”，您认为 DataSet 方法可以解决这个问题吗？

我几乎不相信任何查询都应该使用 50 个字段（所以它不是关于 DataFrame/Dataset，而是一次使用这么多字段）。创建子数据集并加入它们，让程序员的生活更轻松。