图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Pig › 知识型› Hadoop pig进阶语法详解
分享

Hadoop pig进阶语法详解

nettman 2014-8-22 20:03:12 发表于 知识型 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 0 10494

问题导读
1.pig是否区分大小写？
2.pig如何加载数据？
3.pig中using的都用来什么？
4.as是否可以用来定义模式？
5.foreach支持什么关键词？






Pig Latin是一种数据流语言，变量的命名规则同java中变量的命名规则，变量名可以复用（不建议这样做，这种情况下相当与新建一个变量，同时删除原来的变量）

  1. A = load 'NYSE_dividends' (exchange, symbol, date, dividends);

  2. A = filter A by dividends > 0;

  3. A = foreach A generate UPPER(symbol);
复制代码

注释：--单行注释；/*……*/多行注释；
。Pig Latin关键词不区分大小写，比如load，foreach，但是变量名和udf区分大小写，COUNT是udf，所以不同于count。
。Load 加载数据
默认加载当前用户的home目录（/users/yourlogin），可以在grunt下输入cd 命令更改当前所在目录。
  1. divs = load '/data/examples/NYSE_dividends'

复制代码

也可以输入完整的文件名
  1. divs = load ‘hdfs://nn.acme.com/data/examples/NYSE_dividends’
复制代码

默认使用TAB(\t)作为分割符，也可以使用using定义其它的分割符
  1. divs = load 'NYSE_dividends' using PigStorage(',');
复制代码

注意：只能用一个字符作为分割符
还可以使用using定义其它的加载函数
  1. divs = load 'NYSE_dividends' using HBaseStorage();
复制代码


as用于定义模式
  1. divs = load 'NYSE_dividends' as (exchange, symbol, date, dividends);
复制代码


也可以使用通配符加载一个目录下的所有文件，该目录下的所有子目录的文件也会被加载。通配符由hadoop文件系统决定，下面是hadoop 0.20所支持的通配符
glob
comment

?
Matches any single character.

*
Matches zero or more characters.

[abc]
Matches a single character from character set (a,b,c).

[a-z]
Matches a single character from the character range (a..z), inclusive. The first character must be lexicographically less than or equal to the second character.

[^abc]
Matches a single character that is not in the character set (a, b, c). The ^ character must occur immediately to the right of the opening bracket.

[^a-z]
Matches a single character that is not from the character range (a..z) inclusive. The ^ character must occur immediately to the right of the opening bracket.

\c
Removes (escapes) any special meaning of character c.

{ab,cd}
Matches a string from the string set {ab, cd}
。as 定义模式，可用于load ** [as (ColumnName[:type])]，foreach…generate ColumnName [as newColumnName]
。store存储数据，默认用using PigStorage 使用tab作为分割符。

也可以输入完整路径比如hdfs://nn.acme.com/data/examples/processed.
可以使用using调用其它存储函数或其它分割符
  1. store processed into 'processed' using

  2.       HBaseStorage();
复制代码
  1. store processed into 'processed' using PigStorage(',');
复制代码

