Pig是一个基于Hadoop的大规模数据分析平台&＃xff0c;它提供的SQL-LIKE语言叫Pig Latin&＃xff0c;该语言的编译器会把类SQL的数据分析请求转换为一系列经过优化处理的MapReduce运算。Pig为复杂的海量数据并行计算提供了一个简单的操作和编程接口。

pig关系操作符&＃xff1a;

关系名上面有个student

还需要dump g&＃xff1b;

产生结果 &＃xff1a;one is age,another is a bag;

对多个关系进行分组&＃xff0c;合并之后的分组。

----------------------------------------------------------------------------------------------------------------------------------------------------------

所有命令和脚本都在Pig 0.12.0 & Hadoop 2.2.0下测试通过。

准备两个数据文件&＃xff1a;

1&＃xff09;student.txt 结构为&＃xff08;班级号&＃xff0c;学号&＃xff0c;成绩&＃xff09;,字段间逗号分隔。

C01,N0101,82

C01,N0102,59

C01,N0103,65

C02,N0201,81

C02,N0202,82

C02,N0203,79

C03,N0301,56

C03,N0302,92

C03,N0306,72

2&＃xff09;teacher.txt 结构为&＃xff08;班级号&＃xff0c;教师&＃xff09;,字段间逗号分隔。

C01,Zhang

C02,Sun

C03,Wang

C04,Dong

加载和存储&＃xff08;Load&＃xff0c;Store&＃xff09;

执行以下命令

records &＃61; load&＃39;hdfs://localhost:9000/input/student.txt&＃39; using PigStorage(&＃39;,&＃39;) as(classNo:chararray, studNo:chararray, score:int);

dump records;

store records into &＃39; hdfs://localhost:9000/input/student_out&＃39; using PigStorage(&＃39;:&＃39;);

然后查看hdfs://localhost:9000/input/student_out目录下的part-m-00000文件&＃xff0c;其内容如下&＃xff1a;

C01:N0101:82

C01:N0102:59

C01:N0103:65

C02:N0201:81

C02:N0202:82

C02:N0203:79

C03:N0301:56

C03:N0302:92

C03:N0306:72

其中的load是加载操作&＃xff0c;store是存储操作。他们分别可以指定其分隔符&＃xff0c;比如上例中的逗号和分号。

筛选&＃xff08;Filter&＃xff09;

执行以下命令&＃xff1a;

records_c01 &＃61; filter records byclassNo&＃61;&＃61;&＃39;C01&＃39;;

dump records_c01;

结果如下&＃xff1a;

(C01,N0101,82)

(C01,N0102,59)

(C01,N0103,65)

注意&＃xff1a;判断是否相等要用两个等号。

Foreach Generate

Foreach对关系中的每一个记录循环&＃xff0c;然后按指定模式生成一个新的关系。

执行以下命令&＃xff1a;

score_c01 &＃61; foreach records_c01generate &＃39;Teacher&＃39;,$1,score;

dump score_c01;

结果如下&＃xff1a;

(Teacher,N0101,82)

(Teacher,N0102,59)

(Teacher,N0103,65)

生成的新的关系中包括三个字段&＃xff0c;第一个字段是常量&＃xff0c;第二个字段是学号&＃xff08;我们是通过索引号引用的&＃xff09;&＃xff0c;第三个字段是分数&＃xff08;我们通过字段名引用的&＃xff09;。

分组&＃xff08;group&＃xff09;

执行以下命令&＃xff1a;

grouped_records &＃61; group recordsby classNo parallel 2;

dump grouped_records;

结果如下&＃xff1a;

(C02,{(C02,N0203,79),(C02,N0202,82),(C02,N0201,81)})

(C01,{(C01,N0103,65),(C01,N0102,59),(C01,N0101,82)})

(C03,{(C03,N0306,72),(C03,N0302,92),(C03,N0301,56)})

其中的Paraller 2表示启用2个Reduce操作。

如何统计每个班级及格和优秀的学生人数呢&＃xff1f;执行以下两个命令&＃xff1a;

result &＃61; foreach grouped_records {

         fail &＃61;filter records by score <60;

         excellent &＃61;filter records by score >&＃61;90;

         generategroup, COUNT(fail) as fail, COUNT(excellent) as excellent;

};

