Sqoop詳解

通常一個組織中有價值的資料都要存儲在關系型資料庫系統中。但是為了進一步進行處理，有些資料需要抽取出來，通過MapReduce程式進行再次加工。為了能夠和HDFS系統之外的資料庫系統機型互動，MapReduce程式需要使用外部API來通路資料。Sqoop就是一個開源的工具，它允許使用者将資料從關系型資料庫抽取到hadoop中；也可以把MapReduce處理完的資料導回到資料庫中。

1、sqoop使用

在學習sqoop使用之前，我們需要檢視sqoop都是可以完成什麼任務，通過鍵入：sqoop help，我們就可以看到sqoop可以提供的服務。在項目中，我們主要使用的是sqoop import服務，在使用的過程中，我們還會經曆很多定制修改，講逐一講解。

1.1将資料從資料庫導入到hadoop中

導入指令：sqoop import –connect jdbc:mysql://hostname:port/database –username root –password 123456 –table example –m 1。在這裡講解一下指令的構成，如下：

1、--connect jdbc:mysql://hostname:port/database指定mysql資料庫主機名和端口号和資料庫名；

2、--username root 指定資料庫使用者名

3、--password 123456 指定資料庫密碼

4、--table example mysql中即将導出的表

5、-m 1  指定啟動一個map程序，如果表很大，可以啟動多個map程序

6、導入到HDFS中的路徑  預設：/user/grid/example/part-m-00000

注意：預設情況下，Sqoop會将我們導入的資料儲存為逗号分隔的文本檔案。如果導入資料的字段内容存在分隔符，則我們可以另外指定分隔符、字段包圍字元和轉義字元。使用指令行參數可以指定分隔符、檔案格式、壓縮以及對導入過程進行更細粒度的控制。

1.2、生成代碼

         除了能夠将資料庫表的内容寫到HDFS,Sqoop還生成了一個Java源檔案（example.java）儲存在目前的本地目錄中。在運作了前面的sqoop import指令之後，可以通過ls example.java指令看到這個檔案。代碼生成時Sqoop導入過程的必要組成部分，他是在Sqoop将源資料庫的表資料寫到HDFS之前，首先用生成的代碼對其進行反序列化。

         生成的類中能夠儲存一條從被導入表中取出的記錄。該類可以在MapReduce中使用這條記錄，也可以将這條記錄儲存在HDFS中的一個SequenceFile檔案中。在導入過程中，由Sqoop生成的SequenceFile檔案會生成的類，将每一個被導入的行儲存在其鍵值對格式中“值”的位置。

         也許你不想将生成的類命名為example，因為每一個類的執行個體隻對應與一條記錄。我們可以使用另外一個Sqoop工具來生成源代碼，并不執行導入操作，這個生成的代碼仍然會檢查資料庫表，以确定與每個字段相比對的資料類型：

         Sqoop codegen –connect jdbc:mysql://localhost/yidong –table example –class-name example

         Codegen工具隻是簡單的生成代碼，他不執行完整的導入操作。我們指定希望生成一個名為example的類，這個類将被寫入到example.java檔案中。在之前執行的導入過程中，我們還可以指定—class-name和其他代碼生成參數。如果你意外的删除了生成的源代碼，或希望使用不同于導入過程的設定來生成代碼，都可以用這個工具來重新生成代碼。

         如果計劃使用導入到SequenceFile檔案中的記錄，你将不可避免的用到生成的類（對SequenceFile檔案中的資料進行反序列化）。在使用文本檔案中的記錄時，不需要用生成的代碼。

1.3、深入了解資料庫導入

         在深入了解之前，我們需要先想一個問題：Sqoop是通過一個MapReduce作業從資料庫中導入一個表，這個作業從表中抽取一行行記錄，然後寫入到HDFS。MapReduce是如何記錄的？

         下圖是Sqoop從資料庫中導入到HDFS的原理圖：

