火龙果软件-PIG学习记录

1、PIG学习记录PIG是什么？Pig提供了一个基于Hadoop的并行地执行数据流处理的引擎。Pig运行于Hadoop之上，它同时使用到Hadoop分布式文件系统HDFS和Hadoop处理系统MapReduce。数据类型基本数据类型：l int：整型，4字节l long：长整型，8字节l float：单精度浮点型，4字节l double：双精度浮点型，8字节l chararray：字符串l bytearray：字节串复杂数据类型：l Map：map是一种chararray和数据元素之间的键值对映射，其中数据元素可以是任意的pig类型，包括复杂类型。通常使用方括号划定map结构name# bob,

2、age#55l Tuple：tuple是一个定长的，包含有序pig数据元素的集合，一个tuple相当于SQL中的一行，而tuple字段相当于SQL中的列。Tuple常量使用圆括号来指示tuple结构，使用逗号划分字段(bob,55)l Bag：bag是一种无序的tuple集合。Bag常量通过花括号进行划分(bob,55),(sally,52),(john,25)NULL：表示这个值是未知的模式：加载数据时显示声明模式：dividends = load NYSE_dividends as (exchange:chararray,symbol:chararray,date:chararray,di

3、vidend:float);使用加载函数定义模式：mdata = load mydata using HCatLoader();猜测模式：calcs = foreach daily generate $7/1000,$3*100.0,SUBSTRING($0,0,1),$6-$3;根据运算符两端数据类型和函数参数的数据类型猜测出模式不明的变量的模式数据类型语法举例intintas(a:int)longlongas(a:long)floatfloatas(a:float)doubledoubleas(a:double)chararraychararrayas(a:chararray)bytear

4、raybytearrayas(a:bytearray)mapmap或者maptype,其中type必须是一个合法的数据类型。这就声明了map中所有的值都是这个数据类型的as(a:map,b:mapint)tupletuple()或者tuple(list_of_fields)，其中list_of_fields是一组通过逗号分隔的字段声明as(a:tuple(),b:tuple(x:int,y:int)bagbag或者bagt:list_of_fields)，其中list_of_fields是一组通过逗号分隔的字段声明。需要注意的是，bag中的tuple必须要有一个名称，这里指定的名称为t，虽然用

