Sqoop開發(fā)者指南

1、Sqoop開發(fā)者指南目錄1 . 引言12支持的發(fā)行版 .13Sqoop 發(fā)行版 .14先決條件 .1I5Sqoop 源碼.26開發(fā)者 API 參考 .26.1外部 API .26.2擴展 API .36.2.1的.錯誤！未定義書簽。6.3Sqoop 內部實現(xiàn) .56.3.1 .56.3.2的.錯誤！未定義書簽。6.3.3Hadoop 大數(shù)據(jù)處理技術架構 .錯誤！未定義書簽。II1 引言如果你是一個開發(fā)者或應用程序的程序員打算修改的 Sqoop或通過使用 Sqoop 內部 API 進行擴展，你應該閱讀此文檔。 以下各節(jié)描述的每一個 API 的作用，包括內部 API 和支持其他數(shù)據(jù)庫必須的 API

2、。2 支持的發(fā)行版本文檔適用于 Sqoop V1.4.2 。3 Sqoop 發(fā)行版Apache的 Sqoop是 Apache軟件基金會的一個開放源碼的軟件產品。Sqoop 產品的發(fā)展在： /repos/asf/sqoop/trunk。在該網站上，你可以得到： 最新版本的源代碼 An issue tracker 包含 WIKI 的 Sqoop 文檔4 先決條件Sqoop開發(fā)需要如下先決知識： JAVA開發(fā)：熟練掌握 JDBC熟練掌握 Hadoop的 API（包括新的 MapReduce 0.2 版本以上的 API）1 關系型數(shù)據(jù)庫和 SQL本文檔假定您使用

3、的是Linux 或類似 Linux 的環(huán)境。如果你使用的是Windows，您可能能夠使用cygwin 完成以下任務。如果你使用的是Mac OS X，你應該看到一些兼容性錯誤（如果有的話）。Sqoop 主要在 Linux 環(huán)境上進行測試。5 Sqoop 源碼您可以從如下網站上獲取Sqoop的源代碼：/repos/asf/sqoop/trunkSqoop源代碼被保存在一個“ Git ”倉庫中，從倉庫中獲取 Sqoop源碼的操作說明被提供在如上網站中提供的一個“ TODO”頁面上。編譯指令由源碼根目錄下的“COMPILING.txt ”文件提供。6 開發(fā)者 A

4、PI 參考本章節(jié)的內容提供給那些需要整合或者擴展Sqoop 的應用程序開發(fā)者。下面三個部分都是為了接下來的用例而寫的： 使用 Sqoop生成的類和公共類 Sqoop擴展（即，于更多數(shù)據(jù)庫交互的額外的ConnManager的實現(xiàn)） 修改 Sqoop內部實現(xiàn)每一部分逐步深入的描述Sqoop系統(tǒng)。6.1 外部 APISqoop在執(zhí)行的時候自動生成將關系型數(shù)據(jù)庫里表導入（import ）到 HDFS系統(tǒng)中的 java 類，這個類包含導入表的每一列的成員字段，這個類的一個實例保存導入表的每一行（每一行為一個實例），這個類通過Hadoop實現(xiàn)序列化 API，命名為 Writable和 DBWritable

5、接口。并且包含其他比較方便的方法： parse() ：可以解析帶分隔符文本的方法。 toString() ：可以保留用戶選擇的分隔符的方法2確保自動生成的類中的所有方法都存在于下面的抽象類中：com.cloudera.sqoop.lib.SqoopRecord.SqoopRecord實例可能依賴于 Sqoop的公共 API。所有的類都存在于 com.cloudera.sqoop.lib 包中。這些被簡要的描述如下。 Sqoop 客戶端并不需要直接與這些類發(fā)生關系， 盡管由 Sqoop生成的類依賴于他們。 因此，這些 AOP被認為是公開的，并且需要繼續(xù)優(yōu)化的。 RecordParser 類將解析

6、文本文件中的一行到一個字段列表中， 使用可控的分隔符和引號字符。 靜態(tài)類 FieldFormatter 提供了一個被用于 SqoopRecord.toString() 的方法，該方法處理引用和轉義字符。 封裝 ResultSet 、PreparedStatement objects 、SqoopRecords 之間的數(shù)據(jù)是通過 JdbcWritableBridge 實現(xiàn)的。 BigDecimalSerializer 在 Writable 接口之上包含了一組對 BigDecimal（小數(shù)）對象序列化的方法。6.2 擴展 API6.2.1HBase 的序列化擴展本章節(jié)包含了 API 和最基本的用于