在導入開始之前，Sqoop使用JDBC來檢查将要導入的表。他檢索出表中所有的列以及列的SQL資料類型。這些SQL類型（VARCHAR、INTEGER）被映射到Java資料類型（String、Integer等）,在MapReduce應用中将使用這些對應的java類型來儲存字段的值。Sqoop的代碼生成器使用這些資訊來建立對應表的類，用于儲存從表中抽取的記錄。例如前面提到過的example類。

對于導入來說，更關鍵的是DBWritable接口的序列化方法，這些方法能使Widget類和JDBC進行互動：

         Public void readFields(resultSet _dbResults)throws SQLException;

         Public void write(PreparedStatement _dbstmt)throws SQLException;

JDBC的ResultSet接口提供了一個使用者從檢查結果中檢索記錄的遊标；這裡的readFields()方法将用ResultSet中一行資料的列來填充Example對象的字段。

Sqoop啟動的MapReduce作業用到一個InputFormat,他可以通過JDBC從一個資料庫表中讀取部分内容。Hadoop提供的DataDriverDBInputFormat能夠為幾個Map任務對查詢結果進行劃分。為了擷取更好的導入性能，查詢會根據一個“劃分列”來進行劃分的。Sqoop會選擇一個合适的列作為劃分列（通常是表的主鍵）。

在生成反序列化代碼和配置InputFormat之後，Sqoop将作業發送到MapReduce叢集。Map任務将執行查詢并将ResultSet中的資料反序列化到生成類的執行個體，這些資料要麼直接儲存在SequenceFile檔案中，要麼在寫到HDFS之前被轉換成分割的文本。

Sqoop不需要每次都導入整張表，使用者也可以在查詢中加入到where子句，以此來限定需要導入的記錄：Sqoop –query <SQL>。

         導入和一緻性：在向HDFS導入資料時，重要的是要確定通路的是資料源的一緻性快照。從一個資料庫中并行讀取資料的MAP任務分别運作在不同的程序中。是以，他們不能共享一個資料庫任務。保證一緻性的最好方法就是在導入時不允許運作任何進行對表中現有資料進行更新。

1.4、使用導入的資料

一旦資料導入HDFS，就可以供定制的MapReduce程式使用。導入的文本格式資料可以供Hadoop Streaming中的腳本或者TextInputFormat為預設格式運作的MapReduce作業使用。

為了使用導入記錄的個别字段，必須對字段分割符進行解析，抽取出的字段值并轉換為相應的資料類型。Sqoop生成的表類能自動完成這個過程，使你可以将精力集中在真正的要運作的MapReduce作業上。

1.5、導入的資料與hive

Hive和sqoop共同構成一個強大的服務于分析任務的工具鍊。Sqoop能夠根據一個關系資料源中的表來生成一個hive表。既然我們已經将表的資料導入到HDFS中，那麼就可以直接生成相應hive表的定義，然後加載儲存在HDFS中的資料，例如：

         Sqoop create-hive-table –connect jdbc:mysql://localhoust/yidong –table example –fields-terminated-by “,”

         Load data inpath ‘example’ into table example

         注：在為一個特定的已導入資料集建立相應的hive表定義時，我們需要指定該資料集所使用的分隔符。否則，sqoop将允許hive使用自己預設的風格符。

         如果想直接從資料庫将資料導入到hive，可以将上述三個步驟（将資料導入HDFS，建立hive表，将hdfs中的資料導入hive）縮短為一個步驟。在進行導入時，sqoop可以生成hive表的定義，然後直接将資料導入hive表：

         Sqoop import –connect jdbc:mysql://localhost/hadoopguide –table widgets –m 1 –hive-import

1.6、導入大對象

很多資料庫都具有在一個字段中儲存大量資料的能力。取決于資料是文本還是二進制類型，通常這些類型為CLOB或BLOB。資料庫一般會對這些“大對象”進行特殊處理。Sqoop将導入的大對象資料存儲在LobFile格式的單獨檔案中，lobfile格式能夠存儲非常大的單條記錄。Lobfile檔案中的每條記錄儲存一個大對象。

在導入一條記錄時，所有的“正常”字段會在一個文本檔案中一起物化，同時還生成一個指向儲存CLOB或BLOB列的lobfile檔案的引用。

