1.?? 整體描述

Pig包括兩部分：①用于描述數(shù)據(jù)流的語言，稱為Pig Latin；②用于運行Pig Latin程序的執(zhí)行環(huán)境，當前有兩個環(huán)境：單JVM中的本地執(zhí)行環(huán)境和hadoop集群上的分布式執(zhí)行環(huán)境。

一個Pig Latin程序有一系列的操作和變換組成。每個操作或變換對輸入進行數(shù)據(jù)處理，然后產(chǎn)生輸出結(jié)果。這些操作整體上描述了一個數(shù)據(jù)流。Pig執(zhí)行環(huán)境把數(shù)據(jù)流翻譯為可執(zhí)行的內(nèi)部表示，并運行它。在Pig內(nèi)部，這些變換操作轉(zhuǎn)換成一些列mapReduce作業(yè)，作為程序員，多數(shù)情況下并不需要知道這些轉(zhuǎn)換是如何進行的。

1.1.?Pig產(chǎn)生的原因

Pig是一種探索大數(shù)據(jù)規(guī)模數(shù)據(jù)集的腳本語言。MapReduce的一個缺點是開發(fā)周期太長。編寫mapper和reducer，對代碼進行編譯和打包、提交作業(yè)，獲取結(jié)果，這整個過程非常耗時，即便使用streaming能在這個過程中去除掉代碼的編譯和打包步驟，但這個過程仍然很耗時。

Pig的誘人之處在于它能夠用控制臺上的五六行Pig Latin代碼輕松處理TB級的數(shù)據(jù)。注意：雅虎90%的MapReducejob是Pig生成的。Twitter80%以上是Pig生成的。

1.2.? Pig的優(yōu)勢

Pig提供了多個命令用于檢查和處理程序中的數(shù)據(jù)結(jié)構(gòu)，因此，他能夠很好的支持程序員寫查詢。

Pig支持在輸入數(shù)據(jù)的一個有代表性的子集上試運行，這樣一來，用戶可以在處理整個數(shù)據(jù)集前檢查程序中是否有錯誤。

Pig被設計為可擴展的。處理路徑中的每個部分，載入、存儲、過濾、分組、連接，都是可以定制的。這些操作都可以使用用戶定義函數(shù)進行修改。且UDF比用MapReduce程序開發(fā)的庫更易于重用。

1.3.?Pig的劣勢

Pig并不適合所有的數(shù)據(jù)處理任務，和MapReduce一樣，他是為數(shù)據(jù)批處理而設計的。如果想執(zhí)行的查詢只涉及到一個大型數(shù)據(jù)集中的一小部分數(shù)據(jù)，Pig的實現(xiàn)不會很好，因為它要掃描整個數(shù)據(jù)集或者其中的很大一部分。

在有些情況下，Pig的表現(xiàn)不如MapReduce程序。例如執(zhí)行效率和某些靈活度上。

1.4.? Pig與數(shù)據(jù)庫的比較

Pig Latin和sql很相似。Group by和describe這樣的操作更加強了這種感覺。但是，兩種語言之間，以及Pig和RDBMS之間，有幾個方面是不同的。

他們之間最顯著的不同是：Pig Latin是一種數(shù)據(jù)流編程語言，而sql是一種描述型編程語言。換句話說，一個Pig Latin程序是相對于輸入的一步步操作。其中每一步都是對數(shù)據(jù)的一個簡單的變換。相反，SQL語句是一個約束的集合，這些約束結(jié)合在一起，定義了輸出。在很多方面看，用Pig Latin編程更像在RDBMS中“查詢規(guī)劃器”這一層對數(shù)據(jù)進行操作。查詢規(guī)劃器決定了如何將描述型語句轉(zhuǎn)化為一些列執(zhí)行的步驟。

RDBMS把數(shù)據(jù)存儲在嚴格定義了模式的表內(nèi)。Pig對它所處理的數(shù)據(jù)要求則寬松的多：可以在運行時定義格式，而且這是可選的。本質(zhì)上，Pig可以在任何來源的元組上進行操作。它使用UDF從原始格式中讀取元組。最常用的輸入格式是用制表符隔離的字段組成的文本文件。Pig為這種輸入提供了內(nèi)置加載函數(shù)。和傳統(tǒng)的數(shù)據(jù)庫不同，Pig并不提供專門的數(shù)據(jù)導入過程將數(shù)據(jù)加載到RDBMS。在第一步處理中，數(shù)據(jù)是從文件系統(tǒng)中加載的。

