使用 Apache Pig 處理數據

Apache Pig 是一個高級過程語言，適合于使用 Hadoop 和 MapReduce 平臺來查詢大型半結構化數據集。通過允許對分布式數據集進行類似 SQL 的查詢，Pig 可以簡化 Hadoop 的使用。本文將探索 Pig 背后的語言，并在一個簡單的 Hadoop 集群中發現其用途。

Hadoop 的普及和其生態系統的不斷壯大并不令人感到意外。Hadoop 不斷進步的一個特殊領域是 Hadoop 應用程序的編寫。雖然編寫 Map 和 Reduce 應用程序并不十分復雜，但這些編程確實需要一些軟件開發經驗。Apache Pig 改變了這種狀況，它在 MapReduce 的基礎上創建了更簡單的過程語言抽象，為 Hadoop 應用程序提供了一種更加接近結構化查詢語言 (SQL) 的接口。因此，您不需要編寫一個單獨的 MapReduce 應用程序，您可以用 Pig Latin 語言寫一個腳本，在集群中自動并行處理與分發該腳本。

Pig Latin 示例

讓我們從一個簡單的 Pig 示例開始介紹，并剖析該示例。Hadoop 的一個有趣的用法是，在大型數據集中搜索滿足某個給定搜索條件的記錄（在 Linux? 中被稱為 grep）。清單 1 顯示了在 Pig 中實現該過程的簡單性。在所顯示的三行代碼中，只有一行是真正的搜索。第一行只是將測試數據集（消息日志）讀取到代表元組集合的包中。用一個正則表達式來篩選該數據（元組中的惟一條目，表示為 $0 或 field 1），然后查找字符序列 WARN。最后，在主機文件系統中將這個包存儲在一個名為 warnings 的新文件中，這個包現在代表來自消息的包含 WARN 的所有元組。

如您所見，這個簡單的腳本實現了一個簡單的流，但是，如果直接在傳統的 MapReduce 模型中實現它，則需要增加大量的代碼。這使得學習 Hadoop 并開始使用數據比原始開發容易得多。

現在讓我們更深入地探討 Pig 語言，然后查看該語言的一些功能的其他示例。

Pig Latin 的基礎知識

Pig Latin 是一個相對簡單的語言，它可以執行語句。一調語句 就是一個操作，它需要輸入一些內容（比如代表一個元組集的包），并發出另一個包作為其輸出。一個包 就是一個關系，與表類似，您可以在關系數據庫中找到它（其中，元組代表行，并且每個元組都由字段組成）。

用 Pig Latin 編寫的腳本往往遵循以下特定格式，從文件系統讀取數據，對數據執行一系列操作（以一種或多種方式轉換它），然后，將由此產生的關系寫回文件系統。您可以在 清單 1 中看到該模式的最簡單形式（一個轉換）。

Pig 擁有大量的數據類型，不僅支持包、元組和映射等高級概念，還支持簡單的數據類型，如 int、long、float、double、chararray 和 bytearray。如果使用簡單的類型，您會發現，除了稱為 bincond 的條件運算符（其操作類似于 C ternary 運算符）之外，還有其他許多算術運算符（比如 add、subtract、multiply、divide 和 module）。并且，如您所期望的那樣，還有一套完整的比較運算符，包括使用正則表達式的豐富匹配模式。

所有 Pig Latin 語句都需要對關系進行操作（并被稱為關系運算符）。正如您在 清單 1 中看到的，有一個運算符用于從文件系統加載數據和將數據存儲到文件系統中。有一種方式可以通過迭代關系的行來 FILTER 數據。此功能常用于從后續操作不再需要的關系中刪除數據。另外，如果您需要對關系的列進行迭代，而不是對行進行迭代，您可以使用 FOREACH 運算符。FOREACH 允許進行嵌套操作，如 FILTER 和 ORDER，以便在迭代過程中轉換數據。

ORDER 運算符提供了基于一個或多個字段對關系進行排序的功能。JOIN 運算符基于公共字段執行兩個或兩個以上的關系的內部或外部聯接。SPLIT 運算符提供了根據用戶定義的表達式將一個關系拆分成兩個或兩個以上關系的功能。最后，GROUP 運算符根據某個表達式將數據分組成為一個或多個關系。表 1 提供了 Pig 中的部分關系運算符列表。

               
messages = LOAD 'messages';
warns = FILTER messages BY $0 MATCHES '.*WARN+.*';
STORE warns INTO 'warnings';

雖然這不是一個詳盡的 Pig Latin 運算符清單，但該表提供了一套在處理大型數據集時非常有用的操作。您可以通過 參考資料 了解完整的 Pig Latin 語言，因為 Pig 有一套不錯的在線文檔。現在嘗試著手編寫一些 Pig Latin 腳本，以了解這些運算符的實際工作情況。

獲得 Pig

