本文主要介绍“什么是火花UDF变长参数法”。在日常操作中,相信很多人对Spark UDF变长参数的方法有所怀疑。边肖查阅了各种资料,整理出简单易用的操作方法,希望能帮你解答“火花UDF变长参数的方法是什么”的疑惑!接下来,请和边肖一起学习!
引子
变量参数对我们来说并不陌生,这是我们用Java写的。
Public void varargs(字符串.这是我们用Scala写的。
defargs(cols : string *): string在Spark中,很多时候我们都有自己的业务逻辑,现成的功能无法满足我们的需求。当我们需要处理同一行中的多列,并通过自己的逻辑将它们组合成一列时,可变长度参数及其变体的实现可以帮助我们。
但是,在Spark UDF中,我们不能使用变长参数来传输值,但是本文从变长参数开始的原因是需求从它开始,通过转换它,我们可以使用变长参数或Seq类型来接收参数。
以下演示由火花壳公司完成。以下三种方法可以实现多列参数传输,它们是
可变参数(接受数组类型)
Seq类型参数(接受数组类型)
行类型参数(接受结构类型)
变长参数类型的UDF
定义UDF方法
defmyncatvarargs(sep : string,cols 3360 string *): string=cols . filter(_!=null)。注册UDF函数
因为可变长度参数只能由方法定义,所以这里使用一些应用函数来转换它们。
你可以看到udf的定义如下
Userdefinedfunction(函数2,字符串类型,list(字符串类型,ArrayType(字符串类型,true)),即变长参数转换为数组类型,函数只包含两个参数,所以变长参数列表也可以看作是不可用的。
变长参数列表传值
我们构建一个数据框架如下
Valdf=sc。并行(数组((' aa ',' bb ',' cc '),(' DD ',' ee ',' ff '))。todf ('a ',' b ',' c ')然后直接在多个String中传递给myConcatVarargsUDF。
DF。选择(myncatvarargsudf(lit('-'))、col ('a ')、col ('b ')、col ('c ')。显示结果中出现以下错误
Java . lang . class castexception : anon fun $ 1 cannotbecasttoscala . function 4可以看出,使用变长参数表是不支持Spark的,它会被识别为四个参数的函数,而UDF定义为两个参数的函数而不是四个参数!
变换:使用array()转换做第二个参数
我们使用Spark提供的Array()函数将参数转换为数组类型。
Df。选择(myconcat Varargsudf (lit ('-'))、数组(col ('a ')、col ('b ')、col ('c '))。显示结果如下
- |UDF(-,数组(A,B,C))| - |n
bsp; aa-bb-cc| | dd-ee-ff| +-------------------+
由此可以看出,使用变长参数构造的UDF方法,可以通过构造Array的方式传参,来达到多列合并的目的。
使用Seq类型参数的UDF
上面提到,变长参数***被转为ArrayType,那不禁要想我们为嘛不使用Array或List类型呢?
实际上在UDF里,类型并不是我们可以随意定义的,比如使用List和Array就是不行的,我们自己定义的类型也是不行的,因为这涉及到数据的序列化和反序列化。
以Array/List为示例的错误
下面以Array类型为示例
定义函数
val myConcatArray = (cols: Array[String], sep: String) => cols.filter(_ != null).mkString(sep)
注册UDF
val myConcatArrayUDF = udf(myConcatArray)
可以看到给出的UDF签名是
UserDefinedFunction(<function2>,StringType,List())
应用UDF
df.select(myConcatArrayUDF(array(col("A"), col("B"), col("C")), lit("-"))).show
会发现报错
scala.collection.mutable.WrappedArray$ofRef cannot be cast to [Ljava.lang.String
同样List作为参数类型也会报错,因为反序列化的时候无法构建对象,所以List和Array是无法直接作为UDF的参数类型的
以Seq做参数类型
定义调用如下
val myConcatSeq = (cols: Seq[Any], sep: String) => cols.filter(_ != null).mkString(sep) val myConcatSeqUDF = udf(myConcatSeq) df.select(myConcatSeqUDF(array(col("A"), col("B"), col("C")), lit("-"))).show
结果如下
+-------------------+ |UDF(array(A,B,C),-)| +-------------------+ | aa-bb-cc| | dd-ee-ff| +-------------------+
使用Row类型参数的UDF
我们可以使用Spark functions里struct方法构造结构体类型传参,然后用Row类型接UDF的参数,以达到多列传值的目的。
def myConcatRow: ((Row, String) => String) = (row, sep) => row.toSeq.filter(_ != null).mkString(sep) val myConcatRowUDF = udf(myConcatRow) df.select(myConcatRowUDF(struct(col("A"), col("B"), col("C")), lit("-"))).show
可以看到UDF的签名如下
UserDefinedFunction(<function2>,StringType,List())
结果如下
+--------------------+ |UDF(struct(A,B,C),-)| +--------------------+ | aa-bb-cc| | dd-ee-ff| +--------------------+
使用Row类型还可以使用模式提取,用起来会更方便
row match { case Row(aa:String, bb:Int) => }
***
对于上面三种方法,变长参数和Seq类型参数都需要array的函数包装为ArrayType,而使用Row类型的话,则需要struct函数构建结构体类型,其实都是为了数据的序列化和反序列化。三种方法中,Row的方式更灵活可靠,而且支持不同类型并且可以明确使用模式提取,用起来相当方便。
而由此我们也可以看出,UDF不支持List和Array类型的参数,同时 自定义参数类型 如果没有混合Spark的特质实现序列化和反序列化,那么在UDF里也是 无法用作参数类型 的。当然,Seq类型是可以 的,可以接多列的数组传值。
此外,我们也可以使用柯里化来达到多列传参的目的,只是不同参数个数需要定义不同的UDF了。
到此,关于“Spark UDF变长参数的方法是什么”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注网站,小编会继续努力为大家带来更多实用的文章!
内容来源网络,如有侵权,联系删除,本文地址:https://www.230890.com/zhan/145790.html