Pig對復雜、嵌套數(shù)據(jù)結(jié)構(gòu)的支持也使其不同于能處理平面數(shù)據(jù)類型的SQL。Pig的語言能夠和UDF以及流操作緊密集成。他的這一能力及嵌套數(shù)據(jù)結(jié)構(gòu)，使Pig Latin比大多數(shù)SQL的變種具有更強的定制能力。

幾個支持在線和低延時查詢的特性是RDBMS有但Pig沒有的，例如事務和索引。Pig并不支持隨機讀寫和幾十毫秒的查詢。他也不支持對一小部分數(shù)據(jù)的隨機寫。和MapReduce一樣，所有的寫都是批量的、流式的寫操作。

Hive介于Pig和傳統(tǒng)的RDBMS之間。和Pig一樣，hive也被設計為用HDFS作為存儲。但是他們之間有著顯著的區(qū)別。Hive的查詢語言HiveQL，是基于SQL的。任何熟悉SQL的人都可以輕松使用HiveQL寫查詢。和RDBMS相同，Hive要求所有數(shù)據(jù)必須存儲在表中，表必須有模式，而模式由Hive進行管理。但是，Hive允許為預先存在于HDFS的數(shù)據(jù)關(guān)聯(lián)一個模式。所以，數(shù)據(jù)的加載步驟是可選的。和Pig一樣，Hive也不支持低延時查詢。

2.??Pig Latin

2.1.? Pig Latin結(jié)構(gòu)

一個Pig Latin程序由一組語句構(gòu)成，一個語句可以理解為一個操作，或一個命令。并且為保證語句的正確性，一般以分號作為結(jié)束符。例如：

grouped_records = GROUP records BY year；

Pig latin有一個關(guān)鍵詞列表。其中的單詞在Pig Latin中有特殊含義，不能用于標識符。這些單詞包括操作（load，illustarte）、命令（cat，ls）、表達式（matches，FLATTEN）以及函數(shù)（DIFF，MAX）等。詳見后面的介紹。

Pig Latin的大小寫敏感性采用混合的規(guī)則。操作和命令是大小寫無關(guān)的，而別名和函數(shù)名是大小寫敏感的。

2.2.? Pig Latin注釋

Pig Latin有兩種注釋方法。

①雙減號表示單行注釋。Pig Latin解析器會忽略從第一個減號開始到行尾的所有內(nèi)容，例如：

??????? --my program

???????? DUMP A;

??????②C語言風格的注釋更靈活。這是因為它使用/*和*/符號表示注釋開始和結(jié)束。這樣，注釋即可以跨多行，也可以內(nèi)嵌在某一行內(nèi)，例如：

/**description of my program spaning*multiple lines.*/A = LOAD ‘input/pig/join/A’;Dump A;

2.3.? Pig Latin語句

在Pig Latin程序執(zhí)行時，每個命令按次序進行解析。如果遇到語法錯誤或者其它錯誤，例如無定義的別名，解析器會終止運行，并顯示錯誤消息。解析器會給每個關(guān)系操作建立一個邏輯計劃。邏輯計劃構(gòu)成了Pig Latin程序的核心。解析器把為一個語句創(chuàng)建的邏輯計劃加到到目前為止已經(jīng)解析完的程序的邏輯計劃上，然后繼續(xù)處理下一條語句。

特別需要注意的是，在整個程序邏輯計劃沒有構(gòu)造完成前，Pig并不處理數(shù)據(jù)。

多查詢執(zhí)行：在交互模式下，STORE和dump一樣，總是會觸發(fā)語句的執(zhí)行。但是在批處理模式下，他不會觸發(fā)執(zhí)行，這是為了性能考慮而進行的設計。在批處理模式下，Pig會解析整個腳本。例如：??

???????? 針對一個關(guān)系可以用EXPLAIN命令查看Pig創(chuàng)建的邏輯和物理計劃。EXPLAIN也會顯示MapReduce計劃，即顯示物理操作是如何組成MapReduce作業(yè)的。這個辦法，可以查看Pig會為你的查詢組成多少個MapReduce作業(yè)。

2.4.? Pig Latin關(guān)系操作

2.5.? Pig Latin診斷操作

2.6.?Pig Latin自定義函數(shù)輔助操作

為了在Pig腳本中使用用戶自定義函數(shù)，Pig Latin提供了register和define語句。

