Sqoop數(shù)據(jù)采集格式問題

Apache Sqoop是一款開源的工具，主要用于在Hadoop(Hive)與傳統(tǒng)的數(shù)據(jù)庫(mysql、postgresql…)間進行數(shù)據(jù)的傳遞，可以將一個關系型數(shù)據(jù)庫（例如 ： MySQL ,Oracle ,Postgres等）中的數(shù)據(jù)導進到Hadoop的HDFS中，也可以將HDFS的數(shù)據(jù)導進到關系型數(shù)據(jù)庫中。

Sqoop項目開始于2009年，最早是作為Hadoop的一個第三方模塊存在，后來為了讓使用者能夠快速部署，也為了讓開發(fā)人員能夠更快速的迭代開發(fā)，Sqoop獨立成為一個Apache項目。

一、Sqoop工作原理

• 數(shù)據(jù)導入：Sqoop通過MapReduce任務來實現(xiàn)數(shù)據(jù)的并行導入。首先，它會將關系型數(shù)據(jù)庫中的數(shù)據(jù)表按照一定的規(guī)則進行分區(qū)，然后為每個分區(qū)啟動一個Map任務，同時從數(shù)據(jù)庫中讀取相應分區(qū)的數(shù)據(jù)，并將數(shù)據(jù)寫入到HDFS或其他Hadoop存儲系統(tǒng)中。這樣可以充分利用Hadoop集群的分布式計算能力，提高數(shù)據(jù)導入的效率。
• 導出過程：與導入類似，Sqoop也會將數(shù)據(jù)進行分區(qū)處理，然后通過Map任務將Hadoop中的數(shù)據(jù)讀取出來，并按照目標關系型數(shù)據(jù)庫的格式要求，將數(shù)據(jù)寫入到數(shù)據(jù)庫中。

Sqoop通過創(chuàng)建一個數(shù)據(jù)傳輸?shù)腗R程序，進而實現(xiàn)數(shù)據(jù)傳輸。

Sqoop安裝：

• JAVA環(huán)境配置
• Hadoop環(huán)境配置
• 相關數(shù)據(jù)庫驅(qū)動包

只要環(huán)境滿足以上設置，直接解壓Sqoop安裝包即可安裝，修改配置后即可使用。

二、Sqoop命令格式

基礎使用語法:

sqoop import | export \
--數(shù)據(jù)庫連接參數(shù)
--HDFS或者Hive的連接參數(shù)
--配置參數(shù)

數(shù)據(jù)傳輸常用參數(shù):

Sqoop命令的使用方法可以通過sqoop -h命令查看相關使用方法，此處不在贅述了

三、Oracle數(shù)據(jù)采集格式問題

場景：

Step1: 查看業(yè)務數(shù)據(jù)庫中 CISS_SERVICE_WORKORDER 表的數(shù)據(jù)條數(shù)。

select count(1) as cnt from CISS_SERVICE_WORKORDER;  178609條

Step2: 采集CISS_SERVICE_WORKORDER的數(shù)據(jù)到HDFS上

sqoop import \
--connect jdbc:oracle:thin:@oracle.bigdata.cn:1521:helowin \  
--username ciss \
--password 123456 \
--table CISS4.CISS_SERVICE_WORKORDER \
--delete-target-dir \
--target-dir /test/full_imp/ciss4.ciss_service_workorder \
--fields-terminated-by "\001" \   #指定字段分割符
-m 1  #指定并行度

Hive默認使用\001作為表字段的分隔符，但也可以在創(chuàng)建表時指定特殊的分隔符。

Step3: 使用Hive查看導入數(shù)據(jù)表的行數(shù)

create external table test_text(
line string # 將導入的數(shù)據(jù)一行作為表中的一列
)
location '/test/full_imp/ciss4.ciss_service_workorder';
select count(*) from test_text;  195825條

問題:
Sqoop采集完數(shù)據(jù)后，HDFS數(shù)據(jù)中存儲的數(shù)據(jù)行數(shù)跟源數(shù)據(jù)庫的數(shù)據(jù)量不符合。

