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

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

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

一、Sqoop工作原理

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

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

Sqoop安裝：

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

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

二、Sqoop命令格式

基礎(chǔ)使用語(yǔ)法:

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

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

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

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

場(chǎng)景：

Step1: 查看業(yè)務(wù)數(shù)據(jù)庫(kù)中 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默認(rèn)使用\001作為表字段的分隔符，但也可以在創(chuàng)建表時(shí)指定特殊的分隔符。

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

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

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

原因:

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

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ù)結(jié)構(gòu)，ODS層數(shù)據(jù)盡量抱持原結(jié)構(gòu)

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

常見的文件格式介紹:

Avro格式特點(diǎn)

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

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 \    # 選擇文件存儲(chǔ)格式為AVRO
  --fields-terminated-by "\001" \
  -m 1

Hive建表指定文件的存儲(chǔ)格式:

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

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

四、Sqoop增量采集方案

Sqoop 支持兩種增量模式：

• append 模式：

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

• lastmodified 模式：

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

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

Append模式:

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

特點(diǎn)：只能導(dǎo)入增加的數(shù)據(jù)，無(wú)法導(dǎo)入更新的數(shù)據(jù)

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

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

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

此處的last_value需要手動(dòng)填寫，因此可以使用Sqoop的job管理進(jìn)行自動(dòng)記錄。

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

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

特點(diǎn)：既導(dǎo)入新增的數(shù)據(jù)也導(dǎo)入更新的數(shù)據(jù)

場(chǎng)景：表中的記錄會(huì)新增或更新，且每次更新都會(huì)修改 lastmode 時(shí)間戳。一般無(wú)法滿足要求，所以不用。

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

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

若同一條記錄被多次更新，且 lastmode 時(shí)間超過(guò) --last-value，Sqoop 會(huì)多次導(dǎo)入該記錄。

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

 --merge-key id  # 假設(shè) id 是主鍵列

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

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

場(chǎng)景：一般用于自定義增量采集每天的分區(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ù)設(shè)置的sql進(jìn)行數(shù)據(jù)導(dǎo)入，因此是最常用的場(chǎng)景。

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

最新評(píng)論