dump result;

结果如下&＃xff1a;

(C01,1,0)

(C02,0,0)

(C03,1,1)

 

题外话&＃xff1a;

flatten操作&＃xff0c;可以将数据格式扁平化。我们分别通过tuple和bag来看看flatten的作用&＃xff1a;

1&＃xff09;  Flatten对tuple的作用

执行以下命令:

a&＃61; foreach records generate $0,($1,$2);

dumpa;

输出结果如下&＃xff1a;

(C01,(N0101,82))

(C01,(N0102,59))

(C01,(N0103,65))

(C02,(N0201,81))

(C02,(N0202,82))

(C02,(N0203,79))

(C03,(N0301,56))

(C03,(N0302,92))

(C03,(N0306,72))

然后&＃xff0c;执行:

b &＃61; foreach a generate $0,flatten($1);

dump b;

结果如下&＃xff1a;

(C01,N0101,82)

(C01,N0102,59)

(C01,N0103,65)

(C02,N0201,81)

(C02,N0202,82)

(C02,N0203,79)

(C03,N0301,56)

(C03,N0302,92)

(C03,N0306,72)

由此看见&＃xff0c;flatten作用于tuple时&＃xff0c;将flatten对应的字段&＃xff08;tuple&＃xff09;中的字段扁平化为关系中的字段。&＃xff08;不知道该如何解释比较好&＃xff09;

2&＃xff09;  Flatten对bag的作用

执行以下命令

c &＃61; foreach records generate $0,{($1),($1,$2)};

dump c;

结果如下&＃xff1a;

(C01,{(N0101),(N0101,82)})

(C01,{(N0102),(N0102,59)})

(C01,{(N0103),(N0103,65)})

(C02,{(N0201),(N0201,81)})

(C02,{(N0202),(N0202,82)})

(C02,{(N0203),(N0203,79)})

(C03,{(N0301),(N0301,56)})

(C03,{(N0302),(N0302,92)})

(C03,{(N0306),(N0306,72)})

接下来执行&＃xff1a;

d &＃61; foreach c generate $0,flatten($1);

dump d;

结果如下&＃xff1a;

(C01,N0101)

(C01,N0101,82)

(C01,N0102)

(C01,N0102,59)

(C01,N0103)

(C01,N0103,65)

(C02,N0201)

(C02,N0201,81)

(C02,N0202)

(C02,N0202,82)

(C02,N0203)

(C02,N0203,79)

(C03,N0301)

(C03,N0301,56)

(C03,N0302)

(C03,N0302,92)

(C03,N0306)

(C03,N0306,72)

可以看出&＃xff0c;flatten作用于bag时&＃xff0c;会消除嵌套关系&＃xff0c;生成类似于笛卡尔乘积的结果。&＃xff08;不好表达&＃xff0c;读者可以细细体会&＃xff09;。

Stream操作

可以将Python程序嵌入到Pig中使用。

建立一个Python文件pass.py&＃xff0c;内容如下&＃xff1a;

#! /usr/bin/envpython

import sys

 

for line insys.stdin:

         (c,n,s) &＃61; line.split()  

         if int(s) >&＃61; 60:

                   print "%s\t%s\t%s"%(c,n,s)

 

执行以下命令:

define pass &＃96;pass.py&＃96; SHIP(&＃39;/home/user/pass.py&＃39;);

records_pass &＃61; stream records through pass as(classNo:chararray, studNo:chararray, score:int);

dump records_pass;

结果如下&＃xff1a;

(C01,N0101,82)

(C01,N0103,65)

(C02,N0201,81)

(C02,N0202,82)

(C02,N0203,79)

(C03,N0302,92)

(C03,N0306,72)

可以看出&＃xff0c;统计结果为所有及格的记录&＃xff08;>&＃61;60&＃xff09;。

其中&＃xff0c;ship用于将python程序提交到Hadoop集群中去。

请注意第一个命令中的&＃96;pass.py&＃96;&＃xff0c;不是用单引号括起来的&＃xff0c;是用键盘1左边的那个键上的字符括起来的。&＃xff08;不知道这个字符怎么称呼&＃xff0c;只知道是一种标注符号&＃xff09;

Join

先执行以下两条命令&＃xff1a;

