spark structstream 性能(sparkudf使用缓存数据吗)

本文主要介绍“什么是火花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变长参数的方法是什么”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注网站，小编会继续努力为大家带来更多实用的文章！