原因:

• Sqoop以文本格式導入數(shù)據(jù)時，HDFS的默認文件格式為textfile，默認的換行符是特殊字符\n。
• Oracle中的數(shù)據(jù)列中如果出現(xiàn)了\n、\r、\t等特殊字符，就會被劃分為多行

Oracle數(shù)據(jù):

Sqoop轉(zhuǎn)換后的數(shù)據(jù):

Hive表中的數(shù)據(jù):

解決方法:

• 方案一:
  • 刪除或者替換數(shù)據(jù)中的換行符
  • Sqoop參數(shù) --hive-drop-import-delims 刪除換行符
  • Sqoop參數(shù) --hive-delims-replacement char 替換換行符

不建議使用，破壞原始數(shù)據(jù)結構，ODS層數(shù)據(jù)盡量抱持原結構

• 方案二:
  • 采用特殊的存儲格式，AVRO格式

常見的文件格式介紹:

Avro格式特點

• 優(yōu)點
  • 二進制數(shù)據(jù)存儲，性能好、效率高
  • 使用JSON描述模式，支持場景更豐富
  • Schema和數(shù)據(jù)統(tǒng)一存儲，消息自描述(將表中的一行數(shù)據(jù)作為對象存儲，并且Schema為元數(shù)據(jù))
  • 模式定義允許定義數(shù)據(jù)的排序
• 缺點
  • 只支持Avro自己的序列化格式
  • 少量列的讀取性能比較差，壓縮比較低
• 場景：基于行的大規(guī)模結構化數(shù)據(jù)寫入、列的讀取非常多或者Schema變更操作比較頻繁的場景

Sqoop使用Avro格式:

  sqoop import \
  -Dmapreduce.job.user.classpath.first=true \
  --connect jdbc:oracle:thin:@oracle.bigdata.cn:1521:helowin \
  --username ciss \
  --password 123456 \
  --table CISS4.CISS_SERVICE_WORKORDER \
  --delete-target-dir \
  --target-dir /test/full_imp/ciss4.ciss_service_workorder \
  --as-avrodatafile \    # 選擇文件存儲格式為AVRO
  --fields-terminated-by "\001" \
  -m 1

Hive建表指定文件的存儲格式:

create external table test_avro(
line string
)
stored as avro
location '/test/full_imp/ciss4.ciss_service_workorder';

AVRO 數(shù)據(jù)以 二進制序列化 存儲，字段通過預定義的 模式（Schema） 解析，而非依賴分隔符，即使字段內(nèi)容包含逗號、換行符等特殊字符，也不會影響數(shù)據(jù)結構的正確性。
Schema 定義（JSON 格式），明確描述了字段名稱、類型、順序等信息。

四、Sqoop增量采集方案

Sqoop 支持兩種增量模式：

• append 模式：

適用于 僅追加數(shù)據(jù) 的表（如日志表），基于 遞增列（如自增主鍵 id）采集新數(shù)據(jù)。

• lastmodified 模式：

適用于 數(shù)據(jù)會更新 的表（如用戶表），基于 時間戳列（如 last_update_time）采集新增或修改的數(shù)據(jù)。

append模式要求源數(shù)據(jù)表具備自增列，如建表時設置的自增id
lastmodified模式要求源數(shù)據(jù)表具有時間戳字段。

Append模式:

要求：必須有一列自增的值，按照自增的int值進行判斷

特點：只能導入增加的數(shù)據(jù)，無法導入更新的數(shù)據(jù)

場景：數(shù)據(jù)只會發(fā)生新增，不會發(fā)生更新的場景

