pig实战: pig常用语法总结

阅读导读：
1.pig中表、字段、一条数据与bag、tuple、field是如何对应的？
2.pig的tuple字段数和数据库一样吗？
3.grunt模式下如何快速跳到行首和行末？
4.Pig中的各operator（操作符），哪些会触发reduce过程？
5.如何使用Pig？
6.pig中cogroup的作用？







本文主要来源于google资料的自我总结，对常用的pig语法即“算法”进行了总结。

基础概念：

　　relation bag tuple field data
　　关系(relation) = 包(bag)
　　一个包是一个元组(tuple）的集合,在pig中用{}扩起来表示一个包
　　一个元组(tuple)是若干个字段的有序集合(order set),在pig的数据结构中，用()扩起来标识一个元组
　　一个字段(field)是列数据(data)的标识；
　　和数据库的对应关系：
　　　　pig database
　　　　relation/bag table
　　　　tuple one record
　　　　field field(one volume)
　　但是pig中tuple的字段数是随意的，这点和数据库不同。


运行模式以及注释：

1、 运行模式
　　　　a) 本地
　　　　　　i. pig –x local
　　　　b) 集群
　　　　　　i. pig –x mapreduce
　　　　　　ii. 或者pig
　　　　c) 批处理命令
　　　　　　i. 将命令批量写入xx.pig
　　　　　　ii. 用本地或者集群模式运行，如pig xx.pig
2、注释
　　　　a) 段落注释：/**/
　　　　b) 行注释: --


基本用法框架：
　　a) 输入
　　　　i. A = Load ‘输入路径’ USING PigStorage(“\t”) AS (name:chararray, age:int, gpa:float);
　　　　ii. 逻辑：用pig自带的PigStorage读入输入路径的数据，对每一行用”\t”分割，并用字符串类型的name、int类型的age、float类型的gpq作为每行数据的字段名
　　b) 中间处理
　　　　i. B = FOREACH A GENERATE name;
　　c) 输出
　　　　i. DUMP B; --输出到控制台
　　　　ii. 或者STORE B INTO ‘输出路径’ USING PigStorage();


基本语法实例：

1、计算多维度组合下的平均值 的实际例子
　　　　　　需求：求文件中第2、3、4组合的第4、5列的平均值
　　　　　　脚本解释：
　　　　　　　　A = Load '1.txt' USING PigStorage(' ') AS(col1:chararray, col2:int, col3:int, col4:int, col5:double, col6:double);
　　　　　　　　将1.txt中的每行数据用" "分割，然后安装(col1:chararray, col2:int, col3:int, col4:int, col5:double, col6:double)解析数据，并装载到每个bag/tuple/filed
　　　　　　　　col1是每列的别名，如果不指定，在后面可以用$0,$n来索引，但是可读性较差

　　　　　　　　A的结构是{col1:chararray, col2:int, col3:int, col4:int, col5:double, col6:double)}

　　　　　　　　B = GROUP A BY (col2, col3, col4);
　　　　　　　　对A用col2/col3/col4的组合进行分组，然后按组将每条tuple汇集成一个bag，
　　　　　　　　B的数据结构是B:{group:(col2,col3,col4),A:bag{:tuple,tuple}}
　　　　　　　　C = FOREACH B GENERATE group, AVG(col5), AVG(col6);
　　　　　　　　FOREACH是遍历每个组中汇集的tuple，并对其用组合函数处理字段
　　　　　　　　C的结构是C:{group:(col2,col3,col4),double,double}
　　　　　　　　DUMP C；//单机运行打印调试信息
　　　　　　　　STORE C INTO “output”;//存储结果