2.7.? Pig Latin提供的hadoop交互指令

文件系統(tǒng)相關(guān)的命令可以對任何hadoop文件系統(tǒng)的文件或目錄進行操作。這些命令非常像hadoop fs命令。并且可以使用Pig的fs命令訪問所有的hadoop文件系統(tǒng)。

2.8.? Pig Latin表達式

2.9.?Pig Latin類型

①??Pig原子類型：數(shù)值、文本和二進制類型。

Pig有四種數(shù)值類型：int、long、float、double。他們和java中對應的數(shù)值類型相同。

Pig有bytearray類型，它類似于表示二進制大對象的java的byte數(shù)組。

Pig有Chararray類型，它類似與用utf-16格式表示文本類型的java.lang.String。

Pig有chararray類型，可以用來加載或存儲UTF-8格式的數(shù)據(jù)。

Pig沒有一種類型對應于java的Boolean、byte、short、char。Java的這些數(shù)據(jù)類型都能方便的使用Pig的int類型或者chararray類型表示。

②?Pig復雜類型：元組（tuple）、包（bag）、映射（map）。

復雜類型通常從文件加載數(shù)據(jù)或者使用關(guān)系操作進行構(gòu)建。

映射類型只能從文件，因為Pig的關(guān)系操作不能創(chuàng)建映射。如果需要，還可以用udf生成映射。

雖然關(guān)系和包在概念上是相同的，但事實上Pig對他們的處理稍有不同。關(guān)系是頂層構(gòu)造結(jié)構(gòu)，而包必須在某個關(guān)系中。

3.?? 模式

? ? ? Pig中的一個關(guān)系可以有一個關(guān)聯(lián)的模式。模式為關(guān)系的字段指定名稱和類型。例如load語句和as子句就是用來關(guān)聯(lián)模式與關(guān)系的，如下： ??

在上面的例子中，年份已經(jīng)聲明為整數(shù)類型的，而不是chararray類型。如果對年份這一字段進行算術(shù)操作，那么用整數(shù)類型更為合適；相反，如果執(zhí)行把它作為一個標識符，那么chararray類型就更為合適。

? ? ? 我們也可以完全忽略類型聲明，例如： ??

在這個例子中，我們在模式中只定義了字段名稱：year、temperature、quality。默認的數(shù)據(jù)類型為bytearray。他表示一個二進制串。

? ? ? 不必為每一個字段都給出類型。可以讓某些字段的類型為默認的字節(jié)數(shù)組，就像如下模式聲明所示例子中的year字段： ??

但是，如果要用這種方式確定模式，必須在模式中定義每一個字段。同樣，不能只確定字段的類型而不給出其名稱。另一方面，模式本身是可選的。可以省略AS子句。如下所示：? ?

只能使用位置符號引用沒有對應模式的關(guān)系中的字段：$0表示關(guān)系中第一個字段，$1表示第二個，以此類推。他們的類型都是默認的bytearray。??

雖然不為字段分配類型很省事，但如果指定字段的類型，我們能使Pig Latin程序更清晰，也使程序更高效。因此，一般情況下，建議指明字段的數(shù)據(jù)類型。

3.1.? 驗證與空值

在Pig中，如果一個值無法被強制轉(zhuǎn)換為模式中聲明的類型，Pig會用空值null代替。在輸出到屏幕（或使用store存儲）時，空值null被顯示（或存儲）為一個空位。

Pig會為非法字段產(chǎn)生一個警告，但是他不會終止處理。大數(shù)據(jù)集普遍都有被損壞的值、無效值或意料之外的值，因而逐步修正每一條無法解析的記錄一般都不太現(xiàn)實。作為一種替代方案，我們可以一次性的把所有的記錄都找出來，然后再去處理他們。我們可以修正我們的程序，把這些記錄過濾掉。

3.2.? 模式合并

在Pig中，不會為數(shù)據(jù)流中每一個產(chǎn)生的關(guān)系聲明模式。在大多數(shù)情況下，Pig能夠根據(jù)關(guān)系操作之輸入關(guān)系的模式來確定輸出結(jié)果的模式。

那么模式是如何傳播到新關(guān)系的呢？有些關(guān)系操作并不改變模式。因此，limit操作產(chǎn)生的模式就和他所處理的關(guān)系的模式相同。對于其他操作，情況可能更復雜一些。例如，UNION操作將兩個或多個關(guān)系合并成一個，并試圖同時合并傳入關(guān)系的模式。如果這些模式由于數(shù)據(jù)類型或者字段個數(shù)不相同而不兼容，那么UNION產(chǎn)生的模式是不確定的。

