Apache Pig如何自定义UDF函数？

悠悠 2022-04-12 06:53 240阅读 0赞

\[b\]\[color=green\]\[size=large\]近日由于工作所需，需要使用到Pig来分析线上的搜索日志数据，散仙本打算使用hive来分析的，但由于种种原因，没有用成，而Pig（pig0.12-cdh）散仙一直没有接触过，所以只能临阵磨枪了，花了两天时间，大致看完了pig官网的\[url=http://pig.apache.org/docs/r0.12.1/basic.html\]文档\[/url\]，在看文档期间，也是边实战边学习，这样以来，对pig的学习，会更加容易，当然本篇不是介绍如何快速学好一门框架或语言的文章，正如标题所示，散仙打算介绍下如何在Pig中，使用用户自定义的UDF函数，关于学习经验，散仙会在后面的文章里介绍。  
  
\[/size\]\[/color\]\[/b\]  
  
\[b\]\[color=olive\]\[size=large\]一旦你学会了UDF的使用，就意味着，你可以以更加灵活的方式来使用Pig，使它扩展一些为我们的业务场景定制的特殊功能，而这些功能，在通用的pig里是没有的，举个例子：  
  
你从HDFS上读取的数据格式，如果使用默认的PigStorage（）来加载，存储可能只支持有限的数据编码和类型，如果我们定义了一种特殊的编码存储或序列化方式，那么当我们使用默认的Pig来加载的时候，就会发现加载不了，这时候我们的UDF就派上用场了，我们只需要自定义一个LoadFunction和一个StoreFunction就可以解决，这种问题。  
  
  
本篇散仙根据官方文档的例子，来实战一下，并在hadoop集群上使用Pig测试通过：  
我们先来看下定义一个UDF扩展类，需要几个步骤：  
\[table\]  
|序号|步骤|说明  
|1|在eclipse里新建一个java工程，并导入pig的核心包|java项目  
|2|新建一个包，继承特定的接口或类，重写自定义部分|核心业务  
|3|编写完成后，使用ant打包成jar|编译时需要pig依赖，但不用把pig的jar包打入UDF中  
|4|把打包完成后的jar上传到HDFS上|pig运行时候需要加载使用  
|5|在pig脚本里，注册我们自定义的udf的jar包|注入运行时环境  
|6|编写我们的核心业务pig脚本运行|测试是否运行成功  
\[/table\]  
\[/size\]\[/color\]\[/b\]  
\[b\]\[color=green\]\[size=large\]项目工程截图如下：\[/size\]\[/color\]\[/b\]  
  
\[img\]http://dl2.iteye.com/upload/attachment/0104/7919/7dcf5b26-7ec1-3089-91d9-32d606e4135b.jpg\[/img\]  
  
  
  
\[b\]\[color=green\]\[size=large\]核心代码如下：\[/size\]\[/color\]\[/b\]

package com.pigudf;
    
    import java.io.IOException;
    
    import org.apache.pig.EvalFunc;
    import org.apache.pig.data.Tuple;
    import org.apache.pig.impl.util.WrappedIOException;
    /**
     * 自定义UDF类,对字符串转换大写
     * @author qindongliang
     * */
    public class MyUDF extends EvalFunc<String> {
       
    
    	@Override
    	public String exec(Tuple input) throws IOException {
       
    
    		 //判断是否为null或空，就跳过
    		if(input==null||input.size()==0){
       
    			return null;
    		}
    		try{
       
    			//获取第一个元素
    			String str=(String) input.get(0);
    			//转成大写返回
    			return str.toUpperCase();
    
    		}catch(Exception e){
       
    			throw WrappedIOException.wrap("Caught exception processing input row ",e);
    		}
    	}
    
    
    }

\[b\]\[color=olive\]\[size=large\]关于打包的ant脚本，散仙会在文末上传附件，下面看下造的一些测试数据（注意，文件一定要上传到HDFS上，除非你是local模式）：\[/size\]\[/color\]\[/b\]

grunt> cat s.txt
    zhang san,12
    Song,34
    long,34
    abC,12
    grunt>

\[b\]\[color=olive\]\[size=large\]我们在看下，操作文件和jar包是放在一起的：\[/size\]\[/color\]\[/b\]

grunt> ls
    hdfs://dnode1:8020/tmp/udf/pudf.jar<r 3>        1295
    hdfs://dnode1:8020/tmp/udf/s.txt<r 3>   36
    grunt>

\[b\]\[color=green\]\[size=large\]最后，我们看下pig脚本的定义：\[/size\]\[/color\]\[/b\]

--注册自定义的jar包
    REGISTER pudf.jar; 
    --加载测试文件的数据，逗号作为分隔符
    a = load 's.txt' using PigStorage(',');   
    --遍历数据，对name列转成大写
    b =  foreach a generate com.pigudf.MyUDF((chararray)$0); 
    --启动MapReduce的Job进行数据分析
    dump b

\[b\]\[color=green\]\[size=large\]最后，我们看下结果，只要过程不出现异常和任务失败，就证明我们的udf使用成功：\[/size\]\[/color\]\[/b\]

Counters:
    Total records written : 4
    Total bytes written : 64
    Spillable Memory Manager spill count : 0
    Total bags proactively spilled: 0
    Total records proactively spilled: 0
    
    Job DAG:
    job_1419419533357_0147
    
    
    2014-12-30 18:10:24,394 [main] INFO  org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MapReduceLauncher - Success!
    2014-12-30 18:10:24,395 [main] INFO  org.apache.hadoop.conf.Configuration.deprecation - fs.default.name is deprecated. Instead, use fs.defaultFS
    2014-12-30 18:10:24,396 [main] INFO  org.apache.pig.data.SchemaTupleBackend - Key [pig.schematuple] was not set... will not generate code.
    2014-12-30 18:10:24,405 [main] INFO  org.apache.hadoop.mapreduce.lib.input.FileInputFormat - Total input paths to process : 1
    2014-12-30 18:10:24,405 [main] INFO  org.apache.pig.backend.hadoop.executionengine.util.MapRedUtil - Total input paths to process : 1
    (ZHANG SAN,12)
    (SONG,34)
    (LONG,34)
    (ABC,12)

\[b\]\[color=green\]\[size=large\]结果没问题，我们的UDF加载执行成功，如果我们还想将我们的输出结果直接写入到HDFS上，可以在pig脚本的末尾，去掉dump命令，加入  
store e into '/tmp/dongliang/result/'; 将结果存储到HDFS上，当然我们可以自定义存储函数，将结果写入数据库，Lucene，Hbase等关系型或一些NOSQL数据库里。  
\[/size\]\[/color\]\[/b\]  
  
  
\[b\]\[color=green\]\[size=large\]  
  
欢迎大家关注微信公众号：我是工程师(woshigcs)，更多与攻城师息息相关的内容，尽在此处。  
二维码扫描：  
\[/size\]\[/color\]\[/b\]  
  
  
\[img\]http://dl2.iteye.com/upload/attachment/0104/6808/bc78dca9-5a52-38cc-b4d5-23128f84a4b7.jpg\[/img\]