2、執行導出

在sqoop中，導出是将hdfs作為一個資料源，而将一個遠端的資料庫作為目标。将一張表從hdfs導出到資料庫時，我們必須在資料庫中建立一張用于接收資料的目标表。雖然sqoop可以推斷出那個java類型适合存儲sql資料類型，但反過來确實行不通。是以，必須由使用者來确定哪些類型是最合适的。

例如：我們打算從hive中導出zip_profits表到mysql資料庫中。

     ①先在mysql中建立一個具有相同序列順序及合适sql表型的目标表：

   Create table sales_by_sip(volume decimal(8,2),zip integer);

②接着運作導出指令：

  Sqoop export –connect jdbc:mysql://localhost/hadoopguide –m 1 –table sales_by_zip –export-dir /user/hive/warehouse/zip_profits –input-fields-terminated-by “\0001”

     ③過mysql來确認導出成功：

        mysql hadoopguide –e ‘select * from sales_by_zip’

     注意：在hive中建立zip_profits表時，我們沒有指定任何分隔符。是以hive使用了自己的預設分隔符；但是直接從檔案中讀取這張表時，我們需要将所使用的分隔符告知sqoop。Sqoop預設記錄是以換行符作為分隔符。是以，可在sqoop export指令中使用—input-fields-terminated-by參數來指定字段分隔符。

2.1、深入了解導出

Sqoop導出功能的架構與其導入功能非常相似，在執行導出操作之前，sqoop會根據資料庫連接配接字元串來選擇一個導出方法。一般為jdbc。然後，sqoop會根據目标表的定義生成一個java類。這個生成的類能夠從文本檔案中解析記錄，并能夠向表中插入類型合适的值。接着會啟動一個MapReduce作業，從HDFS中讀取源資料檔案，使用生成的類解析記錄，并且執行標明的導出方法。

基于jdbc的導出方法會産生一批insert語句，每條語句都會向目标表中插入多條記錄。多個單獨的線程被用于從HDFS讀取資料并與資料庫進行通信，以確定涉及不同系統的I/O操作能夠盡可能重疊執行。

雖然HDFS讀取資料的MapReduce作業大多根據所處理檔案的數量和大小來選擇并行度（map任務的數量），但sqoop的導出工具允許使用者明确設定任務的數量。由于導出性能會受并行的資料庫寫入線程數量的影響，是以sqoop使用combinefileinput類将輸入檔案分組配置設定給少數幾個map任務去執行。

2.2、導出與事務

程序的并行特性，導緻導出操作往往不是原子操作。Sqoop會采用多個并行的任務導出，并且資料庫系統使用固定大小的緩沖區來存儲事務資料，這時一個任務中的所有操作不可能在一個事務中完成。是以，在導出操作進行過程中，送出過的中間結果都是可見的。在導出過程完成前，不要啟動那些使用導出結果的應用程式，否則這些應用會看到不完整的導出結果。

更有問題的是，如果任務失敗，他會從頭開始重新導入自己負責的那部分資料，是以可能會插入重複的記錄。目前sqoop還不能避免這種可能性。在啟動導出作業前，應當在資料庫中設定表的限制（例如，定義一個主鍵列）以保證資料行的唯一性。

2.3、導出與SequenceFile

         Sqoop還可以将存儲在SequenceFile中的記錄導出到輸出表，不過有一些限制。SequenceFile中可以儲存任意類型的記錄。Sqoop的導出工具從SequenceFile中讀取對象，然後直接發送到OutputCollector，由他将這些對象傳遞給資料庫導出OutputFormat。為了能讓Sqoop使用，記錄必須被儲存在SequenceFile鍵值對格式的值部分，并且必須繼承抽象類com.cloudera.sqoop.lib.SqoopRecord。

3、項目案例

在我們移動項目中，有些資料是通過web頁面維護，這些資料都是通過管理者手動添加到分析系統中的。為了使hive可以更好的進行分析，是以需要将這些服務資料，定期導入到我們的hive資料倉儲中，這時sqoop就需要發揮作用了。

