Pig支持使用Java，Python，JavaScript三种语言编写的自定义函数UDF。一般情况下常用Java来实现Pig的自定义函数，本质上就是一个Java程序。Pig的自定义函数一共分为三种：1.自定义的过滤函数；2.自定义的运算函数；3.自定义的加载函数。其中第三种自定义函数的实现最为复杂。
开发环境搭建：
（1）下载依赖的Jar包到本地目录如E:/piglib：
PIG_HOME/lib/.jar
HADOOP_HOME/share/hadoop/common/.jar
HADOOP_HOME/share/hadoop/mapreduce/.jar
$HADOOP_HOME/share/hadoop/mapreduce/lib/.jar
（2）打开eclipse，新建Java工程PigTest，将上面下载好的全部jar包加到工程的Path路径。
1.自定义的过滤函数
例如：过滤薪水大于2000的员工信息。
package demo.pig
import java.io.IOException;
import org.apache.pig.FilterFunc;
import org.apache.pig.data.Tuple;
public class IsSalaryGreat2000 extends FilterFunc {
/**
* 重写exec方法
* 参数tuple：调用该UDF时传递的参数列表
* 例如：emp2 = filter emp by demo.pig.IsSalaryTooHigh(sal);
* 返回值：Boolean，sal>2000的为过滤条件
*/
@Override
public Boolean exec(Tuple tuple) throws IOException {
//实现过滤逻辑：取出薪水，并判断是否大于2000
int sal = Integer.parseInt(tuple.get(0).toString());
return sal>2000?true:false;
}
}
2.自定义的运算函数
例如：根据员工的薪水，判断薪水的级别。
package demo.pig
import java.io.IOException;
import org.apache.pig.EvalFunc;
import org.apache.pig.data.Tuple;
//泛型<T>：表示返回值的类型
public class CheckSalaryLevel extends EvalFunc<String> {
/**
* 重写exec方法
* 参数tuple：表示调用该UDF时传递的参数列表
* 返回值：String
*/
@Override
public String exec(Tuple tuple) throws IOException {
//实现运算的逻辑：取出薪水sal，然后根据sal大小，返回相应的字符串
int sal = Integer.parseInt(tuple.get(0).toString());
if(sal<1000)
return "Level_A";
else if(sal>=1000 && sal<3000)
return "Level_B";
else
return "Level_C";
}
}
3.自定义的加载函数
load语句在加载数据的时候，默认的加载行为是：把数据文件中的每一行解析成一个tuple。但是在某些情况下我们不想使用这种默认的加载行为，例如在进行WordCount时，我们不希望将一行解析成一个tuple，而是希望将每个单词解析成一个tuple，从而便于处理。
例如：加载数据时完成分词操作。
package demo.pig
import java.io.IOException;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.pig.LoadFunc;
import org.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;
//功能：在加载数据时，直接对数据文件的每一行进行分词操作，并将每个单词单独作为一个Tuple
public class SplitWords extends LoadFunc {
//定义文件输入流对象
private RecordReader reader;
@Override
public InputFormat getInputFormat() throws IOException {
//读入数据的类型：这里指定为文本
return new TextInputFormat();
}
@Override
public Tuple getNext() throws IOException {
//处理每一行数据：这里将读入的每一行数据分词，将每个单词作为一个Tuple返回
//返回值:Tuple类型
//说明：拿到一行数据，返回一个Tuple，该Tuple中嵌套了一张表Bag，
//该Bag中的Tuple为每个单词，从而实现了加载数据时的分词操作
Tuple lineTuple = null;
try {
//判断是否读入了数据
if(!this.reader.nextKeyValue()) {
//没有数据
return lineTuple;
}
//生成返回值对象，使用工厂模式创建tuple
lineTuple = TupleFactory.getInstance().newTuple();
//取出一行数据，例如：I love Beijing
String data = this.reader.getCurrentValue().toString();
//按照空格分词
String[] words = data.split(" ");
//创建一个内嵌表Bag
DataBag inBag = BagFactory.getInstance().newDefaultBag();
for(String word:words) {
//每个单词单独放入一个Tuple中
Tuple wordTuple = TupleFactory.getInstance().newTuple();
wordTuple.append(word);
//将wordTuple加入到inBag中
inBag.add(wordTuple);
}
//将内嵌表inBag放到lineTuple中
lineTuple.append(inBag);
}catch(Exception e) {
e.printStackTrace();
}
//返回处理好的行Tuple
return lineTuple;
}
@Override
public void prepareToRead(RecordReader reader, PigSplit arg1) throws IOException {
//RecordReader reader代表文件的输入流
//初始化
this.reader = reader;
}
@Override
public void setLocation(String path, Job job) throws IOException {
//指定输入的文件路径
FileInputFormat.setInputPaths(job, new Path(path));
}
}
4.自定义函数的使用
（0）启动环境，准备测试数据
[root@bigdata ~]# start-all.sh
[root@bigdata ~]# mr-jobhistory-daemon.sh start historyserver
[root@bigdata ~]# pig
grunt> emp = load '/input/emp.csv' using PigStorage(',') as
(empno:int,ename:chararray,job:chararray,mgr:int,
hiredate:chararray,sal:int,comm:int,deptno:int);
（1）首先将编写好的自定义函数的程序打包成PigUDF.jar，并上传至服务器的/root/input目录下。
（2）注册Jar包
grunt> register /root/input/PigUDF.jar;
（3）定义自定义函数的别名
grunt> define IsSalaryGreat2000 demo.pig.IsSalaryGreat2000;
grunt> define CheckSalaryLevel demo.pig.CheckSalaryLevel;
grunt> define SplitWords demo.pig.SplitWords;
（4）调用自定义函数
grunt> emp2 = filter emp by IsSalaryGreat2000(sal);
grunt> dump emp2;
log
(7566,JONES,MANAGER,7839,1981/4/2,2975,,20)
(7698,BLAKE,MANAGER,7839,1981/5/1,2850,,30)
(7782,CLARK,MANAGER,7839,1981/6/9,2450,,10)
(7788,SCOTT,ANALYST,7566,1987/4/19,3000,,20)
(7839,KING,PRESIDENT,,1981/11/17,5000,,10)
(7902,FORD,ANALYST,7566,1981/12/3,3000,,20)
grunt> emp3 = foreach emp generate empno,ename,sal,CheckSalaryLevel(sal);
grunt> dump emp3;
log
(7369,SMITH,800,Level_A)
(7499,ALLEN,1600,Level_B)
(7521,WARD,1250,Level_B)
(7566,JONES,2975,Level_B)
(7654,MARTIN,1250,Level_B)
(7698,BLAKE,2850,Level_B)
(7782,CLARK,2450,Level_B)
(7788,SCOTT,3000,Level_C)
(7839,KING,5000,Level_C)
(7844,TURNER,1500,Level_B)
(7876,ADAMS,1100,Level_B)
(7900,JAMES,950,Level_A)
(7902,FORD,3000,Level_C)
(7934,MILLER,1300,Level_B)
grunt> data = load '/input/data.txt' using SplitWords;
grunt> dump data;
log
({(I),(love),(Beijing)})
({(I),(love),(China)})
({(Beijing),(is),(the),(capital),(of),(China)})
本节演示了Pig的三种自定义函数的使用。祝你玩得愉快！