針對數(shù)據(jù)流中的任何關(guān)系，可以使用describe操作來獲取他們的模式。如果要重新定義一個關(guān)系的模式，可以使用帶AS子句的foreach…generate操作來定義輸入關(guān)系的一部分或全部字段的模式。

4.?? 函數(shù)

Pig中的函數(shù)有四種類型：

• 計算函數(shù)（Eval function）：計算函數(shù)獲取一個或多個表達式作為輸入，并返回一個表達式。例如：MAX.
• 過濾函數(shù)（Filter function）：過濾函數(shù)式一類特殊的計算函數(shù)。這類函數(shù)返回的是邏輯布爾值。例如：IsEmpty.
• 加載函數(shù)（Load function）：加載函數(shù)指明如何從外部存儲加載數(shù)據(jù)到一個關(guān)系。
• 存儲函數(shù)（Store function）：存儲函數(shù)指明如何把一個關(guān)系中的內(nèi)容存到外部存儲。

4.1.? Pig的內(nèi)置函數(shù)

如果表中沒有你需要的函數(shù)，可以自己寫。但是在此之前，你可以看一下piggybank。這是一個pig社區(qū)共享的pig函數(shù)庫。網(wǎng)址為：http://wiki.apache.org/pig/PiggyBank.

4.2.? 用戶自定義函數(shù)

? ? 在自定義函數(shù)這一章節(jié)中，涉及到的知識比較多，而且所有的書籍中也僅僅是寫了幾個簡單的例子，還根據(jù)例子說了一些注意事項，不是很全面。如果涉及到自定義函數(shù)的編寫。建議查看如下兩個網(wǎng)址：http://archive.cloudera.com/cdh4/cdh/4/pig-0.10.0-cdh4.1.0/udf.html和http://pig.apache.org/docs/r0.11.1/udf.html。在這里有詳細的講解和例子。

4.2.1. ?UDF實例