5、户可能永远无法直接访问t这个tuple(a:bag,b:bagt:(x:int,y:int)类型转换intlongfloatdoublechararrayint可以可以可以可以long可以但会被截断可以可以可以float可以但会被截断可以但会被截断可以可以double可以但会被截断可以但会被截断可以但会被截断可以chararray可以但非数值字符会转换为null可以但非数值字符会转换为null可以但非数值字符会转换为null可以但非数值字符会转换为null显式转换：unitended = foreach player generate (int)goal;隐式转换：rough = foreac

6、h daily generate volume*close;（其中volume为int，close为float，会将volume转换为float）加载直接加载文件：divs = load /data/examples/NYSE_dividends;通过加载函数加载：divs = load NYSE_dividends using HBaseStorage();默认加载函数：divs = load NYSE_dividends using PigStorage(,);存储直接存储文件：store processed into /data/examples/processed;通过加载函数加载：s

7、tore processed into processed using HBaseStorage();默认加载函数：store processed into processed using PigStorage(,);输出：dump processed;关系操作：foreach：接受一组表达式，然后在数据管道中将它们应用到每条记录中B = foreach A generate user,id;gain = foreach prices generate $6-$3;beginning = foreach prices generate open.close;与group by语句结合进行统计B

8、 = group A1 by x;C = foreach B generate SUM(A1.yz);使用用户自定义函数（UDF）Upped = foreach divs generate UPPER(symbol) as symbol,dividends;describe：显示字段模式describe sym;filter：过滤流记录项startswithcm = filter divs by symbol matches CM.*;startswithcm = filter divs by not symbol matches CM.*;*x = null会返回nullgroup：可以把具

9、有相同键值的数据聚合在一起进行分组grpd = group daily by stock;cnt = foreach grpd generate group,COUNT(daily);grpd = group daily all;-对所有字段进行分组cnt = foreach grpd generate COUNT(daily);order by：对用户的数据进行排序，产生一个全排序的输出结果bydate = order daily by date;bydatensymbol = order daily by date,symbol;使用desc进行降序排序Byclose = order da

10、ily by close desc,open;-open还是升序排序distinct：去除重复值uniq = distinct daily;join：根据键值联合数据，没有被匹配到的数据会被去掉jnd = join daily by symbol,divs by symbol;jnd = join daily by (symbol,date),divs by (symbol,date);outer join:jnd = join daily by (symbol,date) left outer,divs by (symbol,date);jnd = join daily by (symbol

11、,date) right,divs by (symbol,date);jnd = join daily by (symbol,date) full,divs by (symbol,date);*自联需要加载同一数据两次Limit：从结果中拿出几条数据，会额外产生一个reduce过程first10 = limit divs 10;Sample：抽取样本数据，返回一定百分比的行数的数据，通过一个0到1的double指定some = sample divs 0.1Parallel：附加到任一个关系操作符后面控制reduce阶段的并行，可以触发reduce的操作符有group、order、distin

12、ct、join、cogroup和crossbysymbl = group daily by symbol parallel 10;*通过set default_parallel设置脚本范围有效的并行数注册UDF：register your_path_to_piggybank/piggybank,jar;把函数所在的jar包导入进来register production.py using jython as bballudfs;注册Python函数，Python脚本必须在用户当前目录下define：可用于为用户的java UDF定义一个别名define reverse org.apache.pi

13、g.piggybank.evaluation.string.Reverse();define convert com.acme.financial.CurrencyConverter(dollar,euro);调用静态Java函数InvokeForInt、InvokeForLong、InvokeForFloat、InvokeForDouble和InvokeForString对应相应的返回值的函数，第1个参数是完整的包名、类名和方法名，第2个参数是一个以空格作为分割符的参数列表，如果这个参数是个数组，会跟在这个类型名称后面，如果不需要参数则省略第2个参数define hex InvokeForS

14、tring(java.lang.Integer.toHexString, int);inhex = foreach nonnull generate symbol,hex(int)volume);define stdev InvokeForDouble(com.acme.Stats.stdev, double);A = load input as (id:int,dp:double);B = group A by id;C = foreach B generate group,stdev(A.dp);Pig Latin的高级应用高级关系操作foreach结合flatten进行分组统计B = g

15、roup A by a,b,c;C = foreach B generateflatten(group:a) as a,flatten(group:b) as b,flatten(group:c) as c,COUNT(*) as num;计算出每一个分组的记录数内嵌foreachgrpd = group daily by exchange;uniqcnt = foreach grpgsym = daily.symbol;uniq_sym = distinct sym;generate group,COUNT(uniq_sym);*sym = daily.symbol 相等于sym = for

16、each grpd generate daily.symbol理论上任何一个Pig Latin关系操作符在foreach语句内都是合法的，但是目前只支持distinct、filter、limit和order该功能可用于从一组记录中找出前k个元素grpd = group divs by symbol;top3 = foreach grpdsorted = order divs by dividends desc;top = limit sorted 3;generate group,flatten(top);使用不同的join实现方法当进行Join时，最后一个表不会放入内存，而是以stream的

17、方式进行处理，所以最好把最大的一个表放置到Join语句的最后。 Pig实现了以下三种定制的Join以进一步优化。 1) Replicated Join（分片-冗余join） 当进行Join的一个表比较大，而其他的表都很小(能够放入内存)时，Replicated Join会非常高效。Replicated Join会把所有的小表放置在内存当中，然后在Map中读取大表中的数据记录，和内存中存储的小表的数据进行Join，得到Join结果，无需Reduce。可以在Join时使用 Using 'replicated'语句来触发Replicated Join，大表放置在最左端，其余小表(可以

18、有多个)放置在右端。jnd = join daily by (exchange,symbol),divs by (exchange,symbol) using replicated;*第2个输入即divs会被加载到内存中 2) Skewed Join 当进行Join的两个表中，一个表数据记录针对key的分布极其不均衡的时候，简单的使用Hash来分配Reduce端的key时，可能导致某些Reducer上的数据量特别大，降低整个集群的性能。Skewed Join可以首先对左边的表的key统计其分布，然后决定Reduce端的key的分布，尽量使得Reduce端的数据分布比较均衡。可以在Join时使用

