在Apache Spark中使用UDF

在Apache Spark中使用UDF

用户自定义函数（UDF）是大多数SQL环境的一个关键特性，其主要用于扩展系统的内置功能。UDF允许开发人员通过抽象其低级语言实现在更高级语言（如SQL）中应用的新函数。Apache Spark也不例外，其为UDF与Spark SQL工作流集成提供了各种选项。

在本篇博文中，我们将回顾Python、Java和Scala上的Apache Spark UDF和UDAF（用户自定义的聚合函数）实现的简单示例。我们还将讨论重要的UDF API功能和集成点，包括各发行版本之间的当前可用性。总而言之，我们将介绍一些重要的性能注意事项，使您对应用程序中利用UDF的选择有所了解。

Spark SQL UDFs

UDF转换了表中单个行的数值，为每行生成单个对应的输出值。例如，大多数的SQL环境都提供了一个UPPER函数，同时返回了一个大写版本的字符串以作为输入。

Spark SQL UDAF函数

或者，通过包含实现jar文件（使用含有spark-submit的-jars选项），以Scala和Java语言实现的UDF可以从PySpark中进行访问，然后通过SparkContext对象的私有引用执行器JVM、底层Scala或装载在jar文件中的Java UDF实现来访问UDF定义。Holden Karau在一次精彩的演讲中[2]对这种方法进行了探讨。请注意，在此技术中所使用的一些Apache Spark私有变量不是正式面向终端用户的。这样做还带来了额外的好处，允许将UDAF（目前必须在Java和Scala中定义）用于PySpark，下文中的示例中使用了前面在Scala中定义的SUMPRODUCT UDAF进行证明：

性能注意事项

了解Apache Spark UDF功能的性能影响是非常重要的。例如，Python UDF（比如我们的CTOF函数）导致数据在运行UDF逻辑的执行器JVM和Python注释器之间被序列化 - 与Java或Scala中的UDF实现相比，这大大降低了性能。缓解这种序列化瓶颈的潜在解决方案包括以下方面：

如上一节所述，从PySpark中访问Hive UDF。Java UDF实现可以由执行器JVM直接访问。请再次注意，这种方法只用于从Apache Spark的SQL查询语言访问UDF。 这种方法的使用也可以参考PySpark访问在Java或Scala中执行的UDF，如我们之前定义的Scala UDAF示例所示。

一般来说，UDF逻辑应尽可能的精简，因为可能每一行都会被调用。例如，在扩展到10亿行时，UDF逻辑中的一个步骤需要耗费100毫秒的时间才能完成，从而很快就会导致重大的要性能问题。

Spark SQL的另一个重要组成部分是Catalyst查询优化器。这一功能随着每个版本而扩展，通常可以为Spark SQL查询提供显著的性能改进；然而，任意UDF实现代码对于Catalyst而言可能不是很好理解（虽然分析字节码的未来功能[3]被认为可以解决这一问题）。因此，使用Apache Spark内置SQL查询函数功能通常可以带来最佳性能，并且应该是在避免引入UDF时考虑的第一种方法。高级用户寻求利用Catalyst与其代码更紧密地结合，可以参考以下Chris Fregly的演讲[4]，该演讲人使用Expression.genCode优化UDF代码，并且使用了新的Apache Spark 2.0实验功能[5]，其为定制Catalyst优化程序规则提供了一个可即插即用的API。

结论

当Spark SQL的内置功能需要扩展时，UDF是一个非常有用的工具。本篇博文中提供了一次UDF和UDAF实现的演练，并讨论了其集成步骤，以在Spark SQL中利用Spark SQL中现有的Java Hive UDF。UDF可以在Python、Scala、Java和（在Spark 2.0中）R中实现，同时UDAF 可以在以及Scala和Java的UDAF中实现。当在PySpark中使用UDF时，必须考虑数据序列化成本，并且应该考虑采用上文所讨论的两个策略来解决这个问题。最后，我们探讨了Spark SQL的Catalyst优化器，以及基于性能考虑的因素，在解决方案中引入UDF之前坚持使用内置SQL函数的性能考虑因素。

代码Spark 1.6.0）

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。