【正文】 educe作業(yè)相關的參數，包括map數、reduce數、是否使用Combiner或Reducer以及自定義的MapReduce Java代碼等。圖43自定義MR Java組件前臺界面2. 后臺實現后臺接收前臺傳過來的代碼，將其寫入到特定的目錄下面，生成Java源代碼文件。通過JavaCompiler進行編譯時可以指定一個輸出源，編譯過程中出現的錯誤信息將被寫入到這個輸出源。如果編譯通過，那么在特定路徑下新建一個Jar文件，將主工程jar包里面除自定義MapReduce外的所有文件拷貝到該Jar文件。構建完畢后，發(fā)送成功的信息給前臺，而這個Jar文件的路徑通過setJar()方法指定給MapReduce任務。（說明：主工程jar包中包含了自定義MR Java代碼的Driver、Mapper、）如果編譯不通過，讀取上述輸出源中的錯誤信息并發(fā)送給前臺展示。3. 處理流程圖44 自定義MR Java組件執(zhí)行流程 自定義MR Jar組件ETL工具不提供驅動程序，而是將驅動程序和Mapper、Reducer等類一起編譯打包，后臺接收后解析jar包，查找指定的驅動程序類并執(zhí)行其main方法。該實現機制類似于在外殼環(huán)境下使用如下命令：hadoop jar arguments1. 前臺實現前臺界面接收MapReduce任務相關的參數，包括map數、reduce數以及jar包地址等。圖45 自定義MR Jar組件前臺界面2. 后臺實現后臺接收前臺傳過來的參數和jar文件，將jar文件放到特定的目錄，并利用自定義的類加載器加載jar包中的類，然后利用Java反射機制運行驅動類中的main方法。3. 處理流程圖46 自定義MR Jar組件執(zhí)行流程 并行ETL工具的集成對Hive和Pig的集成主要考慮以下幾個方面的問題：功能組件的設計、元數據處理的遷移和管理、工作流的解析。 功能組件的實現針對Hive和Pig各自不同的語言特點，設計不同的功能組件，使得由這些功能組件構建的ETL工作流（即一個有向無環(huán)圖）可以被解析成具有相同邏輯的Hive或Pig腳本并能提交給執(zhí)行引擎執(zhí)行。1. Hive的基本操作對于Hive，要將HQL腳本按照其執(zhí)行邏輯進行語句的拆分，分離出粒度更細的基本操作，然后將這些基本操作對應到功能組件上。下面的圖示顯示了粒子操作劃分的基本思路：圖47 HQL腳本拆解流程示例在具體的實現方面，對Hive的基本操作進行分類和篩選，分類的結果如下：1） 元數據操作類：主要包括建表、修改表結構、刪除表三種語句。216。 建表： CREATE [EXTERNAL] TABLE tableName(col1 type1, col2 type2,…) ROW FORMAT DELIMITED FIELDS TERMINATED BY ‘,’ LINE TERMINATED BY ‘\n’ LOCATION ‘/dataDir’ [PARTITIONED BY (key1 type1, key2 type2,…)]。216。 修改表結構： ALTER TABLE tableName RENAME TO newname。 ALTER TABLE tableName ADD COLUMNS (col3 type3, col4 type4)。 ALTER TABLE tableName ADD PARTITION (key1=39。value139。, key2=39。value239。) location 39。/location139。 PARTITION (key1=39。value339。, key2=39。value439。) location 39。/ location239。 ALTER TABLE tableName DROP PARTITION (key1=39。value139。, key2=39。value239。)。 ALTER TABLE tableName CHANGE [COLUMN] col_old_name col_new_name column_type [COMMENT col_ment] [FIRST|AFTER column_name]。 （備注：對列的改變只會修改 Hive 的元數據，而不會改變實際數據。用戶應該確保元數據的定義和實際數據結構的一致性） 考慮到實際應用中修改表結構的操作并不常用，因此暫不考慮這部分操作。216。 刪除表： DROP TABLE tableName。（備注：外部表只刪表結構，內部表會同時刪除數據；該操作可以作為選項提供給用戶，在流程結束時根據此選項配置決定是否刪除表以及數據）2） 實際操作類：實際完成對數據進行處理所涉及的操作。216。 INSERT 操作： INSERT [OVERWRITE|INTO] TABLE tableName216。 SELECT操作： SELECT [DISTINCT] column_exp1, column_exp1,… from table1。216。 GROUP BY操作： SELECT aggKey1,aggKey2,…,aggExpression1, aggExpression2… FROM tableName GROUP BY aggKey1,aggKey2,…。216。 WHERE操作： SELECT * FROM table1 WHERE age20。216。 JOIN操作： SELECT , FROM table1 a {LEFT|RIGHT|FULL} [OUTER] JOIN table2 b on (=)。 SELECT FROM table1 a LEFT SEMI JOIN table2 b on (=)。(備注：這種類型的JOIN操作的應用場景類似于SQL中IN后面跟一個子查詢的場景，其SELECT子句中不能有右表的字段存在。另外，隨著Hive優(yōu)化機制的日益完善，一些自動化的識別技術被應用于對系統(tǒng)資源和輸入數據規(guī)模等信息的探查，進而基于這些信息進行查詢計劃的優(yōu)化。以JOIN為例，用戶不再需要顯式地將小表放在左邊，也不用顯式地指定是否使用map端的JOIN，Hive會自動探查表的大小從而將小表移到合適的位置或者在內存夠用的情況下自動進行map端的JOIN。盡管目前像SELECT /*+ MAPJOIN(things) */ sales.*, things.* FROM salse JOIN things ON (=) 這樣的語句還可以使用并且在某些情況下生成的MR查詢計劃仍要優(yōu)于自動優(yōu)化生成的MR查詢計劃，但是自動優(yōu)化技術的優(yōu)勢已經顯現，對于相對簡單的查詢，自動優(yōu)化的結果已經完全滿足需求，因此，本論文將不考慮在組件中添加對這樣一種形式的查詢語句的支持。)216。 UNION ALL操作： select_Statement UNION ALL select_Statement [UNION ALL select_Statement]。(備注：select_Statement表示SELECT查詢語句)216。 ORDER BY操作： SELECT * FROM table1 ORDER BY id [ASC|DESC]。 (備注：ORDER BY保證全部的輸出有序。 = strict 模式下，一些危險的查詢操作不能進行。因為ORDER BY為了保證全部的輸出有序，map端過來的所有數據都會被傳給同一個Reducer進行reduce操作，如果在數據量大的情況下會出現負荷過重無法輸出結果的情況，因此屬于危險操作。此時ORDER BY必須指定 LIMIT，否則執(zhí)行會報錯)216。 SORT BY操作： SELECT * FROM table1 SORT BY id [ASC|DESC] (備注：SORT BY只保證每個Task的輸出有序)216。 DISTRIBUTE BY操作： SELECT * FROM table1 DISTRIBUTE BY id；(備注：將具有同樣id的記錄放到同一個節(jié)點上)216。 CLUSTER BY操作： SELECT * FROM table1 CLUSTER BY id；(備注：相當于SORT BY和DISTRIBUTE BY一起對同樣的字段進行操作；因為ORDER BY、SORT BY、DISTRIBUTE BY和CLUSTER BY功能上有重合，所以將它們合并到一個組件中并在組件中內置其相互關系，保證用戶不會同時使用有沖突的操作)216。 LIMIT操作： SELECT col1, col2 FROM table1 LIMIT limit_num。3） 擴展功能類：216。 參數設置：SET =strict。216。 添加UDF的jar文件：ADD JAR ‘/path/’。216。 創(chuàng)建臨時函數名：CREATE TEMPORARY FUNCTION funcName AS ‘’。216。 MapReduce腳本： 添加腳本文件：ADD FILE ‘/path/’。 使用腳本文件：TRANSFORM操作和MAP、REDUCE操作均被設計用來使用腳本進行數據處理。(備注：因為是基于Hadoop Streaming技術實現，而Streaming的性能相對于寫UDF要低不少，所以暫時不予考慮)基于上述分析，本文共實現10個Hive組件，其名稱及配置參數如下表所示：表41 Hive組件及配置參數示意表操作名參數參數意義Readlocation輸入數據路徑fieldSep字段分隔符lineSep行分隔符isInternal是否內部表partitionKeys分區(qū)鍵tableName表名fieldsInfo字段信息，包括字段名和字段類型Setparameters參數名和對應的參數值addPaths添加的資源對應的文件路徑tempFunctions臨時函數名以及對應的UDF類SelectoutputExpressions輸出字段信息，包括表達式、表達式別名和表達式返回值類型useDistinct是否去重WhereconditionExpression過濾表達式GroupByoutputInfo輸出字段信息，包括表達式、表達式別名和表達式返回值類型aggKeys聚集鍵值JoinoutputInfo輸出字段信息，包括表達式、表達式別名和表達式返回值類型joinType關聯類型，包括內連接、外連接和左半連接joinExpression關聯表達式，=SortByorder_sortExpressionOrder By或者Sort By表達式信息cluster_distributeExpressionCluster By或者Distribute By表達式信息LimitlimitNum輸出條數UnionAlloutputInfo輸出字段信息，包括字段名和字段類型Storelocation輸入數據路徑fieldSep字段分隔符lineSep行分隔符isInternal是否內部表partitionKeys分區(qū)鍵tableName表名fieldsInfo字段信息，包括字段名和字段類型overwrite是否覆蓋原來的數據2. Pig的基本操作對于Pig，因為Pig Latin本身屬于流程式語言，所以只需將其實現不同功能的語句對應成不同的功能組件即可，然后按照該語句的語法結構，為功能組件配置對應的參數輸入接口即可。下面的圖示顯示了粒子操作劃分的基本思路：圖48 Pig Latin腳本拆解流程示例在具體的實現方面，同處理Hive一樣，對涉及到的基本操作進行分類和篩選，分類的結果如下：1） 元數據操作類：相比于Hive要簡單需要，因為Pig不保存元數據，臨時用臨時設置元數據。216。 設置元數據：alias=LOAD 39。data39。 [USING function] [AS schema]。 USING子句用來指定序列化和反序列化的方式，可使用內嵌的也可以使用自定義的； schema在省略的情況下會根據默認的行分隔符和字段分隔符以及默認的數據類型來解析數據，同時生成的元數據也沒有字段名信息。因為Pig支持以字段序號的方式來訪問字段，因此這種方式并不為錯。但是這種靈活性在字段數較多邏輯較復雜的時候容易出錯，因此本次集成將嚴格定義元數據。2） 實際操作類：實際完成查詢操作涉及的語句。216。 FILTER操作： alias = FILTER alias1 BY condition_expression。216。 GROUP操作： alias = GROUP alias1 { ALL | BY expression} [USING 39。collected39。 | 39。merge39。][PARTITION BY partitioner] [PARALLEL n]。 區(qū)別于Hive的GROUP BY操作，Pig的GROUP操作確實只完成了聚集的操作，并沒有進行匯總。Hive的絕大多數基本操作都離不開SELECT操作，因此，其GROUP BY必須借助SELECT才能完成聚集匯總的操作。而Pig的每一個基本操作都是獨立的，均不需要和其他基本操作發(fā)生任何聯系，每個操作都可以進行自己的輸出。216。 COGROUP操作： alias = COGROUP alias1 { ALL | BY expression}[INNER], alias2 { ALL