【正文】 而Pig的每一個(gè)基本操作都是獨(dú)立的，均不需要和其他基本操作發(fā)生任何聯(lián)系，每個(gè)操作都可以進(jìn)行自己的輸出。 alias = GROUP alias1 { ALL | BY expression} [USING 39。因?yàn)镻ig支持以字段序號(hào)的方式來訪問字段，因此這種方式并不為錯(cuò)。下面的圖示顯示了粒子操作劃分的基本思路：圖48 Pig Latin腳本拆解流程示例在具體的實(shí)現(xiàn)方面，同處理Hive一樣，對(duì)涉及到的基本操作進(jìn)行分類和篩選，分類的結(jié)果如下：1） 元數(shù)據(jù)操作類：相比于Hive要簡單需要，因?yàn)镻ig不保存元數(shù)據(jù)，臨時(shí)用臨時(shí)設(shè)置元數(shù)據(jù)。216。 CLUSTER BY操作： (備注：ORDER BY保證全部的輸出有序。以JOIN為例，用戶不再需要顯式地將小表放在左邊，也不用顯式地指定是否使用map端的JOIN，Hive會(huì)自動(dòng)探查表的大小從而將小表移到合適的位置或者在內(nèi)存夠用的情況下自動(dòng)進(jìn)行map端的JOIN。 SELECT * FROM table1 WHERE age20。 INSERT [OVERWRITE|INTO] TABLE tableName216。用戶應(yīng)該確保元數(shù)據(jù)的定義和實(shí)際數(shù)據(jù)結(jié)構(gòu)的一致性） ALTER TABLE tableName DROP PARTITION (key1=39。/location139。1. Hive的基本操作對(duì)于Hive，要將HQL腳本按照其執(zhí)行邏輯進(jìn)行語句的拆分，分離出粒度更細(xì)的基本操作，然后將這些基本操作對(duì)應(yīng)到功能組件上。如果編譯通過，那么在特定路徑下新建一個(gè)Jar文件，將主工程jar包里面除自定義MapReduce外的所有文件拷貝到該Jar文件。瀏覽器通過Web Server 與服務(wù)器端進(jìn)行交互。2) 并行算法層：包括內(nèi)置ETL操作并行算法、HQL和Pig Latin腳本執(zhí)行算法、自定義MapReduce執(zhí)行模塊以及表達(dá)式引擎。優(yōu)化規(guī)則的可擴(kuò)展方面則主要是設(shè)計(jì)合理的規(guī)則表示方法和實(shí)現(xiàn)機(jī)制，從而使得新規(guī)則的添加比較方便。216。Dispatcher實(shí)現(xiàn)類的構(gòu)造方法接收一個(gè)MapRule, NodeProcessor的參數(shù)，Rule為具體的優(yōu)化規(guī)則樣式，NodeProcessor為該樣式對(duì)應(yīng)的操作；Dispatcher實(shí)現(xiàn)類就是要從這個(gè)Map中選擇最優(yōu)的Rule，然后用對(duì)應(yīng)的NodeProcessor來對(duì)Rule中的樣式所對(duì)應(yīng)邏輯計(jì)劃中的子結(jié)構(gòu)進(jìn)行處理以完成優(yōu)化。依照MR執(zhí)行計(jì)劃中的依賴關(guān)系，MapReduce作業(yè)被按序提交到Hadoop集群并執(zhí)行。相對(duì)于Kettle，Talend對(duì)Pig進(jìn)行了比較深入的整合，它基于Pig設(shè)計(jì)開發(fā)了一套組件，不同的組件對(duì)應(yīng)于Pig中不同的語句，這些組件構(gòu)建的ETL流程被Talend自動(dòng)解析為Pig Latin腳本并通過Pig來執(zhí)行。 Hive的集成如前所述，Informatica PowerCenter在保證原有ETL流程構(gòu)建邏輯不變的情況下提供對(duì)Hive的集成，一個(gè)流程解析模塊負(fù)責(zé)將用戶構(gòu)建的ETL流程解析成HQL腳本并提交給Hive執(zhí)行。RunJar會(huì)通過URLClassLoader加載jar文件中的資源，并使用Java反射機(jī)制調(diào)用MainClass的main方法，調(diào)用時(shí)將hadoop jar命令中其他的參數(shù)作為main方法的實(shí)參。 //job配置信息(JobName)。要完成一個(gè)MapReduce作業(yè)，除了有完成相應(yīng)階段的類以外，還需要有一個(gè)驅(qū)動(dòng)程序，該驅(qū)動(dòng)程序?qū)υ撟鳂I(yè)進(jìn)行一些必要的參數(shù)配置，如輸入輸出路徑、Map階段或者Reduce階段要用到的參數(shù)等。Datastage內(nèi)嵌了一種類BASIC語言，可以寫一段處理程序來增加靈活性以滿足特殊需求，然而這也要求用戶對(duì)于這種特有的語言進(jìn)行額外的學(xué)習(xí)。本章還介紹了分布式計(jì)算技術(shù)，而重點(diǎn)是現(xiàn)今被很多公司和研究機(jī)構(gòu)應(yīng)用于實(shí)際工作中的Hadoop以及基于Hadoop的一些開源ETL工具。 元數(shù)據(jù)元數(shù)據(jù)為描述數(shù)據(jù)的數(shù)據(jù)，主要是描述數(shù)據(jù)屬性的信息，用來支持如指示存儲(chǔ)位置、歷史數(shù)據(jù)、資源查找、文件紀(jì)錄等功能。特定的優(yōu)化規(guī)則往往還會(huì)以使用手冊的形式告知用戶，以便用戶設(shè)計(jì)方案時(shí)作為參考。另外，Hive需要定義與數(shù)據(jù)相關(guān)的元數(shù)據(jù)并對(duì)這些元數(shù)據(jù)進(jìn)行管理。同樣的邏輯，用戶的代碼（腳本）編寫量大幅減小，同時(shí)，ETL方案可讀性大幅增加。MapReduce對(duì)兩個(gè)階段的讀寫操作和一些中間操作進(jìn)行了封裝，輸入數(shù)據(jù)分別以key,value和key,values的形式提供給用戶使用，用戶不用關(guān)心如何讀寫、如何分發(fā)、如何歸并等細(xì)節(jié)，而只用關(guān)心實(shí)際處理邏輯，也就是關(guān)注Mapper類的map方法和Reducer類的reduce方法的實(shí)現(xiàn)。通過MapReduce，開發(fā)人員可以很輕易地實(shí)現(xiàn)并行計(jì)算從而可以處理海量數(shù)據(jù)。通過NameNode，HDFS對(duì)外提供一個(gè)統(tǒng)一的訪問接口，在用戶看來，文件是一個(gè)統(tǒng)一的整體，可以以文件為單位進(jìn)行相應(yīng)的操作；而實(shí)際的文件則是被分割成若干數(shù)據(jù)塊，分布于同一集群中不同的DataNode上。上述框架實(shí)現(xiàn)了分布式計(jì)算和分布式存儲(chǔ)更高層次的抽象，使用戶不用關(guān)注復(fù)雜的內(nèi)部并行工作機(jī)制、不用具備豐富的分布式系統(tǒng)知識(shí)及開發(fā)經(jīng)驗(yàn)，即可實(shí)現(xiàn)大規(guī)模分布式系統(tǒng)的部署，以及海量數(shù)據(jù)的并行處理[13]。轉(zhuǎn)換主要是對(duì)抽取出來的數(shù)據(jù)進(jìn)行一系列的轉(zhuǎn)換操作，使之成為滿足目標(biāo)數(shù)據(jù)集中模型需求的形式。第三章是并行ETL工具可擴(kuò)展技術(shù)的研究：主要包括自定義MapReduce代碼嵌入、Hive和Pig的集成以及優(yōu)化規(guī)則可擴(kuò)展三個(gè)方面。本文希望在可用性方面擴(kuò)展現(xiàn)有ETL工具，使之能夠以圖形化的方式比較好地集成開源并行ETL腳本工具Hive和Pig。Hive和Pig也都設(shè)計(jì)了自己的優(yōu)化規(guī)則并設(shè)計(jì)了特定的實(shí)現(xiàn)機(jī)制以保證規(guī)則的可擴(kuò)展[10][11]。和Hive一樣，該工具提供了常用的關(guān)系操作、表達(dá)式操作和內(nèi)置函數(shù)，同時(shí)，支持用戶以UDF和嵌入自定義腳本的形式對(duì)工具進(jìn)行擴(kuò)展，以完成特定的ETL操作[4]。傳統(tǒng)ETL工具中，商用的都基于 C/S 模式，開源的本來是單機(jī)的，在向大數(shù)據(jù)處理方面做過相應(yīng)擴(kuò)展后也屬于C/S模式。正因?yàn)镋TL工具的重要性，市場對(duì)ETL工具擁有強(qiáng)烈的需求，ETL工具蘊(yùn)含的商業(yè)價(jià)值不容小覷，一些企業(yè)開發(fā)出功能強(qiáng)大的商用ETL工具并獲得廣泛的使用，這些企業(yè)也從中收獲頗豐。該方法包括數(shù)據(jù)的收集、整理、分析、挖掘等多個(gè)方面，最終的目的是從中獲得有價(jià)值信息并衍化出一種新的商業(yè)模式。本文開發(fā)基于Hadoop的、B/S模式的并行ETL工具，并研究如何實(shí)現(xiàn)對(duì)并行ETL工具進(jìn)行擴(kuò)展，完成了如下幾項(xiàng)主要工作：通過分析MapReduce并行計(jì)算框架的實(shí)現(xiàn)細(xì)節(jié)，設(shè)計(jì)實(shí)現(xiàn)了兩種方案來向現(xiàn)有工具中嵌入定制的MapReduce代碼以完成功能擴(kuò)展從而可以處理復(fù)雜的需求。同時(shí)，在大數(shù)據(jù)時(shí)代，ETL工具通過整合云計(jì)算技術(shù)以處理海量數(shù)據(jù)?；谶@種設(shè)計(jì)方式，可以很方便地?cái)U(kuò)展優(yōu)化規(guī)則。還有就是，查詢計(jì)劃的優(yōu)化廣泛的應(yīng)用于ETL工具中，如何設(shè)計(jì)優(yōu)化規(guī)則才能使得優(yōu)化規(guī)則具有良好的可擴(kuò)展性值得深入研究。用戶可以拖拽功能組件到相應(yīng)的工作區(qū)域并進(jìn)行參數(shù)配置，然后通過在組件之間連有向邊來設(shè)置執(zhí)行的先后關(guān)系以完成ETL流程的設(shè)計(jì)。開發(fā)人員基于Hadoop開發(fā)了不少針對(duì)大數(shù)據(jù)的并行ETL工具并且這些工具均具有一定程度的可擴(kuò)展性。Talend提供組件運(yùn)行HQL腳本和Pig Latin腳本，同時(shí)針對(duì)Pig開發(fā)了一套組件，支持拖拽工作流以定義ETL處理邏輯[6]。基于上述考慮和作者參與的項(xiàng)目的實(shí)際需求，本文設(shè)計(jì)實(shí)現(xiàn)的并行ETL工具采用了B/S的軟件架構(gòu)，相關(guān)的研究和開發(fā)工作也均是基于此種模式。本論文的主要成果包括：1) 實(shí)現(xiàn)一種自定義MapReduce代碼嵌入機(jī)制，包括兩個(gè)可定制組件，可以讓用戶實(shí)現(xiàn)將自己定義的MapReduce操作應(yīng)用于并行ETL工具中；2) 提供一整套Hive、Pig組件，支持用戶在圖形用戶界面中以構(gòu)建工作流的方式設(shè)計(jì)基于Hive或Pig的ETL工作流而不僅僅是局限于使用腳本，即對(duì)Hive和Pig進(jìn)行更深入的整合，對(duì)并行ETL工具進(jìn)行功能擴(kuò)展；3) 設(shè)計(jì)實(shí)現(xiàn)了一套規(guī)則表示和擴(kuò)展機(jī)制，方便用戶在使用過程中將提煉的規(guī)則集成到現(xiàn)有系統(tǒng)中。具體而言：抽取主要是針對(duì)不同的數(shù)據(jù)源，充分理解數(shù)據(jù)的定義，設(shè)計(jì)不同的數(shù)據(jù)抽取方式，進(jìn)行必要的數(shù)據(jù)類型轉(zhuǎn)換，選擇合適的數(shù)據(jù)傳輸機(jī)制，如數(shù)據(jù)庫直連、通過FTP接口機(jī)等。ETL的方式即先從不同的數(shù)據(jù)源中抽取數(shù)據(jù)，放到ETL工具中來，ETL工具完成相應(yīng)的轉(zhuǎn)化后將處理的結(jié)果載入到目標(biāo)數(shù)據(jù)集中（通常為數(shù)據(jù)倉庫）。一個(gè)Hadoop集群主要包括一個(gè)用于管理這個(gè)文件系統(tǒng)的目錄節(jié)點(diǎn)和多個(gè)用于存儲(chǔ)實(shí)際數(shù)據(jù)的數(shù)據(jù)節(jié)點(diǎn)。同時(shí)，HDFS還提供一些與本地文件系統(tǒng)進(jìn)行交互的功能，如從本地加載數(shù)據(jù)到HDFS和從HDFS獲取數(shù)據(jù)到本地等。reduce方法包含reduce端的主要處理邏輯，每次處理一個(gè)key，Itoratorvalue形式的對(duì)象，其結(jié)果被輸出到HDFS。這種方式因?yàn)閷?duì)于處理邏輯和數(shù)據(jù)結(jié)構(gòu)的了解，可以編寫最有針對(duì)性的代碼進(jìn)而可以實(shí)現(xiàn)較優(yōu)的性能，但是開發(fā)工作量大，調(diào)試過程復(fù)雜，對(duì)于多樣的需求在工作量、準(zhǔn)確性和靈活性方面均無法滿足。相對(duì)于自己編寫MapReduce代碼，其優(yōu)點(diǎn)是學(xué)習(xí)成本低，代碼量小，用戶可以通過少數(shù)的HQL語句快速實(shí)現(xiàn)較為復(fù)雜的MapReduce數(shù)據(jù)處理邏輯。對(duì)同樣的數(shù)據(jù)，不同的應(yīng)用中可以指定不同的元數(shù)據(jù)，顯得比較靈活。為了簡化工作，本論文采用開源的表達(dá)式引擎。另外，這些商用ETL工具均采用了C/S架構(gòu)，而開源ETL工具在提供大數(shù)據(jù)支持之后也屬于一種C/S架構(gòu)。本章重點(diǎn)進(jìn)行并行ETL工具可擴(kuò)展技術(shù)的研究，結(jié)合實(shí)際應(yīng)用需求，提出可行的在并行ETL工具中可以應(yīng)用的可擴(kuò)展技術(shù)并總結(jié)為如下三個(gè)主要方面：可擴(kuò)展組件技術(shù)、開源ETL工具集成技術(shù)和優(yōu)化規(guī)則可擴(kuò)展技術(shù)。由此可以看出，以提供可擴(kuò)展組件的方式來實(shí)現(xiàn)擴(kuò)展ETL工具功能的方式較為普遍。拷貝到HDFS上的這些資源文件又會(huì)被分配了任務(wù)的從節(jié)點(diǎn)拷貝到本地，進(jìn)而構(gòu)造啟動(dòng)JVM的參數(shù)，然后啟動(dòng)一個(gè)新的JVM來運(yùn)行任務(wù)。 //job配置信息setJar()方法接受一個(gè)String類型的參數(shù)，表示jar包對(duì)應(yīng)的本地路徑。為了添加處理大數(shù)據(jù)的能力，傳統(tǒng)ETL工具基本選擇了集成已有并行ETL工具的方式，這樣比較快捷，傳統(tǒng)ETL工具主要關(guān)注如何集成，而不必再進(jìn)行并行ETL模塊的開發(fā)。出于性能的考慮，商用ETL工具都沒有選擇Pig作為其底層執(zhí)行引擎，而開源的Kettle和Talend均選擇了對(duì)Pig進(jìn)行集成。經(jīng)過優(yōu)化的邏輯計(jì)劃又被轉(zhuǎn)換成物理執(zhí)行計(jì)劃，物理執(zhí)行計(jì)劃再經(jīng)過一次優(yōu)化后，特定的拆分規(guī)則被應(yīng)用，物理計(jì)劃被劃分成一個(gè)一個(gè)的階段，分別對(duì)應(yīng)MapReduce的Map和Reduce階段。之后會(huì)按序調(diào)用這些優(yōu)化規(guī)則的transform方法實(shí)現(xiàn)優(yōu)化[15]。2. ，在PlanOptimizer的子類LogicalPlanOptimizer中，規(guī)則集列表被實(shí)例化為一個(gè)ArrayList，優(yōu)化規(guī)則集被按一定順序添加到該列表中。例如，LogicalExpressionSimplifier對(duì)應(yīng)優(yōu)化規(guī)則FilterLogicExpressionSimplifier，它只針對(duì)FILTER操作，所以其buildPattern方法只包含了一個(gè)LOFilter。 圖41可擴(kuò)展并行ETL工具系統(tǒng)架構(gòu)圖系統(tǒng)架構(gòu)由下至上分為四個(gè)部分：1) 平臺(tái)層：Hadoop集群，提供云存儲(chǔ)（HDFS）和并行計(jì)算（MapReduce）支持，以Hadoop API的方式供并行算法層調(diào)用。界面層與應(yīng)用邏輯層通過HttpService的方式交互。1. 前臺(tái)實(shí)現(xiàn)前臺(tái)界面接收與MapReduce作業(yè)相關(guān)的參數(shù)，包括map數(shù)、reduce數(shù)、是否使用Combiner或Reducer以及自定義的MapReduce Java代碼等。圖45 自定義MR Jar組件前臺(tái)界面2. 后臺(tái)實(shí)現(xiàn)后臺(tái)接收前臺(tái)傳過來的參數(shù)和jar文件，將jar文件放到特定的目錄，并利用自定義的類加載器加載jar包中的類，然后利用Java反射機(jī)制運(yùn)行驅(qū)動(dòng)類中的main方法。216。, key2=39。) location 39。（備注：外部表只刪表結(jié)構(gòu)，內(nèi)部表會(huì)同時(shí)刪除數(shù)據(jù)；該操作可以作為選項(xiàng)提供給用戶，在流程結(jié)束時(shí)根據(jù)此選項(xiàng)配置決定是否刪除表以及數(shù)據(jù)）2） 實(shí)際操作類：實(shí)際完成對(duì)數(shù)據(jù)進(jìn)行處理所涉及的操作。 SELECT aggKey1,aggKey2,…,aggExpression1, aggExpression2… FROM tableName GROUP BY aggKey1,aggKey2,…。 SELECT FROM table1 a LEFT SEMI JOIN table2 b on (=)。(備注：select_Statement表示SELECT查詢語句)216。 SELECT * FROM table1 SORT BY id [ASC|DESC] (備注：SORT BY只保證每個(gè)Task的輸出有序)216。 參數(shù)設(shè)置：SET =strict。 alias = FILTER alias1 BY condition_expression。 alias = COGROUP alias1 { ALL | BY expression}[INNER], alias2 { ALL | B。merge39。216。data39。 MapReduce腳本： SELECT col1, col2 FROM table1 LIMIT limit_num。此時(shí)ORDER BY必須指定 LIMIT，否則執(zhí)行會(huì)報(bào)錯(cuò))216。 UNION ALL操作： SELECT , FROM table1 a {LEFT|RIGHT|FULL} [OUTER] JOIN table2 b on (=)。216。 刪除表：value239。, key2=39。 ALTER TABLE tableName ADD PARTITION (key1=39。 建表：3. 處理流程圖44 自定義MR Java組件執(zhí)行流程 自定義MR Jar組件ETL工具不提供驅(qū)動(dòng)程序，