r_student &＃61; load&＃39;hdfs://localhost:9000/input/student.txt&＃39; using PigStorage(&＃39;,&＃39;) as (classNo:chararray, studNo: chararray, score: int);

r_teacher2 &＃61; load&＃39;hdfs://localhost:9000/input/teacher.txt&＃39; using PigStorage(&＃39;,&＃39;) as (classNo:chararray, teacher: chararray);

回到本文开头&＃xff0c;我们有两个数据文件&＃xff0c;分别为学生&＃xff08;班级&＃xff0c;学号&＃xff0c;成绩&＃xff09;&＃xff1b;老师&＃xff08;班级&＃xff0c;姓名&＃xff09;。

执行以下命令&＃xff1a;

r_joined &＃61; join r_student by classNo,r_teacher by classNo;

dump r_joined;

(C01,N0103,65,C01,Zhang)

(C01,N0102,59,C01,Zhang)

(C01,N0101,82,C01,Zhang)

(C02,N0203,79,C02,Sun)

(C02,N0202,82,C02,Sun)

(C02,N0201,81,C02,Sun)

(C03,N0306,72,C03,Wang)

(C03,N0302,92,C03,Wang)

(C03,N0301,56,C03,Wang)

类似于SQL中的内连接Inner Join。当然你也可以使用外连接&＃xff0c;比如&＃xff1a;

r_joined &＃61; join r_student by classNo left outer,r_teacher by classNo;

dump r_joined;

注意&＃xff1a;left outer/right outer要写在第一个关系名的后面。以下语法是错误的&＃xff1a;

r_joined &＃61; join r_student by classNo, r_teacher by classNo leftouter; //错误

 

COGROUP

Join的操作结果是平面的&＃xff08;一组元组&＃xff09;&＃xff0c;而COGROUP的结果是有嵌套结构的。

运行以下命令&＃xff1a;

r1 &＃61; cogroup r_student by classNo,r_teacher by classNo;

dump r1;

结果如下&＃xff1a;

(C01,{(C01,N0103,65),(C01,N0102,59),(C01,N0101,82)},{(C01,Zhang)})

(C02,{(C02,N0203,79),(C02,N0202,82),(C02,N0201,81)},{(C02,Sun)})

(C03,{(C03,N0306,72),(C03,N0302,92),(C03,N0301,56)},{(C03,Wang)})

(C04,{},{(C04,Dong)})

由结果可以看出&＃xff1a;

1&＃xff09;  cogroup和join操作类似。

2&＃xff09;  生成的关系有3个字段。第一个字段为连接字段&＃xff1b;第二个字段是一个包&＃xff0c;值为关系1中的满足匹配关系的所有元组&＃xff1b;第三个字段也是一个包&＃xff0c;值为关系2中的满足匹配关系的所有元组。

3&＃xff09;  类似于Join的外连接。比如结果中的第四个记录&＃xff0c;第二个字段值为空包&＃xff0c;因为关系1中没有满足条件的记录。实际上第一条语句和以下语句等同&＃xff1a;

r1&＃61; cogroup r_student by classNo outer,r_teacher by classNo outer;

 

如果你希望关系1或2中没有匹配记录时不在结果中出现&＃xff0c;则可以分别在关系中使用inner而关键字进行排除。

执行以下语句&＃xff1a;

r1 &＃61; cogroup r_student by classNo inner,r_teacher byclassNo outer;

dump r1;

结果为&＃xff1a;

(C01,{(C01,N0103,65),(C01,N0102,59),(C01,N0101,82)},{(C01,Zhang)})

(C02,{(C02,N0203,79),(C02,N0202,82),(C02,N0201,81)},{(C02,Sun)})

(C03,{(C03,N0306,72),(C03,N0302,92),(C03,N0301,56)},{(C03,Wang)})

 

如先前我们讲到的flatten&＃xff0c;执行以下命令&＃xff1a;

r2 &＃61; foreach r1 generate flatten($1),flatten($2);

dump r2;

结果如下&＃xff1a;

(C01,N0103,65,C01,Zhang)

(C01,N0102,59,C01,Zhang)

(C01,N0101,82,C01,Zhang)

(C02,N0203,79,C02,Sun)

(C02,N0202,82,C02,Sun)

(C02,N0201,81,C02,Sun)