19、Using 'skewed'语句来触发Skewed Join，需要进行统计的表(亦即key可能分布不均衡的表)放置在左端。jnd = join cinfo by city,users by city using skewed;*join连接的第2个输入将会被进行样本抽样并且会将对应的记录的条数非常大的键分拆到各个reducer中。第1个输入会对包含这些值的进行冗余复制然后分发到各个reducer中。 3) Merge Join 当进行Join的两个表都已经是有序的时，可以使用Merge Join。Join时，首先对右端的表进行一次采样，对采样的数据创建索引，记录(key, 文件

20、名, 偏移offset)。然后进行map，读取Join左边的表，对于每一条数据记录，根据前一步计算好的索引来查找数据，进行Join。可以在Join时使用Using 'merge'语句来触发Merge Join，需要创建索引的表放置在右端。jnd = join daily by symbol,divs by symbol using merge;*对于原本无序的表而言，由于要先对表进行排序，所以总体效率并不比普通的join高多少另外，在进行Join之前，首先过滤掉key为Null的数据记录可以减少Join的数据量。Cogroup：是group的一般化方式，不是基于一个键收集一个输

21、入的记录，而是基于一个键收集多个输入的记录。C = cogroup A by id,B by id;C:group:int,A:id:int,val:float,B:id:int,val2:intcogroup可以用于进行一个semi-join的操作grpd = cogroup daily by (exchange,symbol),divs by (exchange,symbol);sjnd = filter grpd by not IsEmpty(divs);final = foreach sjnd generate flatten(daily);*semi_join就是当第1个表的键在第2

22、个表中匹配成功时则返回第1个表的记录，以上操作的原理是先使用要进行join的键对两个表进行收集分组并判断出分组中第2个表的数据不为空的分组，最后对第1个表的数据进行解包。Union：将两个表合并到一起C = union A,B;Pig中的union并不要求两个表具有相同的模式，如果两个表的字段类型可以通过隐式转换则union之后的表的字段类型将会是转换后的类型，如果无法进行隐式转换1则结果表将会是没有模式的如果不希望结果表没有模式可以使用onschemaC = union onschema A,B;在读取数据的过程中，如果给定的输入中没有相关的字段，那么值用null填补Cross：第1个表的每

23、条记录都会与第2个表的每条记录结合tonsodata = cross daily,divs parallel 10;*cross常常会产生大量的数据。如果给定的输入时n和m条记录，那么cross的输出会产生n*m条记录cross可以用来进行非等值join（theta join）crossed = cross daily,divs;tjnd = filter crossed by daily:date < divs:date;*上述操作结果为daily中每条记录都和divs表中所有date更大的记录进行连接在Pig中集成遗留代码和MapReduce程序stream：在流中使用其他程序处理数

24、据define hd highdiv.pl ship(highdiv.pl);highdivs = stream divs through hd as (exchange,symbol,date,dividends);*stream命令中的as语句是非必须的，但是如果没有提供as语句那么结果将没有模式。所使用的可执行文件在每个map任务或者reduce任务中都会被调用一次，不会对于每条记录都调用一次。Pig实例化这个可执行文件然后保证数据持续输入是通过标准输入stdin实现的。同时也会持续检查标准输出stdout，将所得的任意结果传送给数据流中的下一个操作符。这个可执行文件可以选择是否对每个输

