如何使用 spark UDF 返回复杂类型答案

【问题标题】：How to return complex types using spark UDFs如何使用 spark UDF 返回复杂类型
【发布时间】：2017-02-06 14:42:42
【问题描述】：

您好，提前谢谢您。

我的程序是用 java 编写的，我无法迁移到 scala。

我目前正在使用以下行从 json 文件中提取 spark DataFrame：

DataFrame dff = sqlContext.read().json("filePath.son");

SQLContext 和 SparkContext 已正确初始化并完美运行。

问题是我正在读取的 json 具有嵌套结构，我想清理/验证内部数据，而不更改架构。

其中一个数据框的列特别具有“GenericRowWithSchema”类型。

假设我想清理名为“数据”的唯一列。

我想到的解决方案是定义一个名为“cleanDataField”的用户定义函数 (UDF)，然后在“data”列上运行它。代码如下：

UDF1<GenericRowWithSchema,GenericRowWithSchema> cleanDataField = new UDF1<GenericRowWithSchema, GenericRowWithSchema>(){

        public GenericRowWithSchema call( GenericRowWithSchema grws){

            cleanGenericRowWithSchema(grws);

            return grws;

        }
    };

然后我会在 SQLContext 中注册函数：

sqlContext.udf().register("cleanDataField", cleanDataField, DataTypes.StringType);

然后我会打电话给

df.selectExpr("cleanDataField(data)").show(10, false);

为了查看包含干净数据的前 10 行。

最后，问题是这样的：我可以返回复杂的数据（例如自定义类对象）吗？如果可能的话，我该怎么做？我想我必须更改 udf 注册的第三个参数，因为我没有返回字符串，但我应该用什么替换它？

谢谢

【问题讨论】：

标签： java json apache-spark user-defined-functions udf

【解决方案1】：

假设您要构造一个数据类型为struct<companyid:string,loyaltynum:int,totalprice:int,itemcount:int>

为此，您可以执行以下操作：

    // I am just copying the json string as is but you will need to escape it properly for java.

DataType dt = DataType.fromJson({"type":"struct","fields":[{"name":"companyid","type":"string","nullable":false,"metadata":{}},{"name":"loyaltynum","type":"integer","nullable":false,"metadata":{}},{"name":"totalprice","type":"integer","nullable":false,"metadata":{}},{"name":"itemcount","type":"integer","nullable":false,"metadata":{}}]})

然后，您可以在注册 UDF 时将该数据类型用作返回类型。

【讨论】：

【解决方案2】：

我不知道你的问题是否仍然有效，但以防万一，这里是答案：

您需要将第三个参数替换为Encoders.bean(GenericRowWithSchema).schema()

【讨论】：