7、擴展Sqoop的類，這些類使得 Sqoop 可以與更多的數(shù)據(jù)庫提供商做接口。Sqoop使用 JDBC和 DataDrivenDBInputFormat 的從數(shù)據(jù)庫中讀取數(shù)據(jù)，對不同的數(shù)據(jù)庫提供商及 JDBC元數(shù)據(jù)之間的不同，必須為大多數(shù)數(shù)據(jù)庫提供特定的代碼路徑。 Sqoop針對這個問題的解決方案是引入 ConnManager API(com.cloudera.sqoop.manager.ConnMananger)。ConnManager是一個靜態(tài)類，這個類定義了數(shù)據(jù)庫本身相關作用的方法。ConnManager類的大多數(shù)實現(xiàn)都繼承靜態(tài)類SqlManager（ com.cloudera.sqoop

8、.manager.SqlManager ），SqlManager 使用標準的 SQL執(zhí)行大部分的動作。 所有的子類需要實現(xiàn) getConnection() 方案，用于獲取實際的 JDBC數(shù)據(jù)庫鏈接，子類可以重寫其他所有方法。SqlManager 類本身暴露了一個受保護的 API，使開發(fā)人員可以有選擇地重寫行為。例如 getColNamesQuery() 方法允許使3用 getColNames() 的進行修改 SQL查詢，而不需要重寫大多數(shù)的getColNames() 方法。ConnManager的實現(xiàn)通過 Sqoop特定的類（ SqoopOptions ）獲得了大量的配置數(shù)據(jù)，SqoopOpt

9、ions 是可變的。 SqoopOptions 不直接存儲特定的管理信息，相反，他包含了對配置信息的一個指向，通過GenericOptionsParser解析命令行參數(shù)之后由 Tool.getConf() 返回。這使得通過“ - D any.specific.param= any.value ” 擴展參數(shù)，而不需要任何層次的的解析參數(shù)或修改 SqoopOptions 。配置信息中的基礎配置通過調用工作流交給任意的 MapReduce作業(yè)，所以，用戶可以在命令行自定義設置必要的 Hadoop狀態(tài)。目前所有的 ConnManager實現(xiàn)都是無狀態(tài)的，因此，ConnManagers示例的系統(tǒng)在 Sq

10、oop 的生命周期中對同一個ConnManager類實現(xiàn)了多個實例。目前實例化一個 ConnManagers類是一個輕量級的操作，并且很少這樣做。因此ConnManagers不對 operations之間進行緩存。當前的 ConnManagers是由靜態(tài)類 ManagerFactory 的一個實例創(chuàng)建的（參考：/jira/browse/MAPREDUCE-750）。當前的ManagerFactory 的一個實現(xiàn)服務于整個Sqoop：com.cloudera.sqoop.manager.DefaultManagerFactory。Sqoop 的相關擴

11、展不應該修改 DefaultManagerFactory。相反，ManagerFactory 的一個特定的擴展需要由新的 ConnManager提供。值得注意的是 ManagerFactory 有一個單一的名為 accept()的方法，這個方法將確定是否可以實例化為用于用戶輸入信息（user sSqoopOptions ）的 ConnManager。如果可以，將返回一個ConnManager實例，否則返回 null 。ManagerFactory 實例的使擁由設置在sqoop-site.xml中的sqoop.connection.factories支配，擴展庫的用戶可以安裝第三方庫，其中包含一

12、個新的 ManagerFactory 和 ConnManager（ S），配置 sqoop-site.xml使用新的ManagerFactory 。DefaultManagerFactory principly通過存儲在 SqoopOptions中的連接字符串區(qū)分數(shù)據(jù)庫。擴展者可以利用 MapReduce，com.cloudera.sqoop.io和 util包中的類以方便他們的擴展實現(xiàn)。下面一節(jié)中更詳細地描述這些包和類。4Sqoop支持從關系型數(shù)據(jù)庫到HBase的導入操作。當導入數(shù)據(jù)到 HBase的時候，必須轉化成 HBase可以接受的特定的形式： 數(shù)據(jù)必須放到 HBase中的一個或多個表中