在有關 Hadoop 的早期文章中，我采用的方法是將 Hadoop 安裝和配置為一個軟件包。但 Cloudera 通過用 Linux 將它打包為一個虛擬設備，使得 Hadoop 更易于使用。雖然它是一個較大的下載，但它已預建立并配置了虛擬機 (VM)，其中不僅有 Hadoop，還包括了 Apache Hive 和 Pig。因此，利用一個下載和免費提供的 2 型虛擬機管理程序（VirtualBox 或基于內核的虛擬機 [KVM]），您便可以擁有預配置的、已準備就緒的整個 Hadoop 環境。

讓 Hadoop 和 Pig 啟動并運行

下載完您的特定虛擬機文件之后，需要為您的特定虛擬機管理程序創建一個 VM。在 參考資料 中，您可以找到該操作的分步指南。

一旦創建了自己的 VM，就可以通過 VirtualBox 來啟動它，VirtualBox 引導 Linux 內核，并啟動所有必要的 Hadoop 守護進程。完成引導后，從創建一個與 Hadoop 和 Pig 通信的終端開始相關操作。

您可以在兩種模式中任選一種來使用 Pig。第一種是 Local（本地）模式，它不需要依賴 Hadoop 或 Hadoop 分布式文件系統 (HDFS)，在該模式中，所有操作都在本地文件系統上下文中的單一 Java 虛擬機 (JVM) 上執行。另一種模式是 MapReduce 模式，它使用了 Hadoop 文件系統和集群。

Local 模式的 Pig

對于 Local 模式，只需啟動 Pig 并用 exectype 選項指定 Local 模式即可。這樣做可以將您帶入 Grunt 外殼，使您能夠以交互方式輸入 Pig 語句：

在這里，您能夠以交互方式編寫 Pig Latin 腳本的代碼，并查看每個運算符后面的結果。返回 清單 1，并嘗試使用這個腳本（參見 清單 2）。注意，在這種情況下，不需要將數據存儲到某個文件中，只需將它轉儲為一組關系。您可能會在修改后的輸出中看到，每個日志行（與 FILTER 定義的搜索條件相匹配）本身就是一個關系（以括號 [()] 為界）。

$ pig -x local ...
grunt>

如果您已經指定 STORE 運算符，那么它會在一個指定名稱的目錄（而不是一個簡單的常規文件）中生成您的數據。

Mapreduce 模式中的 Pig