(C03,N0306,72,C03,Wang)

(C03,N0302,92,C03,Wang)

(C03,N0301,56,C03,Wang)

 

Cross

执行以下命令&＃xff1a;

r &＃61; cross r_student,r_teacher;

dump r;

结果如下&＃xff1a;

(C03,N0306,72,C04,Dong)

(C03,N0306,72,C03,Wang)

(C03,N0306,72,C02,Sun)

(C03,N0306,72,C01,Zhang)

(C03,N0302,92,C04,Dong)

(C03,N0302,92,C03,Wang)

(C03,N0302,92,C02,Sun)

(C03,N0302,92,C01,Zhang)

(C03,N0301,56,C04,Dong)

(C03,N0301,56,C03,Wang)

(C03,N0301,56,C02,Sun)

(C03,N0301,56,C01,Zhang)

(C02,N0203,79,C04,Dong)

(C02,N0203,79,C03,Wang)

(C02,N0203,79,C02,Sun)

(C02,N0203,79,C01,Zhang)

(C02,N0202,82,C04,Dong)

(C02,N0202,82,C03,Wang)

(C02,N0202,82,C02,Sun)

(C02,N0202,82,C01,Zhang)

(C02,N0201,81,C04,Dong)

(C02,N0201,81,C03,Wang)

(C02,N0201,81,C02,Sun)

(C02,N0201,81,C01,Zhang)

(C01,N0103,65,C04,Dong)

(C01,N0103,65,C03,Wang)

(C01,N0103,65,C02,Sun)

(C01,N0103,65,C01,Zhang)

(C01,N0102,59,C04,Dong)

(C01,N0102,59,C03,Wang)

(C01,N0102,59,C02,Sun)

(C01,N0102,59,C01,Zhang)

(C01,N0101,82,C04,Dong)

(C01,N0101,82,C03,Wang)

(C01,N0101,82,C02,Sun)

(C01,N0101,82,C01,Zhang)

由此可以看出&＃xff0c;cross类似于笛卡尔乘积。一般情况下不建议直接使用cross&＃xff0c;而应该事前对数据集进行筛选&＃xff0c;提高效率。

排序&＃xff08;Order&＃xff09;

执行以下命令&＃xff1a;

r &＃61; order r_student by score desc, classNo asc;

dump r;

结果如下&＃xff1a;

(C03,N0302,92)

(C01,N0101,82)

(C02,N0202,82)

(C02,N0201,81)

(C02,N0203,79)

(C03,N0306,72)

(C01,N0103,65)

(C01,N0102,59)

(C03,N0301,56)

联合&＃xff08;Union&＃xff09;

执行以下语句&＃xff1a;

r_union &＃61; union r_student, r_teacher;

dump r_union;

结果如下&＃xff1a;

(C01,N0101,82)

(C01,N0102,59)

(C01,N0103,65)

(C02,N0201,81)

(C02,N0202,82)

(C02,N0203,79)

(C03,N0301,56)

(C03,N0302,92)

(C03,N0306,72)

(C01,Zhang)

(C02,Sun)

(C03,Wang)

(C04,Dong)

可以看出&＃xff1a;

1&＃xff09;  union是取两个记录集的并集。

2&＃xff09;  关系r_union的schema为未知&＃xff08;unknown&＃xff09;&＃xff0c;这是因为被union的两个关系的schema是不一样的。如果两个关系的schema是一致的&＃xff0c;则union后的关系将和被union的关系的schema一致。

----------------------------------------------------------------------------------------------------------------------------------------------------------

从实例出发

%default file test.txt

A &＃61; load &＃39;$file&＃39; as (date, web, name, food);

B &＃61; load &＃39;$file&＃39; as (date, web, name, food);

C&＃61; cogroup A by $0, B by $1;

describe C;

illustrate C;

dump C;

cogroup命令中$0和$1&＃xff0c;两个列的内容如果不一样&＃xff0c;就是分别生成两个批次的group&＃xff0c;先按A值分组&＃xff0c;在按B对应的值分组。按A的值分组时&＃xff0c;B对应的为空&＃xff0c;则group中有一个空组{}&＃xff1b;但如果内容一样&＃xff0c;如C&＃61; cogroup A by $1, B by $1;就是生成一个批次的group&＃xff0c;其中包含A和B两个表中所有的等于该值的元组。