2、统计行数
　　　　　　　A = LOAD '1.txt' USING PigStorage (' ‘) AS(col1:chararray, col2:int, col3:int, col4:int, col5:double, col6:double);
　　　　　　　B = GROUP A all;
　　　　　　　C = FOREACH B GENERATE COUNT(col2);
　　　　　　　DUMP C;
　　　　　　　注意：如果col2这一列有NULL，则这一行将不会被统计。
　　　　　　　　GROUP A all;没有by关键字。
　　　　　　　　If you want to include NULL values in the count computation, use COUNT_STAR.

3、FLATTEN
　　　　　　同1中的例子，但是C不同
　　　　　　C = FOREACH B GENERATE FLATTEN(group), AVG(col5), AVG(col6);
　　　　　　1中的例子得到的结果结构是(col2, col3, col4),avg1,avg2
　　　　　　但是对group作用FLATTEN后结构变为col2,col3,col4,avg1,avg2；可以看出FLATTEN的作用是解嵌套(un-nest)包或者元组tuple，

4、GROUP的两点说明：
　　　　　　a)用于GROUP的key如果多于一个字段（正如本文前面的例子），则GROUP之后的数据的key是一个元组（tuple），否则它就是与用于GROUP的key相同类型的东西

　　　　　　b)GROUP后得到的结果是一个包或者是relation,其包含的每一个tuple中包含两个字段，第一个字段被命名为'group'，第二个字段是一个包，是含有group对应值的所有tuple的set

5、把数据作为tuple来加载
　　　　　　语法：A = LOAD '1.txt' AS (T: tuple(col1:chararray, col2:int, col3:int,col4:int, col5:double, col6:double);
　　　　　　用describe A；可以查看A的结构。
　　　　　　注意：输入的数据必须是(xx,xx,xx...xx)的结构才能以元组的形式被加载，适用于中间结果的进一步处理。

6、计算多维组合中不重复记录的条数
　　　　　　C = FOREACH B GENERATE (D = DISTINCT col5; group, COUNT(D));
　　　　　　说明：distinct后跟一个字段即可，如果这个字段不同，则肯定不重复，而如果重复，此字段肯定相同，而且重命名D要用等号，后面跟；

7、将relation转为表量scalar。

8、在pig中使用shell脚本进行辅助处理
　　　　　　A = LOAD ‘2.txt' AS (col1:int, col2:int, col3:int, col4:chararray, col5:int);
　　　　　　B = STREAM A THROUGH `awk '{if($4 == "=") print $1"\t"$2"\t"$3"\t"999"\t"$5; else print $0}'`;
　　　　　　DUMP B；
　　　　　　作用：将第4列中的"="换成999

9、向pig脚本传参：
　　　　　　语法：pig -param output_dir="/xxx" xxx.pig;
　　　　　　而在xxx.pig中STORE D INTO ‘$output_dir’；

10、COGROUP的作用：
　　　　　　对两个关系中的包，分别按制定的字段进行分组
　　　　　　A = LOAD '3.txt' AS (acol1:chararray, acol2:int, acol3:int);
　　　　　　B = LOAD '4.txt' AS (bcol1:int, bcol2:chararray, bcol3:int);
　　　　　　C = COGROUP A BY A.acol1, B BY B.bcol2;
　　　　　　DUMP C;
　　　　　　结果的结构{key,{A's one tuple or null},{B's ont tuple or null}};
　　　　　　按A的第一行和B的第二行的值，分别对A、B进行分组组合，如果A中的tuple含有此key则显示在结果中，如果没有则为{},B也一样。

11、piggybank
　　　　　　自定义函数的名称为piggybank。


12、UDF的构造函数会被调用多次，所以在其中做一次工作时要特别小心。

13、LOAD多个目录下的数据
　　　　　　LOAD '/data/201{1,2}'
　　　　　　load/data/2011 /data/2012两个目录下的数据。

14、GROUP 函数不会忽略NULL值，而COUNT会。

15、统计某些字段组合的种数：
　　　　　　A = LOAD '15.txt' AS (col1:int, col2:int);
　　　　　　B = GROUP A BY (col1, col2);
　　　　　　C = GROUP B ALL;
　　　　　　D = FOREACH C GENERATE COUNT(B);
　　　　　　DUMP D;
　　　　　　思路：先根据需求对数据进行分组，然后对分组COUNT；
　　　　　　注意：
　　　　　　　　a)D中COUNT的是B，因为通过C将所有的数据组为一个新的关系，只包含一个tuple，第一个字段为ALL，第二个字段是B的所有tuple组成的一个bag。
　　　　　　　　b)COUNT作用的对象必须是一个bag，所以在统计字段前，要用GROUP X ALL,将X中所有的tuple组成一个bag。