25、入产生一个输出，或者没一定数量的输入产生一个输出，或者只有当所有的输入都被接收后才会产生一个输出以上操作中定义将使用的可执行文件highdiv.pl的别名为hp并通过ship命令加载highdiv.pl到Hadoop中作为任务的一部分define blc blacklistchecker.py cache(/data/shared/badurls#badurls);goodurls = stream normalized through blc as (url,pageid);*cache使用Hadoop的分布式缓存，#号前面的部分是HDFS上的路径，#后面的字符串是可执行文件将读取的文件别名

26、，Hadoop将会把/data/shared/badurls复制到各个task机器的工作目录下啊并命名为badurls mapreduce：可以在数据流中直接添加MapReduce任务goodurls = mapreduce blacklistchecker.jarstore normalized into inputload output as (url,pageid)com.acmeweb.security.BlackListChecker I input o output;*mapreduce命令的第1个参数是包含MapReduce任务的JAR文件，store定义了数据如何流入到MapR

27、educe中的，load获取MapReduce中的输出，需要指定模式，第3个参数是可选的，用于传递给Java命令的参数split：显式地对数据流进行划分split wlogs into apr03 if timestamp < 20110404,apr02 if timestamp < 20110403 and timestamp > 20110401,apr01 if timestamp < 20110402 and timestarmp > 20110331;store apr03 into 20110403;store apr02 into 20110402

28、;store apr01 into 20110401;*相当于使用多个filter对数据流进行筛选，实际上pig会将其重写为这样执行过程控制set：用于设置pig执行MapReduce任务的环境变量Pig提供的set参数参数值数据类型描述debugstring设置日志级别是DEBUG。与在命令行指定-debug DEBUG效果是一样的default_parallelinteger为脚本中所有的reduce操作设置一个缺省的并发级别。string为Hadoop任务分配一个名称。默认情况下这个名称和所执行的脚本的文件名是一样的，或者是一个为交互式进程随机生成的名称job.prior

29、itystring如果Hadoop集群是使用能力调度器对队列进行优先级控制，那么这个参数可以为Pig任务设置优先级。提供的值有5个：very_low、low、normal、high和very_highPig Latin预处理器参数传入参数可以通过命令行传入或者通过一个参数文件传入yesterday = filter daily by date = $DATE;命令行需要传入参数pig p DATE=2009-12-17 daily.pig也可以通过-m或者-param_file命令指定参数文件pig para_file daily.params daily.pig传入文件内的格式是“参数=值”

30、的形式YEAR=2009-MONTH=12-DAY=17DATE=$YEAR$MONTH$DAY也可以在脚本内使用%declare和%default标记定义参数%default parallel_factor 10;grp = group wlogs by pageid parallel $parallel_factor;宏通过define语句定义：define dividend_analysis (daily,year,daily_symbol,daily_open,daily_close)returns analyzeddivs = load NYSE_dividends as (exch

31、ange:chararray,symbol:chararray,date:chararray,dividends:float);divsthisyear = filter divs by date matches $year-.*;dailythisyear = filter $daily by date matches $year-.*;jnd = join divsthisyear by symbol,dailythisyear by $daily_symbol;$analyzed = foreach jnd generate dailythisyear:$daily_symbol,$da

32、ily_close-$daily_open;daily = load NYSE_daily as (exchange:chararray,symbol:chararray,date:chararray,open:float,high:float,low:float,close:float,volume:int,adj_close:float);results = dividend_analysis(daily,2009,symbol,open,close);包含其他Pig Latin脚本Import命令用于将某个Pig Latin脚本加入另一个脚本中import ./examples/ch6/

33、dividend_analysis.pig;daily = load NYSE_daily as (exchange:chararray,symbol:chararray,date:chararray,open:float,high:float,low:float,close:float,volum:int,adj_close:float);results = dividend_analysis(daily,2009,symbol,open,close);开发与测试Pig Latin脚本在命令行使用-c或者-check选项对脚本进行语法检查而不会实际执行，只会对脚本进行解析和执行语法检查。使用

34、选项-dryruun也会检查语法，然会展开宏命令和import部分，以及进行参数替换。describe命令会显示脚本中指定关系的模式describe trimmed;explain用于检测Pig是如何处理不同脚本的。有两种使用方法explain加上别名：explain trimmed;在Grunt交互界面使用命令行pig x local e explain script explain.pig如果要以图形化方式展示出来，需要在命令行中增加-dot out filename命令，这将会打印出一个文件，内容是以DOT语言展示的用户的脚本将会是如何执行的。illustrate：会对用户的数据进行抽样

35、并且对抽样的数据执行用户的脚本，但是当它遇到一些移除数据的操作符是，它会保证有一些数据可以通过这个操作符而有一些不可以。在脚本中使用：illustrate avgdiv;通过命令行使用：pig e illustrate script illustrate.pig编写优质的脚本尽早地并经常地进行过滤，Pig的逻辑优化器会尽可能第将过滤器前置Pig什么时候前推过过滤器cogroup有时进行推测执行如果只是用于cogroup的一个输入而且不包含UDF，那么过滤器会被前推cross有时进行推测执行如果只是用于cross的一个输入那么过滤器会被前推distinct会进行推测执行filter不会进行推测执

36、行会尝试将过滤器合并并且避免数据传递通过第2个操作符。这仅当所有的过滤器前推都完成时才会进行foreach有时进行推测执行当过滤器引用到的字段是之前存在的并且foreach后面也有，并且foreach并没有改变这些字段的时候会进行过滤器前推group有时进行推测执行如果不包含UDF那么过滤器会被前推join有时进行推测执行如果过滤器只用于join操作的一个输入，而且如果对这个输入进行的不是外部连接，那么过滤器会被前推load不会进行推测执行mapreduce不会进行推测执行mapreduce对Pig是不透明的，因为它无法确定进行推测执行是否是安全的sort会进行推测执行split不会进行推测执

37、行store不会进行推测执行stream不会进行推测执行stream对Pig是不透明的，因为它无法确定进行推测执行是否是安全的union会进行推测执行尽早地并经常地进行映射当Pig逻辑优化器认定某些字段不再使用的时候会强有力地移除这些字段，然而在某些情况下Pig的逻辑优化器无法判断某些字段是否不再使用，所有用户最好是在foreach语句中尽早地移除不需要的字段。正确并合理使用join对于replicated join，小表必须是作为最后一个输入给出的。对于skewed join，第2个输入是用于对大的键抽样的。对于默认的join，最右边的输入整个记录是直接流过的，反而其他的输入对应给定键对应的

38、记录是具体化到内存中的。因此如果知道join的一个输入中每个键对应的记录要比其他的多，那么应当把它放在join的最右边的位置。对于merge join，左边的输入是作为MapReduce任务的输入，因此开始的map任务个数取决于这个输入。如果一个输入比另一个输入要大，那么应该将输入大的放置在join的左边以此为任务获得更多的map任务，这也会降低为右边的输入建立索引的抽样步骤的数据量。内置UDF内置加载函数和存储函数加载函数函数名称加载地址所表明的含义构造函数参数描述HBaseStorageHbase表名第一个参数是描述列族的字符串和与Pig字段对应的列，第2个参数选项，字符串类型（可选）从H

39、base中加载数据PigStorage HDFS文件路径第1个参数是字段间的分隔符（可选的，默认为制表符）从HDFS中加载text文件数据TextLoaderHDFS文件路径没有参数按行读取text文本文件，每一行作为一个有一个chararray类型字段的tuple存储函数函数名称加载地址所表明的含义构造函数参数描述HBaseStorageHbase表名第1个参数是描述Pig字段对应的Hbase列族和列的字符串存储数据到Hbase中PigStorageHDFS文件路径第1个参数是字段间的分隔符（可选的，默认为制表符）存储文本文件数据到HDFS中内置求值函数和过滤函数内置数学函数函数名称参数返回

40、值起始版本double ABS(double input)input绝对值0.8double ACOS(double input)input反余弦0.8double ASIN(double input)input反正弦0.8double ATAN(double input)input反正切0.8double CBRT(double input)input立方根0.8double CEIL(double input)input向上取整计算，它返回的是大于或等于函数的参数，并且与之最接近的整数0.8double COS(double input)input余弦0.8double COSH(doub

41、le input)input双曲余弦0.8double EXP(double input)input以e为底的指数函数，返回e（自然对数的底）的幂次方0.8double FLOOR(double input)input返回小于或等于指定表达式的最大整数0.8double LOG(double input)input自然对数0.8double LOG10(double input)input以10为底的对数函数0.8long ROUND(double input)input对输入值进行四舍五入取值0.8double SIN(double input)input正弦0.8double SINH(d

42、ouble input)input双曲正弦0.8double SORT(double input)input平方根0.8double TAN(double input)input正切0.8double TANH(double input)input双曲正切0.8内置聚合函数内置聚合函数函数名称参数返回值起始版本int AVG(int) input)inputinput中的所有值的平均值，null值会被忽略0.2long AVG(int) input)inputinput中的所有值的平均值，null值会被忽略0.2long AVG(float) input)inputinput中的所有值的平均值