COGROUP与join的区别&＃xff1a;自己懒得写&＃xff0c;摘自网络

Join的操作结果是平面的&＃xff08;一组元组&＃xff09;&＃xff0c;而COGROUP的结果是有嵌套结构的。
运行以下命令&＃xff1a;
r1 &＃61; cogroup r_student by classNo,r_teacher by classNo;
dump r1;
结果如下&＃xff1a;
(C01,{(C01,N0103,65),(C01,N0102,59),(C01,N0101,82)},{(C01,Zhang)})
(C02,{(C02,N0203,79),(C02,N0202,82),(C02,N0201,81)},{(C02,Sun)})
(C03,{(C03,N0306,72),(C03,N0302,92),(C03,N0301,56)},{(C03,Wang)})
(C04,{},{(C04,Dong)})
由结果可以看出&＃xff1a;
1&＃xff09;  cogroup和join操作类似。
2&＃xff09;  生成的关系有3个字段。第一个字段为连接字段&＃xff1b;第二个字段是一个包&＃xff0c;值为关系1中的满足匹配关系的所有元组&＃xff1b;第三个字段也是一个包&＃xff0c;值为关系2中的满足匹配关系的所有元组。
3&＃xff09;  类似于Join的外连接。比如结果中的第四个记录&＃xff0c;第二个字段值为空包&＃xff0c;因为关系1中没有满足条件的记录。实际上第一条语句和以下语句等同&＃xff1a;
r1&＃61; cogroup r_student by classNo outer,r_teacher by classNo outer;
如果你希望关系1或2中没有匹配记录时不在结果中出现&＃xff0c;则可以分别在关系中使用inner而关键字进行排除。
执行以下语句&＃xff1a;
r1 &＃61; cogroup r_student by classNo inner,r_teacher byclassNo outer;
dump r1;
结果为&＃xff1a;
(C01,{(C01,N0103,65),(C01,N0102,59),(C01,N0101,82)},{(C01,Zhang)})
(C02,{(C02,N0203,79),(C02,N0202,82),(C02,N0201,81)},{(C02,Sun)})
(C03,{(C03,N0306,72),(C03,N0302,92),(C03,N0301,56)},{(C03,Wang)})

flatten执行命令&＃xff1a;
r2 &＃61; foreach r1 generate flatten($1),flatten($2);
dump r2;
结果如下&＃xff1a;
(C01,N0103,65,C01,Zhang)
(C01,N0102,59,C01,Zhang)
(C01,N0101,82,C01,Zhang)
(C02,N0203,79,C02,Sun)
(C02,N0202,82,C02,Sun)
(C02,N0201,81,C02,Sun)
(C03,N0306,72,C03,Wang)
(C03,N0302,92,C03,Wang)
(C03,N0301,56,C03,Wang)

可以看到&＃xff0c;两个同时flatten&＃xff0c;会自动映射生成多列。

针对cogroup&＃xff0c;我测试了一下&＃xff0c;核心代码如下&＃xff1a;

industry_existed_Data &＃61;  LOAD &＃39;$industryPath&＃39; USING PigStorage(&＃39;,&＃39;) AS (industryId:chararray,guid:chararray,sex:chararray,log_type:chararray);
sample_data &＃61; limit industry_existed_Data 20;
--STORE  sample_data INTO &＃39;/user/wizad/tmp/industry_existed_Data&＃39; USING PigStorage(&＃39;,&＃39;);
--merge with history data 
cogroupIndustryExistCurrentByGuid &＃61; COGROUP industry_existed_Data by guid, industry_current_data by guid;
mydata &＃61; sample cogroupIndustryExistCurrentByGuid 0.1;
dump mydata;
describe cogroupIndustryExistCurrentByGuid;
--dump cogroupIndustryExistCurrentByGuid;

--STORE  mycogroupdata INTO &＃39;/user/wizad/tmp/cogroupIndustryExistCurrentByGuid&＃39; USING PigStorage(&＃39;,&＃39;);

look_for_cogroup &＃61; FOREACH cogroupIndustryExistCurrentByGuid GENERATE $0,$2;
describe look_for_cogroup;