sqoop import \                                   # 執(zhí)行數(shù)據(jù)導入操作
  --connect jdbc:mysql://node3:3306/sqoopTest \  # 連接MySQL數(shù)據(jù)庫（地址：node3，數(shù)據(jù)庫名：sqoopTest）
  --username root \                             # 數(shù)據(jù)庫用戶名：root
  --password 123456 \                           # 數(shù)據(jù)庫密碼：123456
  --table tb_tohdfs \                           # 要導入的源表：tb_tohdfs
  --target-dir /sqoop/import/test02 \           # HDFS目標目錄（數(shù)據(jù)將寫入此路徑）
  --fields-terminated-by '\t' \                 # 字段分隔符為制表符（\t）
  --check-column id \                           # 指定增量檢查列：id（通常是自增主鍵）
  --incremental append \                        # 增量模式為“append”（僅導入新數(shù)據(jù)）
  --last-value 0 \                              # 上次導入的id最大值（初始值為0，首次導入id>0的數(shù)據(jù)）
  -m 1                                          # 使用1個Map任務（單線程）

appebd模式使用last-value記錄上次導入的數(shù)據(jù)id最大值，初次導入一般為全量導入，即id>0

此處的last_value需要手動填寫，因此可以使用Sqoop的job管理進行自動記錄。

sqoop job --create my_job -- import ... --incremental append --check-column id --last-value 0
sqoop job --exec my_job  # 自動更新 last-value

lastmodified模式：
要求：必須包含動態(tài)時間變化這一列，按照數(shù)據(jù)變化的時間進行判斷

特點：既導入新增的數(shù)據(jù)也導入更新的數(shù)據(jù)

場景：表中的記錄會新增或更新，且每次更新都會修改 lastmode 時間戳。一般無法滿足要求，所以不用。

sqoop import \                                   # 執(zhí)行數(shù)據(jù)導入操作
  --connect jdbc:mysql://node3:3306/sqoopTest \  # 連接MySQL數(shù)據(jù)庫（地址：node3，數(shù)據(jù)庫名：sqoopTest）
  --username root \                             # 數(shù)據(jù)庫用戶名：root
  --password 123456 \                           # 數(shù)據(jù)庫密碼：123456
  --table tb_lastmode \                         # 要導入的源表：tb_lastmode
  --target-dir /sqoop/import/test03 \           # HDFS目標目錄（數(shù)據(jù)將寫入此路徑）
  --fields-terminated-by '\t' \                 # 字段分隔符為制表符（\t）
  --incremental lastmodified \                  # 增量模式為“l(fā)astmodified”（采集新增或修改的數(shù)據(jù)）
  --check-column lastmode \                     # 指定時間戳列：lastmode（記錄數(shù)據(jù)的更新時間）
  --last-value '2021-06-06 16:09:32' \          # 上次導入的最大時間值（導入此時間之后的新增/修改數(shù)據(jù)）
  -m 1                                          # 使用1個Map任務（單線程）

lastmodified模式使用時間戳記載數(shù)據(jù)的更新線。

若同一條記錄被多次更新，且 lastmode 時間超過 --last-value，Sqoop 會多次導入該記錄。

解決方案：添加 --merge-key <主鍵列> 參數(shù)，合并新舊數(shù)據(jù)（基于主鍵去重）：

 --merge-key id  # 假設 id 是主鍵列

自定義模式:
要求：每次運行的輸出目錄不能相同

特點：自己實現(xiàn)增量的數(shù)據(jù)過濾，可以實現(xiàn)新增和更新數(shù)據(jù)的采集

場景：一般用于自定義增量采集每天的分區(qū)數(shù)據(jù)到Hive

sqoop  import \
--connect jdbc:mysql://node3:3306/db_order \
--username root \
--password-file file:///export/data/sqoop.passwd \
--query "select * from tb_order where substring(create_time,1,10) = '2021-09-14' or substring(update_time,1,10) = '2021-09-14' and \$CONDITIONS " \
--delete-target-dir \
--target-dir /nginx/logs/tb_order/daystr=2021-09-14 \
--fields-terminated-by '\t' \
-m 1

自定義模式可以根據(jù)設置的sql進行數(shù)據(jù)導入，因此是最常用的場景。

到此這篇關于Apache Sqoop數(shù)據(jù)采集問題的文章就介紹到這了,更多相關Apache Sqoop數(shù)據(jù)采集內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

最新評論