玩转大数据系列之Apache Pig高级技能之函数编程（六）

原创不易，转载请务必注明，原创地址，谢谢配合!
http://qindongliang.iteye.com/

Pig系列的学习文档，希望对大家有用，感谢关注散仙！

如何使用Pig集成分词器来统计新闻词频？

在Hadoop的生态系统中，如果我们要离线的分析海量的数据，大多数人都会选择Apache
Hive或Apache Pig，在国内总体来说，Hive使用的人群占比比较高，
而Pig使用的人相对来说，则少的多，这并不是因为Pig不成熟，不稳定，而是因为Hive提供了类数据库SQL的查询语句，使得大多人上手Hive非常容易，相反而Pig则提供了类Linux
shell的脚本语法，这使得大多数人不喜欢使用。

OK，扯远了，赶紧回来，使用shell的攻城师们，我觉得都会爱上它的，因为在linux系统中，没有比shell更简洁易用了，如果再配上awk和sed更是如虎添翼了。

下面看几个例子，让我们迅速对它熟悉并掌握，先看下我们的测试数据：

Java代码  

1. 1,张三,男,23,中国
2. 2,张三,女,32,法国
3. 3,小花,男,20,英国
4. 4,小红,男,16,中国
5. 5,小红,女,25,洛阳
6. 6,李静,女,25,中国河南安阳
7. 7,王强,男,11,英国
8. 8,张飞,男,20,美国

1,张三,男,23,中国
2,张三,女,32,法国
3,小花,男,20,英国
4,小红,男,16,中国
5,小红,女,25,洛阳
6,李静,女,25,中国河南安阳
7,王强,男,11,英国
8,张飞,男,20,美国

再看下pig脚本：

Java代码  

1. --定义pig函数1 支持分组统计数量
2. DEFINE group_and_count (A,group_key,number_reduces) RETURNS B {
3. d = group $A by $group_key parallel $number_reduces;
4. $B = foreach d generate group, COUNT($1);
5. };
6. --定义pig函数2 支持排序
7. --A 关系引用标量
8. --order_field 排序的字段
9. --order_type 排序方式 desc ? asc ?
10. --storedir 存储的HDFS路径
11. --空返回值
12. define my_order(A,order_field,order_type,storedir) returns void {
13. d = order $A by $order_field $order_type ;
14. store  d into ‘$storedir‘ ;
15. };
16. --定义pig函数3，支持filter过滤，以及宏命令里面调用
17. --定义过滤操作
18. define  myfilter (A,field,count) returns B{
19. b= filter $A by $field > $count ;
20. $B = group_and_count(b,‘sex‘,1);
21. };
22. a = load  ‘/tmp/dongliang/318/person‘ using PigStorage(‘,‘) AS (id:int,name:chararray,sex:chararray,age:int,address:chararray) ;
23. --------pig函数1测试-----------------
24. --定义按名字分组
25. --bb = group_and_count(a,name,1);
26. --定义按性别分组
27. --cc = group_and_count(a,sex,1);
28. --dump bb;
29. --dump cc;
30. -------pig函数2测试------------------
31. --按年龄降序
32. --my_order(a,age,‘desc‘,‘/tmp/dongliang/318/z‘);
33. --dump a;
34. -------pig函数3测试------------------
35. --过滤年龄大于20的，并按性别，分组统计数量
36. r =  myfilter(a,‘age‘,20);
37. dump r;

--定义pig函数1 支持分组统计数量
DEFINE group_and_count (A,group_key,number_reduces) RETURNS B {

 d = group $A by $group_key parallel $number_reduces;

 $B = foreach d generate group, COUNT($1);

};

--定义pig函数2 支持排序
--A 关系引用标量
--order_field 排序的字段
--order_type 排序方式 desc ? asc ?
--storedir 存储的HDFS路径
--空返回值
define my_order(A,order_field,order_type,storedir) returns void {

  d = order $A by $order_field $order_type ;
  store  d into ‘$storedir‘ ;  

}; 

--定义pig函数3，支持filter过滤，以及宏命令里面调用

--定义过滤操作
define  myfilter (A,field,count) returns B{

   b= filter $A by $field > $count ;

   $B = group_and_count(b,‘sex‘,1);

};

a = load  ‘/tmp/dongliang/318/person‘ using PigStorage(‘,‘) AS (id:int,name:chararray,sex:chararray,age:int,address:chararray) ;

--------pig函数1测试-----------------

--定义按名字分组
--bb = group_and_count(a,name,1);
--定义按性别分组
--cc = group_and_count(a,sex,1);
--dump bb;
--dump cc;