IndustryStorageDataTmp &＃61; FOREACH cogroupIndustryExistCurrentByGuid GENERATE FLATTEN($2);
IndustryStorageData &＃61; DISTINCT IndustryStorageDataTmp;
describe IndustryStorageData;

显示结果&＃xff1a;

三个数据的结构如下

cogroupIndustryExistCurrentByGuid: 
{
group: chararray,
industry_existed_Data:{industryId: chararray,guid: chararray,sex: chararray,log_type: chararray},
industry_current_data: {joined_ad_campaign_data::industryId: chararray,joined_Orgin_sex_data::distinct_origin_historical_sex::guid: chararray,joined_Orgin_sex_data::social_sex::sex: chararray,joined_Orgin_sex_data::distinct_origin_historical_sex::log_type: chararray}
}
look_for_cogroup: 
{
group: chararray,
industry_current_data: {joined_ad_campaign_data::industryId: chararray,joined_Orgin_sex_data::distinct_origin_historical_sex::guid: chararray,joined_Orgin_sex_data::social_sex::sex: chararray,joined_Orgin_sex_data::distinct_origin_historical_sex::log_type: chararray}
}
IndustryStorageData: 
{
industry_current_data::joined_ad_campaign_data::industryId: chararray,
industry_current_data::joined_Orgin_sex_data::distinct_origin_historical_sex::guid: chararray,
industry_current_data::joined_Orgin_sex_data::social_sex::sex: chararray,
industry_current_data::joined_Orgin_sex_data::distinct_origin_historical_sex::log_type: chararray
}

可以看出三个数据的结构很复杂&＃xff0c;因为前面做关联所以包含了对象名&＃xff08;或者叫域名&＃xff09;&＃xff0c;指明属于哪个对象。可以只看最后一列名字和格式。

第三个是flatten&＃xff08;$2&＃xff09;的结果。

cogroup有空集问题&＃xff0c;就是对应group中的每个值&＃xff08;cogroup用来关联的key的取值&＃xff09;&＃xff0c;两个集合各自按key值进行group后&＃xff0c;某些key对应的集合为空。

上面的pig代码的实际数据如下&＃xff0c;guid作为关联key&＃xff0c;可以看出很多空集{}&＃xff0c;出现在某些guid的取值对应集合后。

所以取数据时要注意&＃xff0c;只flatten某一列&＃xff0c;会造成其他列数据丢失&＃xff0c;因为对应着该flatten列的空集。

((-1,),{(74,9051235c-a391-4dae-ab22-f93d24a12636,-1,-1,),(75,053e9f48-03bf-4b39-9455-ff412a725a3c,-1,-1,),(74,21ca723c-ec2b-4242-8108-b95436f10e3e,-1,-1,),(74,fec1932a-b0e4-4bf0-b504-8ed8f3c159e7,-1,-1,),(74,d74374ec-8cf4-4c4a-b598-9631f6972cbb,-1,-1,),(74,6780962a-bf75-4c4c-a557-94a7de5a3e36,-1,-1,),(74,14517915-ee3f-4d34-943f-d6f1813afdef,-1,-1,),(74,c5547aca-3b8b-4108-93ba-bf365c106cdd,-1,-1,),(74,e9a986c1-6868-4f7f-baf6-69d8c302583e,-1,-1,),(74,9c1341cf-45b8-48c6-b699-33b1a4215c66,-1,-1,),(74,f16e6222-a84b-4758-ae71-0613c8f34b29,-1,-1,),(74,47cc25ef-05bc-47f4-a32b-3cddaf0ac22b,-1,-1,),(74,d5c1b6b0-38c3-464b-8cb9-70ced875be5f,-1,-1,),(74,6a4f782a-1f5c-45c0-bb3a-4df25c436be3,-1,-1,),(74,23bb2f0c-d629-479d-800e-b86fc3d6e45c,-1,-1,)})
((a50a17bde79ac018,),{(74,863010025134441,a50a17bde79ac018,863010025134441,)})
((a51779f736cd3f54,),{(74,862949029595753,a51779f736cd3f54,862949029595753,)})
((c7ae5867-3b77-4987-b082-ed3867b5c384,),{(74,353627055387065,c7ae5867-3b77-4987-b082-ed3867b5c384,353627055387065,)})