16、两个整数相除，如何得到一个float
　　　　　　A = LOAD '16.txt' AS (col1:int, col2:int);
　　　　　　B = FOREACH A GENERATE (float)col1/col2;
　　　　　　DUMP B;
　　　　　　注意先转型在计算，而不是(float)(col1/col2);

17、UNION两个数据进行统计
　　　　　　A = LOAD '1.txt' AS (col1:int, col2:int);
　　　　　　B = LOAD '2.txt' AS (col3:int, col4:int);
　　　　　　C = UNION A, B;
　　　　　　D = GROUP C BY $0;
　　　　　　E = FOREACH D GENERATE group, SUM(C.$1);
　　　　　　#E = FOREACH D GENERATE FLATTEN(group), SUM(C.$1);
　　　　　　DUMP E;
　　　　　　注意：
　　　　　　　　UNION操作是给关系添加新的tuple，而且UNION后注意观测filed的引用方法，见结构篇
　　　　　　　　多个关系时，对列的操作要指定关系名

18、正则表达式过滤
　　　　　　过滤出符合*//*.qq.com/*的url
　　　　　　A = LOAD '18.txt' AS (col1:int, col2:chararray);
　　　　　　B = FILTER A BY col2 matches '.*//.*\\.qq\\.com/.*';
　　　　　　说明：.*标识至少一个字符
　　　　　　而匹配.字符要进行转义\.，而在''内转义要用两个\\. ;


19、截取字符串：
　　　　　　SUBSTRING(col1, 0, 4):0为startIndex,4为stopIndex，不包含stopIndex;

　　　　　　A = LOAD '19.txt' AS (dataStr:chararray, col2:chararray);
　　　　　　B = FOREACH A GENERATE SUBSTRING(dataStr,0,4);
　　　　　　C = DISTINCT B;
　　　　　　DUMP C;

