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

Question

您好,并提前感谢您.

我的程序是用java编写的,我无法移动到scala.

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

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

SQLContext和SparkContext被正确初始化并完美运行.

问题是我正在读取的json有嵌套结构,我想清理/验证内部数据,而不更改模式.

其中一个数据帧的列具有"GenericRowWithSchema"类型.

假设我想清理那个名为"data"的列.

我想到的解决方案是定义名为"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注册的第3个参数,因为我没有返回一个字符串,但我应该替换它？

谢谢

Answer 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 时使用该数据类型作为返回类型。