新建了一个文件test.txt,随便造了几条测试数据(tab键分隔&＃xff0c;是pig默认的分隔方式&＃xff0c;若要以其他分隔&＃xff0c;load的时候load &＃39;test.txt&＃39; using PigStorage(&＃39;,&＃39;) as .... 就可以以逗号分隔)&＃xff1a;

xiaojun 28      上海
yangna  24      兰州

在pig grant shell中输入中文是没法解析的&＃xff0c;会报错&＃xff0c;比如在shell下面输入:

a &＃61; load &＃39;test.txt&＃39; as (name:chararray,age:int,city:chararray);

b &＃61; filter a by city &＃61;&＃61; &＃39;上海&＃39;; //此句执行会报错。

dump b;

如果需要使用到中文进行数据集的过滤&＃xff0c;可以将语句写到一个单独的pig脚本中,例如我们新建一个test.pig文件&＃xff0c;把上面两句写入到该文件中去,然后再执行pig -x local test.pig,这时可以成功的执行中文字符的过滤语句。

一般情况下&＃xff0c;都会有定期运行的pig脚本&＃xff0c;比如每天运行一次的脚本&＃xff0c;这类脚本往往在内部都要用到当天的日期作为参数&＃xff0c;pig支持参数替换&＃xff0c;参数由前缀$字符来标示&＃xff0c;例如我们需要load每天的数据进行统计分析&＃xff0c;每天的数据在hdfs上是按日期进行进行命名的&＃xff0c;我们可以新建一个test.pig脚本&＃xff1a;

a &＃61; load &＃39;$input&＃39; as (.....);

store a into &＃39;$output&＃39;;

然后我们在命令行可以这样&＃xff1a;pig -param input&＃61;/user/tom/input/2014-12-01.dat -param output&＃61;/user/tom/output/2014-12-01.dat -f test.pig

test.pig中会把$标示符指定的参数替换为-param指定的参数值。

更常用的是通过动态参数来替换&＃xff0c;即shell脚本中经常会使用到的反引号引用的命令。

我们可以把2014-12-01这段通过shell脚本动态输出&＃xff1a;

pig -param input&＃61;/user/tom/input/&＃96;date "&＃43;%Y-%m-%d"&＃96;.dat -param output&＃61;/user/tom/output/&＃96;date "&＃43;%Y-%m-%d"&＃96;.dat -f test.pig

这样就实现了参数的完全动态替换。

假设文件test4.txt有这么两行数据&＃xff1a;

1980080113312121212018
1985080113313131313023

规则是前8位为年月日&＃xff0c;中间11位为手机号码&＃xff0c;后3位表示的是年龄。

我们可以自定义一个加载udf来加载这个文件

[java] view plain copy

1. package com.besttone.pig.udf.load;  
2.   
3. import java.io.IOException;  
4. import java.util.List;  
5.   
6. import org.apache.commons.logging.Log;  
7. import org.apache.commons.logging.LogFactory;  
8. import org.apache.hadoop.io.Text;  
9. import org.apache.hadoop.mapreduce.InputFormat;  
10. import org.apache.hadoop.mapreduce.Job;  
11. import org.apache.hadoop.mapreduce.RecordReader;  
12. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;  
13. import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;  
14. import org.apache.pig.LoadFunc;  
15. import org.apache.pig.backend.executionengine.ExecException;  
16. import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;  
17. import org.apache.pig.data.DataByteArray;  
18. import org.apache.pig.data.Tuple;  
19. import org.apache.pig.data.TupleFactory;  
20.   
21. public class CutLoadFunc extends LoadFunc {  
22.   
23.     protected static final Log log &＃61; LogFactory.getLog(CutLoadFunc.class);  
24.     private final List ranges;  
25.   
26.     private RecordReader reader;  
27.   
28.     private final TupleFactory tupleFactory &＃61; TupleFactory.getInstance();  
29.   
30.     public CutLoadFunc(String cutPattern) throws IOException {  
31.         ranges &＃61; Range.parse(cutPattern);  
32.     }  
33.   
34.     &＃64;Override  
35.     public void setLocation(String location, Job job) throws IOException {  
36.         // TODO Auto-generated method stub  
37.         FileInputFormat.setInputPaths(job, location);  
38.     }  
39.   
40.     &＃64;Override  
41.     public InputFormat getInputFormat() throws IOException {  
42.         // TODO Auto-generated method stub  
43.         return new TextInputFormat();  
44.     }  
45.   
46.     &＃64;Override  
47.     public void prepareToRead(RecordReader reader, PigSplit split)  
48.             throws IOException {  
49.         // TODO Auto-generated method stub  
50.         this.reader &＃61; reader;  
51.     }  
52.   
53.     &＃64;Override  
54.     public Tuple getNext() throws IOException {  
55.         // TODO Auto-generated method stub  
56.   
57.         try {  
58.             if (!reader.nextKeyValue()) {  
59.                 return null;  
60.             }  
61.             Text value &＃61; (Text) reader.getCurrentValue();  
62.             String line &＃61; value.toString();  
63.   
64.             Tuple tuple &＃61; tupleFactory.newTuple(ranges.size());  
65.             for (int i &＃61; 0; i < ranges.size(); i&＃43;&＃43;) {  
66.                 Range range &＃61; ranges.get(i);  
67.                 if (range.getEnd() > line.length()) {  
68.                     log.warn(String.format(  
69.                             "Range end (%s) is longer than line length (%s)",  
70.                             range.getEnd(), line.length()));  
71.                     continue;  
72.                 }  
73.                 tuple.set(i, new DataByteArray(range.getSubstring(line)));  
74.   
75.             }  
76.             return tuple;  
77.   
78.         } catch (InterruptedException e) {  
79.             throw new ExecException(e);  
80.         }  
81.   
82.     }  
83.   
84. }  

