【原创】Hadoop pig进阶语法

逃离我推掉我的手 2022-05-16 08:24 163阅读 0赞

原文地址为： [【原创】Hadoop pig进阶语法][Hadoop pig]

本文来自与作者阅读[ ][Link 1][Programming ][Programming][Pig][] 所做的笔记，转载请注明出处 [http://www.cnblogs.com/siwei1988/archive/2012/08/06/2624912.html][http_www.cnblogs.com_siwei1988_archive_2012_08_06_2624912.html]

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

A = load 'NYSE_dividends' (exchange, symbol, date, dividends);
    A = filter A by dividends > 0;
    A = foreach A generate UPPER(symbol);

。注释：--单行注释；/\*……\*/多行注释；

。Pig Latin关键词不区分大小写，比如load，foreach，但是变量名和udf区分大小写，COUNT是udf，所以不同于count。

。Load 加载数据

默认加载当前用户的home目录（/users/*`yourlogin`*），可以在grunt下输入cd 命令更改当前所在目录。

divs = load '/data/examples/NYSE_dividends'

也可以输入完整的文件名

divs = load ‘hdfs://nn.acme.com/data/examples/NYSE_dividends’

默认使用TAB(\\t)作为分割符，也可以使用using定义其它的分割符

divs = load 'NYSE_dividends' using PigStorage(',');

注意：只能用一个字符作为分割符

还可以使用using定义其它的加载函数

divs = load 'NYSE_dividends' using HBaseStorage();

as用于定义模式

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

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

<table> 
 <thead> 
  <tr> 
   <th align="center">glob</th> 
   <th align="center">comment</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center">?</td> 
   <td align="center">Matches any single character.</td> 
  </tr> 
  <tr> 
   <td align="center">*</td> 
   <td align="center">Matches zero or more characters.</td> 
  </tr> 
  <tr> 
   <td align="center">[abc]</td> 
   <td align="center">Matches a single character from character set (a,b,c).</td> 
  </tr> 
  <tr> 
   <td align="center">[a-z]</td> 
   <td align="center">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.</td> 
  </tr> 
  <tr> 
   <td align="center">[^abc]</td> 
   <td align="center">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.</td> 
  </tr> 
  <tr> 
   <td align="center">[^a-z]</td> 
   <td align="center">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.</td> 
  </tr> 
  <tr> 
   <td align="center">\c</td> 
   <td align="center">Removes (escapes) any special meaning of character c.</td> 
  </tr> 
  <tr> 
   <td align="center">{ab,cd}</td> 
   <td align="center">Matches a string from the string set {ab, cd}</td> 
  </tr> 
 </tbody> 
</table>

。as 定义模式，可用于load \*\* \[as (ColumnName\[:type\])\]，foreach…generate ColumnName \[as newColumnName\]

。store存储数据，默认用using PigStorage 使用tab作为分割符。

store processed into '/data/examples/processed';

也可以输入完整路径比如`hdfs://nn.acme.com/data/examples/processed`.

可以使用using调用其它存储函数或其它分割符

store processed into 'processed' using
          HBaseStorage();

store processed into 'processed' using PigStorage(',');

注意：数据存储并不是存储为一个文件，而是由reduce进程数决定的多个part文件。

。foreach…generate\[\*\]\[begin .. end\]

\*匹配所有，同样适用与udf；

..匹配begin和end之间的部分，包括begin和end

prices = load 'NYSE_daily' as (exchange, symbol, date, open,
    high, low, close, volume, adj_close);
    beginning = foreach prices generate ..open; -- produces exchange, symbol, date, open
    middle    = foreach prices generate open..close; -- produces open, high, low, close
    end       = foreach prices generate volume..; -- produces volume, adj_close

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

divs = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,
              date:chararray, dividends:float);
    sym  = foreach divs generate symbol;
    describe sym;
    
    sym: {symbol: chararray}

divs     = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,
                    date:chararray, dividends:float);
    in_cents = foreach divs generate dividends * 100.0 as dividend, dividends * 100.0; 
    describe in_cents;
    
    in_cents: {dividend: double,double}

\#用于map查找；.用于tuple（元组）投影；

bball = load 'baseball' as (name:chararray, team:chararray,
              position:bag{t:(p:chararray)}, bat:map[]);
    avg = foreach bball generate bat#'batting_average';

A = load 'input' as (t:tuple(x:int, y:int));
    B = foreach A generate t.x, t.$1;

3.获取bag（包）中的数据

A = load 'input' as (b:bag{t:(x:int, y:int)});
    B = foreach A generate b.x;

A = load 'input' as (b:bag{t:(x:int, y:int)});
    B = foreach A generate b.(x, y);

下面的语句将执行不了

A = load 'foo' as (x:chararray, y:int, z:int);
    B = group A by x; -- produces bag A containing all the records for a given value of x
    C = foreach B generate SUM(A.y + A.z);

因为A.y 和 A.z都是bag，符号+对于bag不适用。

正确的做法如下

A = load 'foo' as (x:chararray, y:int, z:int);
    A1 = foreach A generate x, y + z as yz;
    B = group A1 by x;
    C = foreach B generate SUM(A1.yz);

。foreach中嵌套其它语句

--distinct_symbols.pig
    daily    = load 'NYSE_daily' as (exchange, symbol); -- not interested in other fields
    grpd     = group daily by exchange;
    uniqcnt  = foreach grpd {
        
               sym      = daily.symbol;
               uniq_sym = distinct sym;
               generate group, COUNT(uniq_sym);
    };

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

--double_distinct.pig
    divs = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray);
    grpd = group divs all;
    uniq = foreach grpd {
        
                exchanges      = divs.exchange;
                uniq_exchanges = distinct exchanges;
                symbols        = divs.symbol;
                uniq_symbols   = distinct symbols;
                generate COUNT(uniq_exchanges), COUNT(uniq_symbols);
    };

。flatten消除包嵌套关系

--flatten.pig
    players = load 'baseball' as (name:chararray, team:chararray,
                position:bag{t:(p:chararray)}, bat:map[]);
    pos     = foreach players generate name, flatten(position) as position;
    bypos   = group pos by position;

--flatten_noempty.pig
    players = load 'baseball' as (name:chararray, team:chararray,
                position:bag{t:(p:chararray)}, bat:map[]);
    noempty = foreach players generate name,
                ((position is null or IsEmpty(position)) ? {('unknown')} : position)
    as position;
    pos     = foreach noempty generate name, flatten(position) as position;
    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][http_docs.oracle.com_javase_6_docs_api_java_util_regex_Pattern.html]

各种转义字符，转义字符使用方式：\\\\后面跟上转义码

点的转义：. ==> u002E 
    美元符号的转义：$ ==> u0024 
    乘方符号的转义：^ ==> u005E 
    左大括号的转义：{ ==> u007B 
    左方括号的转义：[ ==> u005B 
    左圆括号的转义：( ==> u0028 
    竖线的转义：| ==> u007C 
    右圆括号的转义：) ==> u0029 
    星号的转义：* ==> u002A 
    加号的转义：+ ==> u002B 
    问号的转义：? ==> u003F 
    反斜杠的转义： ==> u005C

下面的例子查找包括CM.的记录

-- filter_not_matches.pig
    divs           = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,
                        date:chararray, dividends:float);
    notstartswithcm = filter divs by not symbol matches '.*CM\\2u002E1.*';

。group之后的数据是一个map，其中key是group所用的键值，value是group针对的变量；

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

--twokey.pig
    daily = load 'NYSE_daily' as (exchange, stock, date, dividends);
    grpd  = group daily by (exchange, stock);
    avg   = foreach grpd generate group, AVG(daily.dividends);
    describe grpd;
    grpd: {group: (exchange: bytearray,stock: bytearray),daily: {exchange: bytearray,
        stock: bytearray,date: bytearray,dividends: bytearray}}

--countall.pig
    daily = load 'NYSE_daily' as (exchange, stock);
    grpd  = group daily all;
    cnt   = foreach grpd generate COUNT(daily);

。cogroup对多个变量进行group

注意：所有key值为null的数据都被归为同一类，这一点和group相同，和join不同。

A = load 'input1' as (id:int, val:float);
    B = load 'input2' as (id:int, val2:int);
    C = cogroup A by id, B by id;
    describe C;
    
    C: {group: int,A: {id: int,val: float},B: {id: int,val2: int}}

。order by

对单列进行排序

--order.pig
    daily   = load 'NYSE_daily' as (exchange:chararray, symbol:chararray,
                date:chararray, open:float, high:float, low:float, close:float,
                volume:int, adj_close:float);
    bydate  = order daily by date;

对多列进行排序

--order2key.pig
    daily          = load 'NYSE_daily' as (exchange:chararray, symbol:chararray,
                        date:chararray, open:float, high:float, low:float,
                        close:float, volume:int, adj_close:float);
    bydatensymbol  = order daily by date, symbol;

desc关键词按降序进行排序，null小于所有词

--orderdesc.pig
    daily    = load 'NYSE_daily' as (exchange:chararray, symbol:chararray,
                date:chararray, open:float, high:float, low:float, close:float,
                volume:int, adj_close:float);
    byclose  = order daily by close desc, open;
    dump byclose; -- open still sorted in ascending order

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

--distinct.pig
    -- find a distinct list of ticker symbols for each exchange
    -- This load will truncate the records, picking up just the first two fields.
    daily   = load 'NYSE_daily' as (exchange:chararray, symbol:chararray);
    uniq    = distinct daily;

。join/left join / right join

null不匹配任何数据

-- join2key.pig
    daily = load 'NYSE_daily' as (exchange, symbol, date, open, high, low, close,
                volume, adj_close);
    divs  = load 'NYSE_dividends' as (exchange, symbol, date, dividends);
    jnd   = join daily by (symbol, date), divs by (symbol, date);

--leftjoin.pig
    daily = load 'NYSE_daily' as (exchange, symbol, date, open, high, low, close,
                volume, adj_close);
    divs  = load 'NYSE_dividends' as (exchange, symbol, date, dividends);
    jnd   = join daily by (symbol, date) left outer, divs by (symbol, date);

也可以同时多个变量，但只用于inner join

A = load 'input1' as (x, y);
    B = load 'input2' as (u, v);
    C = load 'input3' as (e, f);
    alpha = join A by x, B by u, C by e;

也可以自身和自身join，但数据要加载两次

--selfjoin.pig
    -- For each stock, find all dividends that increased between two dates
    divs1     = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,
                    date:chararray, dividends);
    divs2     = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,
                    date:chararray, dividends);
    jnd       = join divs1 by symbol, divs2 by symbol;
    increased = filter jnd by divs1::date < divs2::date and
                    divs1::dividends < divs2::dividends;

下面这样不行

--selfjoin.pig
    -- For each stock, find all dividends that increased between two dates
    divs1     = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,
                    date:chararray, dividends);
    jnd       = join divs1 by symbol, divs1 by symbol;
    increased = filter jnd by divs1::date < divs2::date and
                    divs1::dividends < divs2::dividends;

。union 相当与sql中的union，但与sql不通的是pig中的union可以针对两个不同模式的变量：如果两个变量模式相同，那么union后的变量模式与变量的模式一样；如果一个变量的模式可以由另一各变量的模式强制类型转换，那么union后的变量模式与转换后的变量模式相同；否则，union后的变量没有模式。

A = load 'input1' as (x:int, y:float);
    B = load 'input2' as (x:int, y:float);
    C = union A, B;
    describe C;
    
    C: {x: int,y: float}
    
    A = load 'input1' as (x:double, y:float);
    B = load 'input2' as (x:int, y:double);
    C = union A, B;
    describe C;
    
    C: {x: double,y: double}
    
    A = load 'input1' as (x:int, y:float);
    B = load 'input2' as (x:int, y:chararray);
    C = union A, B;
    describe C;
    
    Schema for C unknown.
    注意：在pig 1.0中 执行不了最后一种union。

注意：union不会剔除重复的行

如果需要对两个具有不通列名的变量union的话，可以使用onschema关键字

A = load 'input1' as (w: chararray, x:int, y:float);
    B = load 'input2' as (x:int, y:double, z:chararray);
    C = union onschema A, B;
    describe C;
    
    C: {w: chararray,x: int,y: double,z: chararray}

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

--thetajoin.pig
    --I recommand running this one on a cluster too
    daily   = load 'NYSE_daily' as (exchange:chararray, symbol:chararray,
                date:chararray, open:float, high:float, low:float,
    close:float, volume:int, adj_close:float);
    divs    = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,
                date:chararray, dividends:float);
    crossed = cross daily, divs;
    tjnd    = filter crossed by daily::date < divs::date;

。limit

--limit.pig
    divs    = load 'NYSE_dividends';
    first10 = limit divs 10;

在pig中除了order by 之外生成的数据都没有固定的顺序。上面的程序每次生成的数据也是不一样的。

。sample 用于生成测试数据，按指定参数选取部分数据。下面的程序选取10%的数据。

--sample.pig
    divs = load 'NYSE_dividends';
    some = sample divs 0.1;

。Parallel 设置pig的reduce进程个数

--parallel.pig
    daily   = load 'NYSE_daily' as (exchange, symbol, date, open, high, low, close,
                volume, adj_close);
    bysymbl = group daily by symbol parallel 10;

parallel只针对一条语句，如果希望脚本中的所有语句都有10个reduce进程，可以使用 set default\_parallel 10命令

--defaultparallel.pig
    set default_parallel 10;
    daily   = load 'NYSE_daily' as (exchange, symbol, date, open, high, low, close,
                volume, adj_close);
    bysymbl = group daily by symbol;
    average = foreach bysymbl generate group, AVG(daily.close) as avg;
    sorted  = order average by avg desc;

如果同时使用parallel和set default\_parallel，那么parallel中的参数将覆盖set default\_parallel

。UDF

注册udf

--register.pig
    register 'your_path_to_piggybank/piggybank.jar';
    divs      = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,
                    date:chararray, dividends:float);
    backwards = foreach divs generate
                    org.apache.pig.piggybank.evaluation.string.Reverse(symbol);

定义udf别名

--define.pig
    register 'your_path_to_piggybank/piggybank.jar';
    define reverse org.apache.pig.piggybank.evaluation.string.Reverse();
    divs      = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,
                    date:chararray, dividends:float);
    backwards = foreach divs generate reverse(symbol);

构造函数带参数的udf

--define_constructor_args.pig
    register 'acme.jar';
    define convert com.acme.financial.CurrencyConverter('dollar', 'euro');
    divs      = load 'NYSE_dividends' as (exchange:chararray, symbol:chararray,
                    date:chararray, dividends:float);
    backwards = foreach divs generate convert(dividends);

。托管java中的静态函数（效率较低）

--invoker.pig
    define hex InvokeForString('java.lang.Integer.toHexString', 'int');
    divs  = load 'NYSE_daily' as (exchange, symbol, date, open, high, low,
                close, volume, adj_close);
    nonnull = filter divs by volume is not null;
    inhex = foreach nonnull generate symbol, hex((int)volume);

如果函数的参数是一个数组，那么传递过去的是一个bag

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);

。multiquery

--multiquery.pig
    players    = load 'baseball' as (name:chararray, team:chararray,
                    position:bag{t:(p:chararray)}, bat:map[]);
    pwithba    = foreach players generate name, team, position,
                    bat#'batting_average' as batavg;
    byteam     = group pwithba by team;
    avgbyteam  = foreach byteam generate group, AVG(pwithba.batavg);
    store avgbyteam into 'by_team';
    flattenpos = foreach pwithba generate name, team,
                    flatten(position) as position, batavg;
    bypos      = group flattenpos by position;
    avgbypos   = foreach bypos generate group, AVG(flattenpos.batavg);
    store avgbypos into 'by_position';

。split

wlogs = load 'weblogs' as (pageid, url, timestamp);
    split wlogs into apr03 if timestamp < '20110404',
              apr02 if timestamp < '20110403' and timestamp > '20110401',
              apr01 if timestamp < '20110402' and timestamp > '20110331';
    store apr03 into '20110403';
    store apr02 into '20110402';
    store apr01 into '20110401';

。设置pig环境

<table>      
 <thead> 
  <tr> 
   <th align="center">Parameter</th> 
   <th align="center">Value Type</th> 
   <th align="center">Description</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center">debug</td> 
   <td align="center">string</td> 
   <td align="center">Sets the logging level to&nbsp;<code>DEBUG</code>. Equivalent to passing&nbsp;<code>-debug DEBUG</code>&nbsp;on the command line.</td> 
  </tr> 
  <tr> 
   <td align="center">default_parallel</td> 
   <td align="center">integer</td> 
   <td align="center">Sets a default parallel level for all reduce operations in the script. See&nbsp;<a title="Parallel" href="http://ofps.oreilly.com/titles/9781449302641/intro_pig_latin.html#parallel" rel="nofollow">the section called “Parallel”</a>&nbsp;for details.</td> 
  </tr> 
  <tr> 
   <td align="center">job.name</td> 
   <td align="center">string</td> 
   <td align="center">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.</td> 
  </tr> 
  <tr> 
   <td align="center">job.priority</td> 
   <td align="center">string Type</td> 
   <td align="center">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&nbsp;<code>very_low</code>,&nbsp;<code>low</code>,&nbsp;<code>normal</code>,&nbsp;<code>high</code>,&nbsp;<code>very_high</code>.</td> 
  </tr> 
 </tbody> 
</table>

。parameter 向pig脚本传递参数

--daily.pig
    daily     = load 'NYSE_daily' as (exchange:chararray, symbol:chararray,
                    date:chararray, open:float, high:float, low:float, close:float,
                    volume:int, adj_close:float);
    yesterday = filter daily by date == '$DATE';
    grpd      = group yesterday all;
    minmax    = foreach grpd generate MAX(yesterday.high), MIN(yesterday.low);

用-p 传递参数，每个变量前都要加一个-p

pig -p DATE=2009-12-17 daily.pig

参数也可以放在一个文件里，每行一个参数，注释部分以\#开头，使用`-m` 或者 `-param_file`.调用参数文件

pig脚本

wlogs = load 'clicks/$YEAR$MONTH01' as (url, pageid, timestamp);

参数文件

#Param file
    YEAR=2009-
    MONTH=12-
    DAY=17
    DATE=$YEAR$MONTH$DAY

执行

pig -param_file daily.params daily.pig

也可以在pig内定义参数%declare 或者 %default，%default定义默认的参数，在特殊情况下可以被覆盖

注意：%declare和%default不能用于以下位置：

*  pig脚本，此脚本非Macro宏，并且脚本被另外一个脚本调用（如果不被调用可以使用）

%default parallel_factor 10;
    wlogs = load 'clicks' as (url, pageid, timestamp);
    grp   = group wlogs by pageid parallel $parallel_factor;
    cntd  = foreach grp generate group, COUNT(wlogs);

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

--macro.pig
    -- Given daily input and a particular year, analyze how
    -- stock prices changed on days dividends were paid out.
    define dividend_analysis (daily, year, daily_symbol, daily_open, daily_close)
    returns analyzed {
        
        divs          = load 'NYSE_dividends' as (exchange: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,
                            $daily_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文件,被引用的文件被执行一遍，相当于拼接在一起，被引用的文件中不能存在自定义变量

--main.pig
    import '../examples/ch6/dividend_analysis.pig';
    
    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');

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

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

转载请注明本文地址： [【原创】Hadoop pig进阶语法][Hadoop pig]

[Hadoop pig]: http://www.itdaan.com/blog/2012/08/06/7345d4693ba5c03fb242f58a2da1dd5a.html
[Link 1]: http://shop.oreilly.com/product/0636920018087
[Programming]: http://shop.oreilly.com/product/0636920018087.do
[Pig]: http://www.javaxxz.com/thread-343628-1-1.html
[http_www.cnblogs.com_siwei1988_archive_2012_08_06_2624912.html]: http://www.cnblogs.com/siwei1988/archive/2012/08/06/2624912.html
[hadoop]: http://www.javaxxz.com/thread-368393-1-1.html
[http_docs.oracle.com_javase_6_docs_api_java_util_regex_Pattern.html]: http://docs.oracle.com/javase/6/docs/api/java/util/regex/Pattern.html