-------pig函数2测试------------------

--按年龄降序
--my_order(a,age,‘desc‘,‘/tmp/dongliang/318/z‘);

--dump a;

-------pig函数3测试------------------

 --过滤年龄大于20的，并按性别，分组统计数量
 r =  myfilter(a,‘age‘,20);

dump r;

在上面的脚本中，散仙定义了三个函数，
（1）分组统计数量
（2）自定义输出存储

一：函数脚本文件

Java代码  

1. --定义pig函数1 支持分组统计数量
2. --A 关系引用标量
3. --group_key 分组字段
4. --使用reduce的个数
5. --返回最终的引用结果
6. DEFINE group_and_count (A,group_key,number_reduces) RETURNS B {
7. d = group $A by $group_key parallel $number_reduces;
8. $B = foreach d generate group, COUNT($1);
9. };
10. --定义pig函数2 支持排序
11. --A 关系引用标量
12. --order_field 排序的字段
13. --order_type 排序方式 desc ? asc ?
14. --storedir 存储的HDFS路径
15. --空返回值
16. define my_order(A,order_field,order_type,storedir) returns void {
17. d = order $A by $order_field $order_type ;
18. store  d into ‘$storedir‘ ;
19. };
20. --定义pig函数3，支持filter过滤，以及宏命令里面调用
21. --A 关系引用标量
22. --field 过滤的字段
23. --count 阈值
24. --返回最终的引用结果
25. define  myfilter (A,field,count) returns B{
26. b= filter $A by $field > $count ;
27. $B = group_and_count(b,‘sex‘,1);
28. };
29. [search@dnode1 pigmacros]$

--定义pig函数1 支持分组统计数量
--A 关系引用标量
--group_key 分组字段
--使用reduce的个数
--返回最终的引用结果
DEFINE group_and_count (A,group_key,number_reduces) RETURNS B {

 d = group $A by $group_key parallel $number_reduces;

 $B = foreach d generate group, COUNT($1);

};

--定义pig函数2 支持排序
--A 关系引用标量
--order_field 排序的字段
--order_type 排序方式 desc ? asc ?
--storedir 存储的HDFS路径
--空返回值
define my_order(A,order_field,order_type,storedir) returns void {

  d = order $A by $order_field $order_type ;
  store  d into ‘$storedir‘ ;  

}; 

--定义pig函数3，支持filter过滤，以及宏命令里面调用
--A 关系引用标量
--field 过滤的字段
--count 阈值
--返回最终的引用结果

define  myfilter (A,field,count) returns B{

   b= filter $A by $field > $count ;

   $B = group_and_count(b,‘sex‘,1);

};

[[email protected] pigmacros]$ 

二，主体脚本文件

Java代码  

1. --导入pig公用的函数库
2. import ‘function.pig‘ ;
3. a = load  ‘/tmp/dongliang/318/person‘ using PigStorage(‘,‘) AS (id:int,name:chararray,sex:chararray,age:int,address:chararray) ;
4. --------pig函数1测试-----------------
5. --定义按名字分组
6. --bb = group_and_count(a,name,1);
7. --定义按性别分组
8. --cc = group_and_count(a,sex,1);
9. --dump bb;
10. --dump cc;
11. -------pig函数2测试------------------
12. --按年龄降序
13. --my_order(a,age,‘desc‘,‘/tmp/dongliang/318/z‘);
14. --dump a;
15. -------pig函数3测试------------------
16. --过滤年龄大于20的，并按性别，分组统计数量
17. r =  myfilter(a,‘age‘,20);
18. dump r;

--导入pig公用的函数库

import ‘function.pig‘ ;

a = load  ‘/tmp/dongliang/318/person‘ using PigStorage(‘,‘) AS (id:int,name:chararray,sex:chararray,age:int,address:chararray) ;

--------pig函数1测试-----------------

--定义按名字分组
--bb = group_and_count(a,name,1);
--定义按性别分组
--cc = group_and_count(a,sex,1);
--dump bb;
--dump cc;

-------pig函数2测试------------------

--按年龄降序
--my_order(a,age,‘desc‘,‘/tmp/dongliang/318/z‘);
--dump a;

-------pig函数3测试------------------

 --过滤年龄大于20的，并按性别，分组统计数量
 r =  myfilter(a,‘age‘,20);
 dump r;

需要注意的是，导入的函数文件，需要用单引号引起来，这样我们就完成了pig函数的重用，是不是非常类似shell的语法呢？有兴趣的同学们，赶紧体验一把吧！