需要通過sqoop将資料從mysql導入到hive資料倉儲的服務有：應用管理、管道管理、自定義事件管理、裡程碑管理。他們對應的資料庫（10.6.219.86）表分别是：base_app、base_channel、base_event、base_milestone。

3.1、采用sqoop指令導入

在本次實踐中，我們僅以導入應用管理表（base_app）為例進行說明。

1、執行sqoop指令将資料從mysql導入到hive中，指令為：

Sqoop import --connect jdbc:mysql://10.1.11.78:3306/video --table base_event --username root --password 123456 -m 1 --hive-import --hive-database video --hive-table base_event --hive-overwrite --fields-terminated-by "\t"--lines-terminated-by “\n”--as-textfile

         指令詳解：

• sqoop import ---執行sqoop導入指令；
• --connect jdbc:mysql://hostname:port/database ---要連接配接的資料庫位址、端口号、資料庫database；
• --table base_app  -----  要操作的資料庫表；
• --username root  -----  連接配接資料庫的使用者名；
• --password 123456  ---  連接配接資料庫的密碼；
• -m 1 ------  要啟動的map數量
• --hive-import  --- 采用hive方式導入
• [--create-hive-table] --- 如果導入的表在hive中不存在的話，sqoop自動在hive中建立該表。但是當表存在的情況下，添加該選項會導緻指令報錯。是以在實際操作中，不建議使用，并且在實際的操作過程中，即使我們不添加該項輔助指令，sqoop也會在hive中建立導入的表。
• --hive-database yidong --- 要将資料庫表導入到hive的那個database中；
• --hive-table base_app --- 要将資料庫表導入到hive的那個表中；
• --hive-overwrite ---如果hive的表中已經存在資料，添加該項操作後，會将原有的資料覆寫掉。
• --fields-terminated-by “\t” --- hive存儲到hdfs中的檔案中字段間的分隔符；
• --lines-terminated-by “\n”– hive存儲到hdfs中的檔案中每行間的分隔符；
• --as-textfile ---hive存儲到hdfs中的檔案格式，采用文本存儲；

常用輔助指令詳解：

1、通過help指令檢視sqoop導入幫助：

Sqoop help import;

2、Sqoop導入行輔助操作詳解：

3、Sqoop hive導入輔助操作詳解：

3.2、通過oozie編寫workflow定時将同步資料

         在本次項目實踐中，我們需要用到oozie的sqoop action。在這章節中，我會示範給大家如何編寫workflow.xml實作定時資料同步的。

         在項目運作的過程中發現，為了能更好的進行資料同步，建議首先把需要同步的資料庫表在環境初始化的過程中，在hive資料倉儲中建立出來。這樣項目運作的過程中，會很輕松的執行。

         編寫workflow.xml進行同步工作，workflow.xml内容如下：

         在該例子中，我們是對四個需要同步的表進行，通過sqoop進行了同步。大家在編寫oozie sqoop action腳本的工程中，需要特别注意：sqoop支援兩種方式配置指令，一種是command,另一種是arg的方式。不管是采用哪種方式，導入導出的過程的指令中，一定不要出現sqoop，而是從sqoop需要執行的指令以後的内容開始（從import或者export開始），請檢視截圖中的紅色标注。

3.3、項目實踐總結歸納

① 首先在初始化同步工作流環境的過程中，先在script.hql檔案中編寫建表語句，在hive資料倉儲中建立出需要進行同步的table。

② 編寫workflow.xml檔案，定義需要同步表的sqoop工作流，且工作流需要執行的指令中，不會出現“sqoop”關鍵字，而是從該關鍵字以後的指令開始，如：在cli情況下我們需要同步資料庫，執行的指令為：sqoop import jdbc:mysql://……;而在workflow.xml配置的command指令中為：import jdbc:mysql://…..，在這裡不用再填寫sqoop，否則會報錯，這是初學人員常遇到的錯誤，謹記！！！！

③ 在sqoop導入mysql資料到hive時，--fields-terminated-by "\t"，自定義的分隔符要為雙引号。否則指定的分隔符無效！