package com.cyou.pig.udfs;import java.io.IOException; import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Arrays; import java.util.Calendar; import java.util.Date; import java.util.List; import java.util.regex.Matcher; import java.util.regex.Pattern;import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.InputFormat; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.RecordReader; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.pig.FileInputLoadFunc; import org.apache.pig.PigException; import org.apache.pig.backend.executionengine.ExecException; import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit; import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigTextInputFormat; import org.apache.pig.bzip2r.Bzip2TextInputFormat; import org.apache.pig.data.DataByteArray; import org.apache.pig.data.Tuple; import org.apache.pig.data.TupleFactory;public class DateRangeLoad extends FileInputLoadFunc {protected RecordReader reader = null;private String loadLocation;private String currentDate;private int offset = 0;private int range = 0;private byte fieldDel = '\t';private ArrayList<Object> mProtoTuple = null;private final TupleFactory mTupleFactory = TupleFactory.getInstance();private final Pattern p= Pattern.compile("%date");private final SimpleDateFormat df = new SimpleDateFormat("yyyyMMdd");public DateRangeLoad(){this.currentDate = df.format(new Date());}public DateRangeLoad(String currentDate){this.currentDate = currentDate;}public DateRangeLoad(String currentDate,String range){this(currentDate);this.range = Integer.parseInt(range);} public DateRangeLoad(String currentDate,String offset,String range){this(currentDate,range);this.offset = Integer.parseInt(offset);}public DateRangeLoad(String currentDate,String offset,String range,String delimiter){this(currentDate,offset,range);if (delimiter.length() == 1) {this.fieldDel = (byte)delimiter.charAt(0);} else if (delimiter.length() > 1 && delimiter.charAt(0) == '\\') {switch (delimiter.charAt(1)) {case 't':this.fieldDel = (byte)'\t';break;case 'x':fieldDel = Integer.valueOf(delimiter.substring(2), 16).byteValue();break;case 'u':this.fieldDel = Integer.valueOf(delimiter.substring(2)).byteValue();break;default:throw new RuntimeException("Unknown delimiter " + delimiter);}} else {throw new RuntimeException("PigStorage delimeter must be a single character");}}@Overridepublic void setLocation(String location, Job job) throws IOException {Matcher m = this.p.matcher(location);if(m.find()){StringBuffer buf = new StringBuffer();Calendar cal=Calendar.getInstance();cal.set(Integer.parseInt(this.currentDate.substring(0, 4)), Integer.parseInt(currentDate.substring(4, 6))-1,Integer.parseInt(currentDate.substring(6, 8)));cal.add(Calendar.DATE, this.offset);buf.append("{").append(this.df.format(cal.getTime()));for(int i=1;i<this.range;i++){cal.add(Calendar.DATE, -1);buf.append(",").append(this.df.format(cal.getTime()));}buf.append("}");location = m.replaceAll(buf.toString());}System.out.println("--------------------------------------------------------------------------");System.out.println(location);System.out.println("--------------------------------------------------------------------------");this.loadLocation = location;FileInputFormat.setInputPaths(job,location);}@Overridepublic InputFormat getInputFormat() throws IOException {if(loadLocation.endsWith(".bz2") || loadLocation.endsWith(".bz")) {return new Bzip2TextInputFormat();} else {return new PigTextInputFormat();}}@Overridepublic void prepareToRead(RecordReader reader, PigSplit split)throws IOException {this.reader = reader;}@Overridepublic Tuple getNext() throws IOException {try {boolean notDone = reader.nextKeyValue();if (!notDone) {return null;}Text value = (Text) reader.getCurrentValue();byte[] buf = value.getBytes();int len = value.getLength();int start = 0;for (int i = 0; i < len; i++) {if (buf[i] == fieldDel) {readField(buf, start, i);start = i + 1;}}// pick up the last fieldreadField(buf, start, len);Tuple t = mTupleFactory.newTupleNoCopy(mProtoTuple);mProtoTuple = null;return t;} catch (InterruptedException e) {int errCode = 6018;String errMsg = "Error while reading input";throw new ExecException(errMsg, errCode, PigException.REMOTE_ENVIRONMENT, e);}}private void readField(byte[] buf, int start, int end) {if (mProtoTuple == null) {mProtoTuple = new ArrayList<Object>();}if (start == end) {// NULL valuemProtoTuple.add(null);} else {mProtoTuple.add(new DataByteArray(buf, start, end));}} }

5.?Pig項目實例

stop_event = load '/user/hive/warehouse/yidong.db/stopevent/dateline=$date/*'; stop_orig = foreach stop_event generate $1 as appid,$2 as appver,($3=='iOS'? 1: ($3=='android'? 2 : 3)) as platform,$4 as channel,$7 as guid,$31 as duration,SUBSTRING($0,0,10) as time; stop_group = group stop_orig by (appid,appver,platform,channel,guid,time); stop_duration = foreach stop_group generate group.appid,group.appver,group.platform,group.channel,group.guid,0 as start_num:int,SUM(stop_orig.duration) as duration:int,group.time as ctime,group.time as ltime;current_event = load '/user/hive/warehouse/yidong.db/startevent/dateline=$date/*'; current_orig = foreach current_event generate $1 as appid,$2 as appver,($3=='iOS'? 1: ($3=='android'? 2 : 3)) as platform,$4 as channel,$7 as guid,SUBSTRING($0,0,10) as time,1 as num; num_group = group current_orig by (appid,appver,platform,channel,guid,time); num_count = foreach num_group generate group.appid,group.appver,group.platform,group.channel,group.guid,COUNT(current_orig.num) as start_num:int,0 as duration:int,group.time as ctime,group.time as ltime;time_event = distinct current_orig; time_orig = foreach time_event generate appid,appver,platform,channel,guid,0 as start_num:int,0 as duration:int,time as ctime,time as ltime;orig_orig = load '/user/hive/warehouse/yidong.db/stat_user' as (appid,appver,channel,platform:int,guid,start_num:int,duration:int,ctime,ltime);current_union = union ONSCHEMA time_orig,num_count,stop_duration,orig_orig; current_group = group current_union by (appid,appver,channel,platform,guid); current_table = foreach current_group generate FLATTEN(group),SUM(current_union.start_num) as start_num:int,SUM(current_union.duration) as duration:int,MIN(current_union.ctime) as ctime,MAX(current_union.ltime) as ltime;store current_table into '/user/hive/warehouse/yidong.db/stat_user_$date';rmf /user/hive/warehouse/yidong.db/stat_user;mv /user/hive/warehouse/yidong.db/stat_user_$date /user/hive/warehouse/yidong.db/stat_user;

?

總結(jié)

以上是生活随笔為你收集整理的Pig详解的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。