UDF（PYTHON / JAVA）入门级开发

开发前的声明

udf开发是在数据分析的时候如果内置的函数解析不了的情况下去做的开发，比方说你只想拆分一个字段，拼接一个字段之类的，就不要去搞udf了，这种基本的需求自带函数完全支持，具体参数可参考文档：

spark-sql内置函数

里面列举了所有函数的介绍和使用，推荐优先使用官方推出的，因为自己写的udf如果对一些异常处理不到位可能会导致数据分析的失败或者意想不到的情况。但是内置的函数真的满足不了那就真的没办法了，比如在做数据分析的时候经常使用的是ip解析，这个通常就需要自己开发udf了。

优质UDF开发示例：

（1）、节假日日期筛选 （2）、UDF开发流程

常见错误：

（1）、Error : java 不支持发行版本X

pom文件查询配置： jar包配置搜索官方地址：http://mvnrepository.com/

udf的分类

udf平常用的多的也就两种，一种java写的，一种python 所写，下面对这两种情况分别做个简单的demo供大家参考。

udf for java（idea）

1.创建maven工程

file->project structure->modules->点击+号->new module->选择maven

点击next，填写groupid（对应包结构）、artifactid（maven仓库对应的坐标）

source java 代码，操作如下图file->project structure，

点击apply或者ok，

2.开始写java代码

pom文件添加依赖

    <dependencies><!--添加Hive依赖--><dependency><groupId>org.apache.hive</groupId><artifactId>hive-exec</artifactId><version>2.0.0</version></dependency><!--添加Hadoop依赖--><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>2.6.5</version></dependency><dependency><groupId>jdk.tools</groupId><artifactId>jdk.tools</artifactId><version>1.6</version><scope>system</scope></dependency></dependencies>

java - main主函数：

package com.songheng.UDF;import org.apache.hadoop.hive.ql.exec.UDF;
import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;/*** 1、继承 UDF 接口*/
public class URLDecode extends UDF {/*** 传参有一些特殊字符，比如%号或者说+号，导致不能解析 所以要进行转换* @param parameter* @return*/public String recursionParser(String parameter) {String decodeParameter = "";try {decodeParameter = URLDecoder.decode(parameter.replaceAll("%(?![0-9a-fA-F]{2})", "%25").replaceAll("\\+", "%2B"), "utf-8");} catch (UnsupportedEncodingException e) {return parameter;}return decodeParameter;}/*** 2、重写evaluate方法*/public String evaluate(String input) {return recursionParser(input);}public static void main(String[] args){URLDecode udf = new URLDecode();String result = udf.evaluate("%E4%B8%AD%E5%9B%BD");System.out.println(result);}
}

3.生成jar包上传到hdfs （参考文章）

在此项目pom文件的路径下执行mvn clean install

将target文件中生成的jar文件上传到hdfs上，路径自己自定义，我直接上传到/。

sudo -u hdfs hdfs dfs -put UrlDecodeUDF.jar /user/olicity/hive/UDF/

4.使用hivesql或者sparksql加载自定义函数

1、添加 jar 包：
add jar /user/olicity/hive/UDF/UrlDecodeUDF.jar;      2、查看导入的 jar 包：
list jars;     3、创建一个名为 myHiveUrlDecodeUDF 的临时函数，关联该jar包：
create temporary function myHiveUrlDecodeUDF as 'com.songheng.UDF.URLDecode';4、查看创建的函数：
show functions;-----------------------------
注意：上述方法仅限于当前会话生效，如需要添加一个永久的函数对应的永久的路径，则需要将jar包放到hdfs上，然后创建函数关联路径即可：
create function locUDF.myHiveUrlDecodeUDF as 'com.songheng.UDF.URLDecode' using jar 'hdfs://hdfs路径/UrlDecodeUDF.jar';use myHiveUrlDecodeUDF;-----------------------------
另外还看到过另一种方法，配置hive-site.xml文件中的hive.aux.jars.path
配置参考如下：<property><name>hive.aux.jars.path</name><value>file:///home/hdfs/fangjs/DefTextInputFormat.jar,file:///jarpath/test.jar</value></property>

【测试sql】：select myHiveUrlDecodeUDF("%E4%B8%AD%E5%9B%BD") as word;>word
中国

udf for python

python udf入门我就不重新阐述了，我看到有一位同学的博客写的很清晰，推荐博客

https://blog.csdn.net/qq_26937525/article/details/54136317

存在的问题：

在数据清洗过程中，因为Python是直接向系统申请资源的，而不是像JAVA是向yarn的ResourceManager申请资源，如果使用的是TransForm而不是UDF的话，会导致启动的Python脚本对内存和CPU的使用不可控，尤其是当启动多个Map时，因为一个map将启动一个Python因此，当同时运行的map有几十个时（测试集群较小），同时将尝试启动相同个数的python（资源够用的话仍然会启动几十个），且此时Map占用的内存是不会释放掉的他在一直等待Python的结果，这将导致python可用的资源仅仅是原本分配给系统的很少的资源（注：在安装Hadoop时，对于单个节点，一般仅仅给系统留出很少的内存，其他的内存全部分给了集群。例如32G物理内存的节点给系统和dataNode+nodeManager的内存就4-8个G，同时CPU核数也不足节点的一半，剩余的内存和cpu核数全部划分给集群使用。需要注意的是，这里虽然说是划分给集群使用，仅仅是逻辑上的划分，即规定集群可以使用的最大的物理内存，超过该内存时MR可以认为是不会抢占分配给系统+DataNode+nodeManager的内存的，但是当集群中没有MR在执行，即没有map或者reduce在执行时，划分给集群的这部分资源是可以被系统使用的。而若有map和Reduce在执行时，运行map和reduce的JVM的资源不会因为系统进程需要使用而被释放掉）所以，所有正在执行的Map一直在等待python的运行结果而没有释放掉其自身占用的资源，故python无法使用分配给集群的资源而只能使用预留给系统+nodeManager+DataNode的4-8G的内存和很少的cpu核数。因此会导致集群的资源无法被高效利用。
综上，使用Transform（Python）执行效率低的根本原因在于Python是直接向操作系统申请资源，而不是向YARN的ResourceManager申请资源，故而导致节点的资源无法高效组织和被利用。此外，不要轻易使用transform！不要轻易使用transform！不要轻易使用transform