13、。 輸入數(shù)據(jù)的列必須到放置到一個column family鐘。 值必須序列化成字節(jié)數(shù)組存放到單元格內所有這些都通過 HBase客戶端 API 中的 Put statements 實現(xiàn)。Sqoop于 HBase的交互執(zhí)行都在com.cloudera.sqoop.hbase包里。記錄從數(shù)據(jù)庫中被deserialzed ，并從 mapper 任務中發(fā)出。 OutputFormat 負責把結果插入到 HBase 中，這是通過一個叫做 PutTransformer 的接口實現(xiàn)的。 PutTransformer 有一個方法稱為 getPutCommand（），它使用一個 Map等領域的數(shù)據(jù)集作為輸入。它返

14、回列表 ，描述了如何將單元格插入到 HBase中。默認的 PutTransformer 實現(xiàn)是 ToStringPutTransformer ，使用基于字符串的方式將每個字段序列化到 HBase中。你可以通過自己的PutTransformer重新這個實現(xiàn)，并且把他添加到map任務的 classpath中，來告訴 Sqoop你將使用你自己的實現(xiàn)， 通過” -D”命令設置自己的 class:sqoop.hbase.insert.put.transformer.class。在您的 PutTransformer實現(xiàn)，特定的 row key、column 、 column family可通過 getRo

15、wKeyColumn（）和 getColumnFamily （）方法。你可以擴展更多的超出限制的” PUT”操作，例如，注入額外的輔助索引的行。 Sqoop 將對 -hbase-table 命令特定的表執(zhí)行所有的” PUT”操作。6.3 Sqoop 內部結構本節(jié)介紹 Sqoop 的內部結構。Sqoop程序是由的主類 com.cloudera.sqoop.Sqoop 驅動。有限數(shù)量的額外的類在同一個包： SqoopOptions （如前所述）， ConnFactory （即熟練操作 ManagerFactory 實例）。6.3.1一般程序流程一般的程序流程如下：5com.cloudera.sqo

16、op.Sqoop是主類和實現(xiàn)了Tool ，一個新的實例 ToolRunner被推出， Sqoop的第一個參數(shù)是一個字符串，并在SqoopTool 中定義并執(zhí)行，SqoopTool 來執(zhí)行用戶的各種請求操作（如：import, export, codegen等）。SqoopTool 將解析其余的參數(shù)，設置相應的字段在SqoopOptions 類，然后執(zhí)行。在 SqoopTool 的 run() 方法中， import 、 export 或者其他正確的指令被執(zhí)行。一般情況下 ConnManager一般在 SqoopOptions 數(shù)據(jù)的基礎上進行實例化。ConnFactory 被用于從 Manag

17、erFactory 中獲取一個 ConnManager。這在前面的部分已經進行了描述。 Imports 、 exports 或者其他大型數(shù)據(jù)的遷移操作通常是一個并行的、可靠的 MapReduce任務。 Import 操作并不是一定要以 MapReduce作業(yè)的方式運行， ConnManager.importTable （）將確定如何以最佳的方式進行 import操作。每一個主要操作實際上都有ConnMananger控制，除了生成代碼的操作，這些操作是由 CompilationManager 和 ClassWriter 做的 ( 都在 com.cloudera.sqoop.orm 包中 ) 。導

18、入數(shù)據(jù)到 Hive 的操作是由com.cloudera.sqoop.hive.HiveImport的 importTable()完成的，這樣做是為了不為使用 ConnManager的實現(xiàn)都擔心。ConnManager 的importTable （）方法接收一個類型為ImportJobContext的參數(shù)，其中包含這個方法所需的各種參數(shù)值。將來，該類可擴展附加參數(shù)，以實現(xiàn)功能更加強大的import 操作。同樣，exportTable （）方法接收的參數(shù)類型ExportJobContext 。這些類包含一系列導入 / 導出，指向 SqoopOptions 對象，以及其他相關的數(shù)據(jù)。6.3.2子包c