43、，null值会被忽略0.2double AVG(double) input)inputinput中的所有值的平均值，null值会被忽略0.2double AVG(bytearray) input)inputinput中的所有bytearray类型值的平均值，会被类型转换为double型：null值会被忽略0.1long COUNT(与SQL中COUNT（字段）的作用相同）inputinput中的记录的个数，包括null值0.1long COUNT_STAR(与SQL中的COUNT(*)的作用相同)inputinput中的所有记录的个数，包括null值0.4int MAX(int) input)

44、inputinput中的最大值，null值会被忽略0.2long MAX(long) input)inputinput中的最大值，null值会被忽略0.2float MAX(float) input)inputinput中的最大值，null值会被忽略0.2double MAX(double) input)inputinput中的最大值，null值会被忽略0.2chararray MAXinputinput中的最大值，null值会被忽略0.2double MAX(bytearray) input)inputinput中所有bytearray类型中值的最大值，会被类型转换为double型；null

45、值会被忽略0.1int MIN(int) input)inputinput中的最小值，null值会被忽略0.2long MIN(long) input)inputinput中的最小值，null值会被忽略0.2float MIN(float) input)inputinput中的最小值，null值会被忽略0.2double MIN(double) input)inputinput中的最小值，null值会被忽略0.2chararray MINinputinput中的最小值，null值会被忽略0.2double MIN(bytearray) input)inputinput中所有bytearray类