[java] view plain copy

1. package com.besttone.pig.udf.load;  
2.   
3. import java.io.IOException;  
4. import java.util.ArrayList;  
5. import java.util.List;  
6.   
7. import org.apache.commons.logging.Log;  
8. import org.apache.commons.logging.LogFactory;  
9.   
10. public class Range {  
11.   
12.     protected static final Log log &＃61; LogFactory.getLog(Range.class);  
13.     private int start;  
14.     private int end;  
15.   
16.     public int getStart() {  
17.         return start;  
18.     }  
19.   
20.     public void setStart(int start) {  
21.         this.start &＃61; start;  
22.     }  
23.   
24.     public int getEnd() {  
25.         return end;  
26.     }  
27.   
28.     public void setEnd(int end) {  
29.         this.end &＃61; end;  
30.     }  
31.   
32.     public static List parse(String cutPattern) throws IOException {  
33.   
34.         String[] rangelist &＃61; cutPattern.split(",");  
35.         if (rangelist.length &＃61;&＃61; 0) {  
36.             throw new IOException("cutPattern参数不合法");  
37.         }  
38.         List list &＃61; new ArrayList(rangelist.length);  
39.         for (String rangestr : rangelist) {  
40.             try {  
41.                 Range range &＃61; new Range();  
42.                 range.setStart(Integer.parseInt(rangestr.split("-")[0]));  
43.                 range.setEnd(Integer.parseInt(rangestr.split("-")[1]));  
44.                 list.add(range);  
45.             } catch (Exception e) {  
46.                 throw new IOException("cutPattern参数不合法");  
47.             }  
48.         }  
49.         return list;  
50.   
51.     }  
52.   
53.     public String getSubstring(String line) {  
54.         if (this.end > line.length()) {  
55.             log.warn(String.format(  
56.                     "Range end (%s) is longer than line length (%s)", this.end,  
57.                     line.length()));  
58.             return null;  
59.         }  
60.         return line.substring(this.start, this.end);  
61.     }  
62.       
63.     public static void main(String[] args) {  
64.         Range range &＃61; new Range();  
65.         range.setStart(19);  
66.         range.setEnd(22);  
67.         System.out.println(range.getSubstring("1980080113312121212018"));  
68.     }  
69. }  

将这两个类打成jar包,然后进入pig grunt,执行一下脚本

register besttonePigUDF.jar ;

a&＃61; load &＃39;test4.txt&＃39; using com.besttone.pig.udf.load.CutLoadFunc(&＃39;0-8,8-19,19-22&＃39;) as (date:chararray,phone:chararray,age:chararray);

dump a;

可以看到内容被成功的加载&＃xff1a;

(19800801,13312121212,018)
(19850801,13313131313,023)