19、om.cloudera.sqoop包中的子包，包括： Hive ：便于數(shù)據(jù)導入到 Hive IO:實現(xiàn)的 java.io. *接口 Lib: 外部公共 API（如前所述） Manager: ConnManager 和 ManagerFactory 的接口和它們的實現(xiàn) Mapreduce: 新的（ 0.20 + ）MapReduce的 API 接口的類6 Orm: 代碼自動生成 Tool: 實現(xiàn) SqoopTool Util: 各種實用工具類IO 包中的 OutputStream 和 BufferedWriter的實現(xiàn)被用于直接向HDFS進行寫入。 SplittableBufferedWrite

20、r允許為客戶端打開一個單一的BufferedWriter，在引擎之下，當達到目標值后連續(xù)的寫入文件系統(tǒng)。這允許在Sqoop import的同時使用壓縮機制（如 gzip ）, 同時允許在 MapR任務之后對數(shù)據(jù)集進行分割。大對象文件存儲系統(tǒng)的代碼也存在于 IO 包中。Mapreduce包中的代碼用于直接與Hadoop MapReduce做接口，將在下一章節(jié)中詳述。ORM包中的代碼用于生產代碼。 它依賴于提供了 com.sun.tools.javac包的 JDK的 tools.jar包。UTIL 包中包含用于整個Sqoop的各種工具 ClassLoaderStack ：管理由當前線程的堆棧使用的

21、 ClassLoader 的實例，這正是當以本地模式運行 MapReduce任務時自動生成代碼寫入當心線程的原理。 DirectImportUtils ：包含一些直接進行 Hadoop import 操作的方便方法。 Executor ：啟動外部進程，并連接這些來生成由一個AsyncSink （見下面更詳細地）的流處理程序。 ExportException ：當 exports 失敗時由 ConnManagers拋出異常。 ImportException ：當 import 失敗時由 ConnManagers拋出異常。 JDBCURL：處理連接字符串的解析，這是類 URL，是不規(guī)范的、不合格的

22、。 PerfCounters ：被用來估計用于顯示給用戶的傳輸速率。 ResultSetPrinter ：漂亮地打印結果集。在不同的時候， Sqoop從外部程序中讀取stdout ，最簡單的模式就是由LocalMySQLManager和 DirectPostgresqlManager執(zhí)行的直接模式 （direct-mode ）import 。之后由 Runtime.exec() 產生一個進程，該進程的標準輸出(Process.getInputStream()和潛在錯誤 (Process.getErrorStream()都需要被7處理。在這些流之間無法讀取更多的數(shù)據(jù)從而導致在寫入更多數(shù)據(jù)之前外部

23、進程阻塞。因此，這些都必須處理，最好以異步的方式。按照 Sqoop 的說法，“異步接收器”是一個線程需要一個InputStream和讀取完成。這些實現(xiàn) AsyncSink 實現(xiàn)。 com.cloudera.sqoop.util.AsyncSink抽象類定義了這家工廠必須執(zhí)行的操作。processStream （）將產生另一個線程立即開始處理從 InputStream中讀取的數(shù)據(jù)參數(shù) ;它必須讀這流來完成的。join（）方法允許外部線程等待，直到處理完成。一些 stock 被同步實現(xiàn)： LoggingAsyncSink 將重復 InputStream 上的一切在log4j INFO語句中。 Nu

24、llAsyncSink消耗所有的輸入和什么都不做。各種 ConnManagers使得外部進程以內部類的方式擁有它們自己的 AsyncSink 實現(xiàn)，它們通過數(shù)據(jù)庫 tools 讀取，并且促使數(shù)據(jù)流向 HDFS，有可能在此期間進行格式轉換。6.3.3與 MapReduce 的接口Sqoop調度 MapReduce作業(yè)產生 imports 和 exports 。配置和執(zhí)行的 MapReduce 工作如下幾個共同的步驟 （配置 InputFormat 配置 OutputFormat 設置映射的實施 ; 等等）。這些步驟是在 com.cloudera.sqoop.mapreduce.JobBase 類中的。 為了使用， JobBase 允許一個用戶來指定 InputFormat ， OutputFormat ，和映射器。JobBase 本身就是 ImportJobBase 和 ExportJobBase 的子集，為特定的配置步驟提供更好的支持，分別由 ImportJobBase 和 ExportJobBase 的子類。 ， ImportJobBase.runImport （）將調用的配置步驟，并運行一個工作表導入 HD