20、连接字符串：
　　　　　　A = LOAD ‘20.txt' AS (col1:chararray, col2:int);
　　　　　　B = FOEACH A GENERATE CONCAT(col1, (chararray)col2);
　　　　　　DUMP B;
　　　　　　注意：连接的字段类型必须是chararray，如果不是要转型
　　　　　　　　嵌套使用：CONCAT(A,CONCAT(B,C);

21、用JOIN求两个数据集的交集&不同数据类型交集会失败
　　　　　　A = LOAD '211.txt' AS (a:int);
　　　　　　B = LOAD '212.txt' AS (b:int);
　　　　　　C = JOIN A BY a, B BY b;
　　　　　　D = GROUP C ALL;
　　　　　　E = FOREACH D GENERATE COUNT(C);
　　　　　　DUMP E;
　　　　　　JOIN后C的结构为：C:{A::a:int, B::b:int}
　　　　　　去重：
　　　　　　　　A = LOAD '211.txt' AS (a:int);
　　　　　　　　B = LOAD '212.txt' AS (b:int);
　　　　　　　　C = JOIN A BY a, B BY b;
　　　　　　　　uniq_C = DISTINCT(C);
　　　　　　　　D = GROUP uniq_C ALL;
　　　　　　　　E = FOREACH D GENERATE COUNT(C);
　　　　　　　　DUMP E;


22、三目运算符使用必须使用():用来替换空值
　　　　　　B = FOREACH A GENERATE ((col1 is null) ? -1 : col1)

23、如何在得到计算结果后，补齐空白
　　　　　　A = LOAD ‘23.txt' AS (col1:int, b:tuple(col2:int, col3:int);
　　　　　　B = FOREACH A GENERATE col1, FLATTEN(b);
　　　　　　C = GROUP B BY B.col1;
　　　　　　D = FOREACH C GENERATE group, SUM(B.col1) AS sum;
　　　　　　E = FOREACH D GENERATE group, ((sum is null) ? 0 : sum);
　　　　　　DUMP E;


24、DISTINCT操作用于去重
正因为它要把数据集合到一起，才知道哪些数据是重复的，因此，它会产生reduce过程。同时，在map阶段，它也会利用combiner来先去除一部分重复数据以加快处理速度。


25、提高Pig job的优先级：
set job.priority HIGH;提高Pig job的运行速度

26、“Scalars can be only used with projections”错误和嵌套/inner FOREACH
　　　　　　在第一列的每种组合中，第二列为3/6的数据分别有多少条

　　　　　　A = LOAD ’26.txt' AS (col1:chararray, col2:int);
　　　　　　B = GROUP A BY col1;
　　　　　　C = FOREACH B {
　　　　　　　　D = FILTER A BY col2 == 3;
　　　　　　　　E = FILTER A BY col2 == 6;
　　　　　　GENERATE group, COUNT(D), COUNT(E)；};
　　　　　　DUMP C;


27、在grunt模式下按Ctrl+A 和 Ctrl+E 代替 HOME 和 END，就可以跳到行首和行末了

28、同一个关系进行JOIN连接必须导入两次，做连接，否则出错

29、外链接JOIN
　　　　　　LEFT：左边的数据全量显示

　　　　　　A = LOAD '291.txt' AS (col1:int, col2:chararray);
　　　　　　B = LOAD '292.txt' AS (col1:int, col2:chararray);
　　　　　　C = JOIN A BY col1 LEFT, B BY col1;
　　　　　　DESCRIBE C;
　　　　　　DUMP C;
　　　　　　这个和数据库的左右连接和内链接一致

30、pig中支持过滤中文，但是在交互模式下不行

31、统计 tuple 中的 field 数，bag 中的 tuple 数，map 中的 key/value 组数用SIZE函数

32、字符此为null 用col is null来判断，但是不能过滤" "，" " 过滤要用SIZE(xx)>0
　　              FILTER A BY (col1 is not null AND (SIZE(col2) > 0));

33、Pig中的各operator（操作符），哪些会触发reduce过程

                     GROUP：由于GROUP操作会将所有具有相同key的记录收集到一起，所以数据如果正在map中处理的话，就会触发shuffle→reduce的过程。
　　              ORDER：由于需要将所有相等的记录收集到一起（才能排序），所以ORDER会触发reduce过程。同时，除了你写的那个Pig job之外，Pig还会添加一个额外的M-R job到你的数据流程中，因为Pig需要对你的数据集做采样，以确定数据的分布情况，从而解决数据分布严重不均的情况下job效率过于低下的问题。

　　              DISTINCT：由于需要将记录收集到一起，才能确定它们是不是重复的，因此DISTINCT会触发reduce过程。当然，DISTINCT也会利用combiner在map阶段就把重复的记录移除。

　　　          JOIN：JOIN用于求重合，由于求重合的时候，需要将具有相同key的记录收集到一起，因此，JOIN会触发reduce过程。

　　　          LIMIT：由于需要将记录收集到一起，才能统计出它返回的条数，因此，LIMIT会触发reduce过程。

　　　          COGROUP：与GROUP类似（参看本文前面的部分），因此它会触发reduce过程。

　　　          CROSS：计算两个或多个关系的叉积。


34、如何统计一个字符串中包含的指定字符数
　　　          shell 脚本 awk -F ":" '{print NF - 1}'
　　　          以:分割字符串，然后打印总的volume数