對于 MapReduce 模式，必須首先確保 Hadoop 正在運行。要做到這一點，最簡單的方法是在 Hadoop 文件系統樹的根上執行文件列表操作，如 清單 3 所示。

               
grunt> messages = LOAD '/var/log/messages'; grunt> warns = FILTER messages BY $0 MATCHES '.*WARN+.*'; grunt> DUMP warns ...
(Dec 10 03:56:43 localhost NetworkManager: <WARN> nm_generic_enable_loopback(): error ...
(Dec 10 06:10:18 localhost NetworkManager: <WARN> check_one_route(): (eth0) error ...
grunt>

如清單 3 所示，如果 Hadoop 成功運行，此代碼的結果會是一個或多個文件組成的列表。現在，讓我們來測試 Pig。從啟動 Pig 開始，然后將目錄更改為您的 HDFS 根，以確定在 HDFS 中是否可以看到外部所看到的結果（參見 清單 4）。

               
$ hadoop dfs -ls / Found 3 items
drwxrwxrwx   - hue    supergroup          0 2011-12-08 05:20 /tmp
drwxr-xr-x   - hue    supergroup          0 2011-12-08 05:20 /user
drwxr-xr-x   - mapred supergroup          0 2011-12-08 05:20 /var
$

到目前為止，一切都很好。您可以在 Pig 中看到您的 Hadoop 文件系統，所以，現在請嘗試從您的本地主機文件系統將一些數據讀取到 HDFS 中。可以通過 Pig 將某個文件從本地復制到 HDFS（參見 清單 5）。

               
$ pig 2011-12-10 06:39:44,276 [main] INFO  org.apache.pig.Main - Logging error messages to...
2011-12-10 06:39:44,601 [main] INFO  org.apache.pig.... Connecting to hadoop file \
system at: hdfs://0.0.0.0:8020
2011-12-10 06:39:44,988 [main] INFO  org.apache.pig.... connecting to map-reduce \
job tracker at: 0.0.0.0:8021
grunt> cd hdfs:/// grunt> ls hdfs://0.0.0.0/tmp     <dir>
hdfs://0.0.0.0/user    <dir>
hdfs://0.0.0.0/var     <dir>
grunt>

接下來，在 Hadoop 文件系統中測試數據現在是安全的，您可以嘗試另一個腳本。請注意，您可以在 Pig 內 cat 文件，查看其內容（只是看看它是否存在）。在這個特殊示例中，將確定在 passwd 文件中為用戶指定的外殼數量（在 passwd 文件中的最后一列）。

要開始執行該操作，需要從 HDFS 將您的 passwd 文件載入一個 Pig 關系中。在使用 LOAD 運算符之前就要完成該操作，但在這種情況下，您可能希望將密碼文件的字段解析為多個獨立的字段。在本例中，我們指定了 PigStorage 函數，您可以使用它來顯示文件的分隔符（本例中，是冒號 [:] 字符）。您也可以用 AS 關鍵字指定獨立字段（或架構），包括它們的獨立類型（參見 清單 6）。

               
grunt> mkdir test grunt> cd test grunt> copyFromLocal /etc/passwd passwd grunt> ls hdfs://0.0.0.0/test/passwd<r 1> 1728

接下來，使用 GROUP 運算符根據元組的外殼將元組分組到該關系中（參見 清單 7）。再次轉儲此關系，這樣做只是為了說明 GROUP 運算符的結果。注意，在這里，您需要根據元組正使用的特定外殼（在開始時指定的外殼）對元組進行分組（作為一個內部包）。

               
grunt> passwd = LOAD '/etc/passwd' USING PigStorage(':') AS (user:chararray, \
passwd:chararray, uid:int, gid:int, userinfo:chararray, home:chararray, \
shell:chararray); grunt> DUMP passwd; (root,x,0,0,root,/root,/bin/bash)
(bin,x,1,1,bin,/bin,/sbin/nologin)
...
(cloudera,x,500,500,,/home/cloudera,/bin/bash)
grunt>

但是，您想要的是在 passwd 文件中指定的獨特外殼的計數。所以，需要使用 FOREACH 運算符來遍歷分組中的每個元組，COUNT 出現的數量（參見 清單 8）。

               
grunt> grp_shell = GROUP passwd BY shell; grunt> DUMP grp_shell; (/bin/bash,{(cloudera,x,500,500,,/home/cloudera,/bin/bash),(root,x,0,0,...), ...})
(/bin/sync,{(sync,x,5,0,sync,/sbin,/bin/sync)})
(/sbin/shutdown,{(shutdown,x,6,0,shutdown,/sbin,/sbin/shutdown)})
grunt>

備注：如果要將該代碼作為一個腳本來執行，只需將腳本輸入到某個文件中，然后使用 pig myscript.pig 來執行它。

診斷運算符

Pig 支持大量診斷運算符，您可以用它們來調試 Pig 腳本。正如您在之前的腳本示例中所看到的，DUMP 運算符是無價的，它不僅可以查看數據，還可以查看數據架構。您還可以使用 DESCRIBE 運算符來生成一個關系架構的詳細格式（字段和類型）。

EXPLAIN 運算符更復雜一些，但也很有用。對于某個給定的關系，您可以使用 EXPLAIN 來查看如何將物理運算符分組為 Map 和 Reduce 任務（也就是說，如何推導出數據）。

表 2 對 Pig Latin 中的診斷運算符及其描述提供了一個列表。

               
grunt> counts = FOREACH grp_shell GENERATE group, COUNT(passwd); grunt> DUMP counts; ...
(/bin/bash,5)
(/bin/sync,1)
(/bin/false,1)
(/bin/halt,1)
(/bin/nologin,27)
(/bin/shutdown,1)
grunt>

用戶定義的函數

雖然 Pig 在本文探討的范圍內是強大且有用的，但是通過用戶定義的函數 (UDF) 可以使它變得更強大。Pig 腳本可以使用您為解析輸入數據、格式化輸出數據甚至運算符等定義的函數。UDF 是用 Java 語言編寫的，允許 Pig 支持自定義處理。UDF 是將 Pig 擴展到您的特定應用程序領域的一種方式。您可以在 參考資料 中了解有關 UDF 開發的更多信息。

Pig 用戶

正如您從這篇短文中可以看到的，Pig 是一個強大的工具，可以在 Hadoop 集群中查詢數據。它是如此強大，Yahoo! 估計，其 Hadoop 工作負載中有 40% 至 60% 由 Pig Latin 腳本產生。在 Yahoo! 的 100,000 個 CPU 中，大約有 50% 的 CPU 仍在運行 Hadoop。

但 Yahoo! 并不是利用 Pig 的惟一組織。您在 推ter 中也會發現 Pig（用于處理日志和挖掘微博數據）；在 AOL 和 MapQuest 上也會發現它（用于分析和批量數據處理）；而在 LinkedIn 上，Pig 用于發現您可能認識的人。據報道，Ebay 使用 Pig 來實現搜索優化，而 adyard 的推薦工具系統有大約一半都使用了 Pig。

展望未來

沒有一本書可以完全列舉 Pig 背后處理大數據的強大功能。即使對于非開發人員而言，Pig 也可以使得執行 Hadoop 集群上的大數據處理變得很容易。Pig 最初是由 Yahoo! 于 2006 年開發，并且此后不久被遷移到 Apache Software Foundation，使得它在全球范圍得到廣泛應用。進行這種遷移是因為 Yahoo! 研究人員意識到 Pig 能為非開發人員提供強大的功能。Hadoop 作為一個基礎架構已經逐漸開始普及，Hadoop 生態系統將會改變大數據的外觀及其日益增長的使用情況。