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

Question

我有这个java代码，其中火花UDF将行作为输入并返回行。还有一个广播变量，它是哈希映射。

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 而不逐个列出它们？我问这个问题的原因是实际的数据帧有超过50列。我看到了这个例子，但无法让它在Java中工作。

有没有办法在UDF中按名称访问行列？现在我正在使用字符串（int）。

UDF输出是一个名为myUDF的结构(struct(col1，col2，col3，myKey))。它变得很长，有50列。我怎么给它起别名呢？

任何帮助是值得赞赏的！

Answer 1

不需要提前知道列名！

您可以将行类型作为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映射col:_*））

然后，您可以访问udf中的dataframe行（结构和数据），以获取所需的任何信息，例如，将该行转换为（column_name）的映射-

val myUdf = udf（（行： Row） =

Answer 2

TL；DR使用数据集。映射（并将UDF替换为map函数）。

如何将dataframe中的所有列传递到UDF，而不一一列出它们？

dataframe.schema.fieldNames

请参见数据集API。

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

引用Row.fieldIndex的大话：

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

并使用索引。

它变得很长，有50列。我怎么给它起别名呢？

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