Pig用户自定义函数（UDF）
我们以气温统计和词频统计为例，讲解以下三种用户自定义函数。
用户自定义函数
什么时候需要用户自定义函数呢？和其它语言一样，当你希望简化程序结构或者需要重用程序代码时，函数就是你不二选择。
Pig的用户自定义函数可以用Java编写，但是也可以用Python或Javascript编写。我们接下来以Java为例。
自定义过滤函数
我们仍然以先前的代码为例：
records = load 'hdfs://localhost:9000/input/temperature1.txt'as (year: chararray,temperature: int);
valid_records = filter records by temperature!=999;
第二个语句的作用就是筛选合法的数据。如果我们采用用户自定义函数，则第二个语句可以写成：
valid_records = filter records by isValid(temperature);
这种写法更容易理解，也更容易在多个地方重用。接下来的问题就是如何定义这个isValid函数。代码如下：
packagecom.oserp.pigudf;
importjava.io.IOException;
importorg.apache.pig.FilterFunc;
importorg.apache.pig.data.Tuple;
public class IsValidTemperature extends FilterFunc {
@Override
public Boolean exec(Tuple tuple)throws IOException {
Object object = tuple.get(0);
int temperature = (Integer)object;
return temperature != 999;
}
}
接下来，我们需要：
1） 编译代码并打包成jar文件，比如pigudf.jar。
2） 通过register命令将这个jar文件注册到pig环境：
register/home/user/hadoop_jar/pigudf.jar //参数为jar文件的本地路径
此时，我们就可以用以下语句调用这个函数：
valid_records = filter records bycom.oserp.pigudf.IsValidTemperature(temperature);
dump valid_records;
看起来这个函数名太长，不便输入。我们可以用定义别名的方式代替：
define isValid com.oserp.pigudf.IsValidTemperature();
valid_records = filter records by isValid(temperature);
dump valid_records;
回到代码，我们可发现：
1） 需要定义一个继承自FilterFunc的类。
2） 重写这个类的exec方法。这个方法的参数只有一个tuple，但是调用时可以传递多个参数，你可以通过索引号获得对应的参数值，比如tuple.get(1)表示取第二个参数。
3） 调用时，需要使用类的全名。（当然你可以自定义别名）
4） 更多的验证需要读者自行在函数中添加，比如判断是否为null等等。
备注：用Eclipse编写Pig自定义函数时，你可能需要引用到一些Hadoop的库文件。比较容易的方式是在新建项目时指定项目类型为MapReduce项目，这样Eclipse就会自动设置库引用的相关信息。
自定义运算函数（Eval function）
仍然以前面的数据文件为例：
1990 21
1990 18
1991 21
1992 30
1992 999
1990 23
假设我们希望通过温度值获得一个温度的分类信息，比如我们把温度大于划分为以下类型:
温度 分类
x>=30 hot
x>=10 and x<30 moderate
x<10 cool
则我们可以定义以下函数，代码如下：
packagecom.oserp.pigudf;
importjava.io.IOException;
importorg.apache.pig.EvalFunc;
importorg.apache.pig.data.Tuple;
public class GetClassification extends EvalFunc<String> {
@Override
public String exec(Tuple tuple)throws IOException {
Object object = tuple.get(0);
int temperature = (Integer)object;
if (temperature >= 30){
return "Hot";
}
else if(temperature >=10){
return "Moderate";
}
else {
return "Cool";
}
}
}
依次输入以下Pig语句：
records = load'hdfs://localhost:9000/input/temperature1.txt' as (year: chararray,temperature:int);
register /home/user/hadoop_jar/pigudf.jar;
valid_records = filter records bycom.oserp.pigudf.IsValidTemperature(temperature);
result = foreach valid_records generateyear,com.oserp.pigudf.GetClassification(temperature);
dump result;
输出结果如下：
(1990,Moderate)
(1990,Moderate)
(1991,Moderate)
(1992,Hot)
(1990,Moderate)
代码比较简单，该类继承自EvalFunc类，且我们要明确定义返回值类型。
有些时候其它类库可能包含有功能相近的Java函数，我们是否可以直接将这些库函数拿过来使用呢？可以。以下语句调用了trim函数，用于去掉name字段前后的空格：
DEFINE trim InvokeForString('org.apache.commons.lang.StringUtils.trim','String');
B = FOREACH A GENERATE trim(name);
其中的InvokeForString是一个Invoker（不知道该如何翻译啊），其通过反射机制调用，返回值是String类型。其它类似的还有InvokeForInt，InvokeForLong, InvokeForDouble，InvokeForFloat等等。
自定义加载函数
我们以词频统计为例，讲解如何自定义加载函数。（统计各个单词出现的频率，由高到低排序）
一般情况下，load语句加载数据时，一行会被生成一个tuple。而统计词频时，我们希望每个单词生成一个tuple。我们的测试数据文件只有两行数据，如下：
Thisis a map a reduce program
mapreduce partition combiner
我们希望load后能得到如下形式的数据，每个单词一个tuple：
(This)
(is)
(a)
(map)
(a)
(reduce)
先看代码：
package com.oserp.pigudf;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.hadoop.io.Text;
importorg.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
importorg.apache.hadoop.mapreduce.RecordReader;
importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;
importorg.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.pig.LoadFunc;
importorg.apache.pig.backend.executionengine.ExecException;
importorg.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.BagFactory;
import org.apache.pig.data.DataBag;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
public class WordCountLoadFunc extends LoadFunc {
private RecordReader reader;
TupleFactorytupleFactory = TupleFactory.getInstance();
BagFactorybagFactory = BagFactory.getInstance();
@Override
public InputFormatgetInputFormat() throws IOException {
return new TextInputFormat();
}
@Override
public Tuple getNext()throws IOException {
try {
// 当读取到分区数据块的末尾时，返回null表示数据已读取完
if (!reader.nextKeyValue()){
return null;
}
Textvalue = (Text)reader.getCurrentValue();
Stringline = value.toString();
String[]words = line.split("\\s+");// 断词
// 因为getNext函数只能返回一个tuple，
// 而我们希望每个单词一个单独的tuple，
// 所以我们将多个tuple放到一个bag里面，
// 然后返回一个包含一个bag的tuple。
// 注：这只是一个用于演示用法的示例，实际中这样使用不一定合理。
List<Tuple>tuples = new ArrayList<Tuple>();
Tupletuple = null;
for (String word : words) {
tuple= tupleFactory.newTuple();
tuple.append(word);
tuples.add(tuple);
}
DataBagbag = bagFactory.newDefaultBag(tuples);
Tupleresult = tupleFactory.newTuple(bag);
return result;
}
catch (InterruptedException e) {
throw new ExecException(e);
}
}
@Override
public void prepareToRead(RecordReader reader,PigSplit arg1)
throws IOException {
this.reader = reader;
}
@Override
public void setLocation(String location, Job job) throws IOException {
FileInputFormat.setInputPaths(job,location);
}
}
依次执行以下命令：
1) records= load 'hdfs://localhost:9000/input/sample_small.txt' usingcom.oserp.pigudf.WordCountLoadFunc() as (words:bag{word:(w:chararray)});
2) flatten_records= foreach records generate flatten($0);
3) grouped_records= group flatten_records by words::w;
4) result= foreach grouped_records generate group,COUNT(flatten_records);
5) final_result= order result by $1 desc,$0;
6) dumpfinal_result;
显示结果如下：
(a,2)
(map,2)
(reduce,2)
(This,1)
(combiner,1)
(is,1)
(partition,1)
(program,1)
注意schema的定义格式：(words:bag{word:(w:chararray)})