注意：数据存储并不是存储为一个文件，而是由reduce进程数决定的多个part文件。
。foreach…generate
  • [begin .. end]
    • *匹配所有，同样适用与udf；
    ..匹配begin和end之间的部分，包括begin和end
    1. prices = load 'NYSE_daily' as (exchange, symbol, date, open,

    2. high, low, close, volume, adj_close);

    3. beginning = foreach prices generate ..open; -- produces exchange, symbol, date, open

    4. middle    = foreach prices generate open..close; -- produces open, high, low, close

    5. end       = foreach prices generate volume..; -- produces volume, adj_close
    复制代码

    一般情况下foreach…generate…重新生成的模式中的数据名和数据类型保持原来的名字和数据类型，但是如果有表达式则不会，可以在generate 变量后使用as关键词定义别名；
    1. divs = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,

    2.           date:chararray, dividends:float);

    3. sym  = foreach divs generate symbol;

    4. describe sym;

    5. 

    6. sym: {symbol: chararray}
    复制代码
    1. divs     = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,

    2.                 date:chararray, dividends:float);

    3. in_cents = foreach divs generate dividends * 100.0 as dividend, dividends * 100.0; 

    4. describe in_cents;

    5. 

    6. in_cents: {dividend: double,double}
    复制代码


    #用于map查找；.用于tuple（元组）投影；
    1. bball = load 'baseball' as (name:chararray, team:chararray,

    2.           position:bag{t:(p:chararray)}, bat:map[]);

    3. avg = foreach bball generate bat#'batting_average';
    复制代码
    1. A = load 'input' as (t:tuple(x:int, y:int));

    2. B = foreach A generate t.x, t.$1;
    复制代码


    3.获取bag（包）中的数据
    1. A = load 'input' as (b:bag{t:(x:int, y:int)});

    2. B = foreach A generate b.x;
    复制代码
    1. A = load 'input' as (b:bag{t:(x:int, y:int)});

    2. B = foreach A generate b.(x, y);
    复制代码


    下面的语句将执行不了
    1. A = load 'foo' as (x:chararray, y:int, z:int);

    2. B = group A by x; -- produces bag A containing all the records for a given value of x

    3. C = foreach B generate SUM(A.y + A.z);
    复制代码



    因为A.y 和 A.z都是bag，符号+对于bag不适用。
    正确的做法如下
    1. A = load 'foo' as (x:chararray, y:int, z:int);

    2. A1 = foreach A generate x, y + z as yz;

    3. B = group A1 by x;

    4. C = foreach B generate SUM(A1.yz);
    复制代码



    。foreach中嵌套其它语句
    1. 

    2. --distinct_symbols.pig

    3. daily    = load 'NYSE_daily' as (exchange, symbol); -- not interested in other fields

    4. grpd     = group daily by exchange;

    5. uniqcnt  = foreach grpd {

    6.            sym      = daily.symbol;

    7.            uniq_sym = distinct sym;

    8.            generate group, COUNT(uniq_sym);

    9. };
    复制代码



    注意:foreach内部只支持distinct, filter, limit, order关键词；最后一句必须是generate；
    1. 

    2. --double_distinct.pig

    3. divs = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray);

    4. grpd = group divs all;

    5. uniq = foreach grpd {

    6.             exchanges      = divs.exchange;

    7.             uniq_exchanges = distinct exchanges;

    8.             symbols        = divs.symbol;

    9.             uniq_symbols   = distinct symbols;

    10.             generate COUNT(uniq_exchanges), COUNT(uniq_symbols);

    11. };

    复制代码



    。flatten消除包嵌套关系
    1. --flatten.pig

    2. players = load 'baseball' as (name:chararray, team:chararray,

    3.             position:bag{t:(p:chararray)}, bat:map[]);

    4. pos     = foreach players generate name, flatten(position) as position;

    5. bypos   = group pos by position;
    复制代码



    1. --flatten_noempty.pig

    2. players = load 'baseball' as (name:chararray, team:chararray,

    3.             position:bag{t:(p:chararray)}, bat:map[]);

    4. noempty = foreach players generate name,

    5.             ((position is null or IsEmpty(position)) ? {('unknown')} : position)

    6.             as position;

    7. pos     = foreach noempty generate name, flatten(position) as position;

    8. bypos   = group pos by position;
    复制代码


    。filter （注：pig中的逻辑语句同样遵循短路原则）
    注意：null == 任何数据
    。filter结合matches使用正则表达式（matches前加not表示不匹配）
    pig中的正则表达式格式和java中的正则表达所一样，参考 http://docs.oracle.com/javase/6/docs/api/java/util/regex/Pattern.html
    各种转义字符，转义字符使用方式：\\后面跟上转义码
    1. 点的转义：. ==> u002E 

    2. 美元符号的转义：$ ==> u0024 

    3. 乘方符号的转义：^ ==> u005E 

    4. 左大括号的转义：{ ==> u007B 

    5. 左方括号的转义：[ ==> u005B 

    6. 左圆括号的转义：( ==> u0028 

    7. 竖线的转义：| ==> u007C 

    8. 右圆括号的转义：) ==> u0029 

    9. 星号的转义：* ==> u002A 

    10. 加号的转义：+ ==> u002B 

    11. 问号的转义：? ==> u003F 

    12. 反斜杠的转义： ==> u005C 
    复制代码



    下面的例子查找包括CM.的记录
    1. -- filter_not_matches.pig

    2. divs           = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,

    3.                     date:chararray, dividends:float);

    4. notstartswithcm = filter divs by not symbol matches '.*CM\\2u002E1.*';
    复制代码



    。group之后的数据是一个map，其中key是group所用的键值，value是group针对的变量；
    可用()同时对多个变量作group，group…all用于所有变量（注意：使用all时没有by），group之后的变量分为两个部分，第一部分变量名是group（不能更改），第二部是和原始bag模式一样的bag。

    1. 

    2. --twokey.pig

    3. daily = load 'NYSE_daily' as (exchange, stock, date, dividends);

    4. grpd  = group daily by (exchange, stock);

    5. avg   = foreach grpd generate group, AVG(daily.dividends);

    6. describe grpd;

    7. grpd: {group: (exchange: bytearray,stock: bytearray),daily: {exchange: bytearray,

    8.     stock: bytearray,date: bytearray,dividends: bytearray}}
    复制代码
    1. --countall.pig

    2. daily = load 'NYSE_daily' as (exchange, stock);

    3. grpd  = group daily all;

    4. cnt   = foreach grpd generate COUNT(daily);
    复制代码



    。cogroup对多个变量进行group
    注意：所有key值为null的数据都被归为同一类，这一点和group相同，和join不同。
    1. A = load 'input1' as (id:int, val:float);

    2. B = load 'input2' as (id:int, val2:int);

    3. C = cogroup A by id, B by id;

    4. describe C;

    5. 

    6. C: {group: int,A: {id: int,val: float},B: {id: int,val2: int}}
    复制代码



    。order by
    对单列进行排序
    1. --order.pig

    2. daily   = load 'NYSE_daily' as (exchange:chararray, symbol:chararray,

    3.             date:chararray, open:float, high:float, low:float, close:float,

    4.             volume:int, adj_close:float);

    5. bydate  = order daily by date;
    复制代码



    对多列进行排序
    1. --order2key.pig

    2. daily          = load 'NYSE_daily' as (exchange:chararray, symbol:chararray,

    3.                     date:chararray, open:float, high:float, low:float,

    4.                     close:float, volume:int, adj_close:float);

    5. bydatensymbol  = order daily by date, symbol;
    复制代码



    desc关键词按降序进行排序，null小于所有词
    1. --orderdesc.pig

    2. daily    = load 'NYSE_daily' as (exchange:chararray, symbol:chararray,

    3.             date:chararray, open:float, high:float, low:float, close:float,

    4.             volume:int, adj_close:float);

    5. byclose  = order daily by close desc, open;

    6. dump byclose; -- open still sorted in ascending order
    复制代码



    。distinct只能去掉整个元组的重复行，不能去掉某几个特定列的重复行
    1. --distinct.pig

    2. -- find a distinct list of ticker symbols for each exchange

    3. -- This load will truncate the records, picking up just the first two fields.

    4. daily   = load 'NYSE_daily' as (exchange:chararray, symbol:chararray);

    5. uniq    = distinct daily;
    复制代码



    。join/left join / right join
    null不匹配任何数据
    1. -- join2key.pig

    2. daily = load 'NYSE_daily' as (exchange, symbol, date, open, high, low, close,

    3.             volume, adj_close);

    4. divs  = load 'NYSE_dividends' as (exchange, symbol, date, dividends);

    5. jnd   = join daily by (symbol, date), divs by (symbol, date);
    复制代码
    1. --leftjoin.pig

    2. daily = load 'NYSE_daily' as (exchange, symbol, date, open, high, low, close,

    3.             volume, adj_close);

    4. divs  = load 'NYSE_dividends' as (exchange, symbol, date, dividends);

    5. jnd   = join daily by (symbol, date) left outer, divs by (symbol, date);
    复制代码


    也可以同时多个变量，但只用于inner join
    1. A = load 'input1' as (x, y);

    2. B = load 'input2' as (u, v);

    3. C = load 'input3' as (e, f);

    4. alpha = join A by x, B by u, C by e;
    复制代码



    也可以自身和自身join，但数据要加载两次
    1. --selfjoin.pig

    2. -- For each stock, find all dividends that increased between two dates

    3. divs1     = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,

    4.                 date:chararray, dividends);

    5. divs2     = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,

    6.                 date:chararray, dividends);

    7. jnd       = join divs1 by symbol, divs2 by symbol;

    8. increased = filter jnd by divs1::date < divs2::date and

    9.                 divs1::dividends < divs2::dividends;
    复制代码




    下面这样不行
    1. --selfjoin.pig

    2. -- For each stock, find all dividends that increased between two dates

    3. divs1     = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,

    4.                 date:chararray, dividends);

    5. jnd       = join divs1 by symbol, divs1 by symbol;

    6. increased = filter jnd by divs1::date < divs2::date and

    7.                 divs1::dividends < divs2::dividends;
    复制代码



    。union 相当与sql中的union，但与sql不通的是pig中的union可以针对两个不同模式的变量：如果两个变量模式相同，那么union后的变量模式与变量的模式一样；如果一个变量的模式可以由另一各变量的模式强制类型转换，那么union后的变量模式与转换后的变量模式相同；否则，union后的变量没有模式。
    1. A = load 'input1' as (x:int, y:float);

    2. B = load 'input2' as (x:int, y:float);

    3. C = union A, B;

    4. describe C;

    5. 

    6. C: {x: int,y: float}

    7. 

    8. A = load 'input1' as (x:double, y:float);

    9. B = load 'input2' as (x:int, y:double);

    10. C = union A, B;

    11. describe C;

    12. 

    13. C: {x: double,y: double}

    14. 

    15. A = load 'input1' as (x:int, y:float);

    16. B = load 'input2' as (x:int, y:chararray);

    17. C = union A, B;

    18. describe C;

    19. 

    20. Schema for C unknown.

    21. 注意：在pig 1.0中 执行不了最后一种union。
    复制代码



    注意：union不会剔除重复的行
    如果需要对两个具有不通列名的变量union的话，可以使用onschema关键字
    1. A = load 'input1' as (w: chararray, x:int, y:float);

    2. B = load 'input2' as (x:int, y:double, z:chararray);

    3. C = union onschema A, B;

    4. describe C;

    5. 

    6. C: {w: chararray,x: int,y: double,z: chararray}
    复制代码



    。cross 相当于离散数学中的叉乘，输入行数分别为m行，n行，输出行数则为m*n行。

    1. 

    2. --thetajoin.pig

    3. --I recommand running this one on a cluster too

    4. daily   = load 'NYSE_daily' as (exchange:chararray, symbol:chararray,

    5.             date:chararray, open:float, high:float, low:float,

    6.             close:float, volume:int, adj_close:float);

    7. divs    = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,

    8.             date:chararray, dividends:float);

    9. crossed = cross daily, divs;

    10. tjnd    = filter crossed by daily::date < divs::date;

    复制代码



    。limit

    1. --limit.pig

    2. divs    = load 'NYSE_dividends';

    3. first10 = limit divs 10;
    复制代码


    在pig中除了order by 之外生成的数据都没有固定的顺序。上面的程序每次生成的数据也是不一样的。
    。sample 用于生成测试数据，按指定参数选取部分数据。下面的程序选取10%的数据。
    1. --sample.pig

    2. divs = load 'NYSE_dividends';

    3. some = sample divs 0.1;
    复制代码



    。Parallel  设置pig的reduce进程个数
    1. --parallel.pig

    2. daily   = load 'NYSE_daily' as (exchange, symbol, date, open, high, low, close,

    3.             volume, adj_close);

    4. bysymbl = group daily by symbol parallel 10;
    复制代码



    parallel只针对一条语句，如果希望脚本中的所有语句都有10个reduce进程，可以使用 set default_parallel 10命令
    1. --defaultparallel.pig

    2. set default_parallel 10;

    3. daily   = load 'NYSE_daily' as (exchange, symbol, date, open, high, low, close,

    4.             volume, adj_close);

    5. bysymbl = group daily by symbol;

    6. average = foreach bysymbl generate group, AVG(daily.close) as avg;

    7. sorted  = order average by avg desc;
    复制代码



    如果同时使用parallel和set default_parallel，那么parallel中的参数将覆盖set default_parallel
    。UDF
    注册udf
    1. --register.pig

    2. register 'your_path_to_piggybank/piggybank.jar';

    3. divs      = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,

    4.                 date:chararray, dividends:float);

    5. backwards = foreach divs generate

    6.                 org.apache.pig.piggybank.evaluation.string.Reverse(symbol);
    复制代码



    定义udf别名
    1. --define.pig

    2. register 'your_path_to_piggybank/piggybank.jar';

    3. define reverse org.apache.pig.piggybank.evaluation.string.Reverse();

    4. divs      = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,

    5.                 date:chararray, dividends:float);

    6. backwards = foreach divs generate reverse(symbol);
    复制代码


    构造函数带参数的udf
    1. --define_constructor_args.pig

    2. register 'acme.jar';

    3. define convert com.acme.financial.CurrencyConverter('dollar', 'euro');

    4. divs      = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,

    5.                 date:chararray, dividends:float);

    6. backwards = foreach divs generate convert(dividends);
    复制代码



    。托管java中的静态函数（效率较低）
    1. --invoker.pig

    2. define hex InvokeForString('java.lang.Integer.toHexString', 'int');

    3. divs  = load 'NYSE_daily' as (exchange, symbol, date, open, high, low,

    4.             close, volume, adj_close);

    5. nonnull = filter divs by volume is not null;

    6. inhex = foreach nonnull generate symbol, hex((int)volume);
    复制代码


    如果函数的参数是一个数组，那么传递过去的是一个bag
    1. define stdev InvokeForDouble('com.acme.Stats.stdev', 'double[]');

    2. A = load 'input' as (id: int, dp:double);

    3. B = group A by id;

    4. C = foreach B generate group, stdev(A.dp);
    复制代码



    。multiquery 
    1. 

    2. --multiquery.pig

    3. players    = load 'baseball' as (name:chararray, team:chararray,

    4.                 position:bag{t:(p:chararray)}, bat:map[]);

    5. pwithba    = foreach players generate name, team, position,

    6.                 bat#'batting_average' as batavg;

    7. byteam     = group pwithba by team;

    8. avgbyteam  = foreach byteam generate group, AVG(pwithba.batavg);

    9. store avgbyteam into 'by_team';

    10. flattenpos = foreach pwithba generate name, team,

    11.                 flatten(position) as position, batavg;

    12. bypos      = group flattenpos by position;

    13. avgbypos   = foreach bypos generate group, AVG(flattenpos.batavg);

    14. store avgbypos into 'by_position';
    复制代码



    。split
    1. wlogs = load 'weblogs' as (pageid, url, timestamp);

    2. split wlogs into apr03 if timestamp < '20110404',

    3.           apr02 if timestamp < '20110403' and timestamp > '20110401',

    4.           apr01 if timestamp < '20110402' and timestamp > '20110331';

    5. store apr03 into '20110403';

    6. store apr02 into '20110402';

    7. store apr01 into '20110401';

    复制代码



    。设置pig环境
    Parameter
    Value Type
    Description

    debug
    string
    Sets the logging level to DEBUG. Equivalent to passing -debug DEBUG on the command line.

    default_parallel
    integer
    Sets a default parallel level for all reduce operations in the script. See the section called “Parallel” for details.

    job.name
    string
    Assigns a name to the Hadoop job. By default the name is the filename of the script being run, or a randomly generated name for interactive sessions.

    job.priority
    string Type
    If your Hadoop cluster is using the Capacity Scheduler with priorities enabled for queues, this allows you to set the priority of your Pig job. Allowed values are very_low, low, normal, high, very_high.
    。parameter 向pig脚本传递参数
    1. --daily.pig

    2. daily     = load 'NYSE_daily' as (exchange:chararray, symbol:chararray,

    3.                 date:chararray, open:float, high:float, low:float, close:float,

    4.                 volume:int, adj_close:float);

    5. yesterday = filter daily by date == '$DATE';

    6. grpd      = group yesterday all;

    7. minmax    = foreach grpd generate MAX(yesterday.high), MIN(yesterday.low);
    复制代码



    用-p 传递参数，每个变量前都要加一个-p
    1. pig -p DATE=2009-12-17 daily.pig
    复制代码


    参数也可以放在一个文件里，每行一个参数，注释部分以#开头，使用-m 或者 -param_file.调用参数文件
    pig脚本
    1. wlogs = load 'clicks/$YEAR$MONTH01' as (url, pageid, timestamp);
    复制代码


    参数文件
    1. #Param file

    2. YEAR=2009-

    3. MONTH=12-

    4. DAY=17

    5. DATE=$YEAR$MONTH$DAY
    复制代码


    执行
    1. pig -param_file daily.params daily.pig
    复制代码


    也可以在pig内定义参数%declare 或者 %default，%default定义默认的参数，在特殊情况下可以被覆盖
    注意：%declare和%default不能用于以下位置：
    • pig脚本，此脚本非Macro宏，并且脚本被另外一个脚本调用（如果不被调用可以使用）
    1. %default parallel_factor 10;

    2. wlogs = load 'clicks' as (url, pageid, timestamp);

    3. grp   = group wlogs by pageid parallel $parallel_factor;

    4. cntd  = foreach grp generate group, COUNT(wlogs);
    复制代码



    。定义Macro宏,相当于子函数

    1. 

    2. --macro.pig

    3. -- Given daily input and a particular year, analyze how

    4. -- stock prices changed on days dividends were paid out.

    5. define dividend_analysis (daily, year, daily_symbol, daily_open, daily_close)

    6. returns analyzed {

    7.     divs          = load 'NYSE_dividends' as (exchange:chararray,

    8.                         symbol:chararray, date:chararray, dividends:float);

    9.     divsthisyear  = filter divs by date matches '$year-.*';

    10.     dailythisyear = filter $daily by date matches '$year-.*';

    11.     jnd           = join divsthisyear by symbol, dailythisyear by $daily_symbol;

    12.     $analyzed     = foreach jnd generate dailythisyear::$daily_symbol,

    13.                         $daily_close - $daily_open;

    14. };

    15. 

    16. daily   = load 'NYSE_daily' as (exchange:chararray, symbol:chararray,

    17.             date:chararray, open:float, high:float, low:float, close:float,

    18.             volume:int, adj_close:float);

    19. results = dividend_analysis(daily, '2009', 'symbol', 'open', 'close');
    复制代码


    。引用pig文件,被引用的文件被执行一遍，相当于拼接在一起，被引用的文件中不能存在自定义变量
    1. --main.pig

    2. import '../examples/ch6/dividend_analysis.pig';

    3. 

    4. daily   = load 'NYSE_daily' as (exchange:chararray, symbol:chararray,

    5.             date:chararray, open:float, high:float, low:float, close:float,

    6.             volume:int, adj_close:float);

    7. results = dividend_analysis(daily, '2009', 'symbol', 'open', 'close');
    复制代码



    默认搜索文件夹为当前文件夹，可以使用set pig.import.search.path设置搜索的路径

    1. set pig.import.search.path '/usr/local/pig,/grid/pig';

    2. import 'acme/macros.pig';
    复制代码



    加微信w3aboutyun，可拉入技术爱好者群
    回复

    使用道具 举报

    自定义广告语

    没找到任何评论，期待你打破沉寂

    高级模式
    B Color Image Link Quote Code Smilies
    您需要登录后才可以回帖 登录 | 立即注册

    本版积分规则

    发表新帖
    nettman

    超级版主

    关注

    • 2911

      主题

    • 3204

      帖子

    • 478

      粉丝

    TA的主题
    关闭

    推荐上一条 /2 下一条

    关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

    Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

    简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000