46、型中值的最小值，会被类型转换为double型；null值会被忽略0.1long SUM(int) input)input数据包中所有值的和；null值会被忽略0.1long SUM(long) input)input数据包中所有值的和；null值会被忽略0.2double SUM(float) input)input数据包中所有值的和；null值会被忽略0.2double SUM(doubel) input)input数据包中所有值的和；null值会被忽略0.2double SUM(bytearray) input)input数据包中所有bytearray类型值的和，会被类型转换为double

47、型；null值会被忽略0.1内置chararray和bytearray处理UDF内置chararray和bytearray处理UDF函数名称参数返回值起始版本chararray CONCAT(chararray c1,chararray c2)c1 c2c1和c2连接0.1bytearray CONCAT(bytearray b1,bytearray b2)b1 b2b1和b2连接0.1int INDEXOF(chararray source,chararray search)source：被查找的字符串 search：要查找的字符串source字符串中第1个seach字符串所在的位置；如果字

48、符串中没有要查找的字符串则返回-10.8int LAST_INDEX_OF(chararray source,chararray search)source：被查找的字符串 search：要查找的字符串source字符串中最后一个search字符串所在的位置；如果字符串中没有要查找的字符串则返回-10.8chararray LCFIRST(chararray input)input第1个字符转换为小写0.8chararray LOWER(chararray input)input所有的输入字符转换为小写0.8chararray REGEX_EXTRACT(chararray source,ch

49、ararray regex,int n)source：被查找的字符串 regex：使用的正则表达式 n：第n个匹配的部分，从0开始source中符合正则表达式的第n个部分；如果没有匹配的部分则返回null0.8(chararray) REGEX_EXTRACT_ALL(chararray source,chararray regex)source：被查找的字符串 regex：使用的正则表达式source中符合正则表达式的所有部分放到tuple中；如果没有匹配到的部分则返回null0.8chararray REPLACE(chararray source,chararray toReplace,chararray newValue)source：被查找的字符串 toReplace：将被替换的字符串 newValue：用于替换的新的字符串source中所有的toReplace字符串被替换为newValue字符串0.8long SIZE(chararray input)inputinput中字符的个数0.2(chararray) STRSPLIT(chararray source)按照空格分隔字符串source：将进行分割的字符串source分割后的使用包含一个字段的tuple存放各个部分0.8(chararray) STRSPLIT(chararra