一. Sqoop介紹
Sqoop
是一個用來將Hadoop
和關系型數據庫中的數據相互轉移的工具,可以將一個關系型數據庫(例如:MySQL、Oracle、Postgres
等)中的數據導進到Hadoop
的HDFS
中,也可以將HDFS
的數據導進到關系型數據庫中。對於某些NoSQL
數據庫它也提供了連接器。Sqoop
,類似於其他ETL
工具,使用元數據模型來判斷數據類型並在數據從數據源轉移到Hadoop
時確保類型安全的數據處理。Sqoop
專為大數據批量傳輸設計,能夠分割數據集並創建Hadoop
任務來處理每個區塊。
本文版本說明
hadoop
版本 :hadoop-2.7.2
hive版本
:hive-2.1.0
sqoop版本:sqoop-1.4.6
二. Mysql 數據導入到 Hive
1). 將mysql
的people_access_log
表導入到hive
表web.people_access_log
,並且hive
中的表不存在。
mysql
中表people_access_log
數據為:
1,15110101010,1577003281739,'112.168.1.2','https://www.baidu.com'
2,15110101011,1577003281749,'112.16.1.23','https://www.baidu.com'
3,15110101012,1577003281759,'193.168.1.2','https://www.taobao.com'
4,15110101013,1577003281769,'112.18.1.2','https://www.baidu.com'
5,15110101014,1577003281779,'112.168.10.2','https://www.baidu.com'
6,15110101015,1577003281789,'11.168.1.2','https://www.taobao.com'
將mysql
數據導入hive
的命令為:
sqoop import \
--connect jdbc:mysql://master1.hadoop:3306/test \
--username root \
--password 123456 \
--table people_access_log \
-m 1 \
--hive-import \
--create-hive-table \
--fields-terminated-by '\t' \
--hive-table web.people_access_log
該命令會啟用一個mapreduce
任務,將mysql
數據導入到hive
表,並且指定了hive
表的分隔符為\t
,如果不指定則為默認分隔符^A(ctrl+A)
。
參數說明
參數 | 說明 |
---|---|
--connect |
mysql 的連接信息 |
--username |
mysql 的用戶名 |
--password |
mysql 的密碼 |
--table |
被導入的mysql 源表名 |
-m |
並行導入啟用的map 任務數量,與--num-mapper 含義一樣 |
--hive-import |
插入數據到hive 當中,使用hive 默認的分隔符,可以使用--fields-terminated-by 參數來指定分隔符。 |
-- hive-table |
hive當中的表名 |
2). 也可以通過--query
條件查詢Mysql
數據,將查詢結果導入到Hive
sqoop import \
--connect jdbc:mysql://master1.hadoop:3306/test \
--username root \
--password 123456 \
--query 'select * from people_access_log where \$CONDITIONS and url = "https://www.baidu.com"' \
--target-dir /user/hive/warehouse/web/people_access_log \
--delete-target-dir \
--fields-terminated-by '\t' \
-m 1
參數 | 說明 |
---|---|
--query |
后接查詢語句,條件查詢需要\$CONDITIONS and 連接查詢條件,這里的\$ 表示轉義$ ,必須有. |
--delete-target-dir |
如果目標hive 表目錄存在,則刪除,相當於overwrite . |
三. Hive數據導入到Mysql
還是使用上面的hive
表web.people_access_log
,將其導入到mysql
中的people_access_log_out
表中.
sqoop export \
--connect jdbc:mysql://master1.hadoop:3306/test \
--username root \
--password 123456 \
--table people_access_log_out \
--input-fields-terminated-by '\t' \
--export-dir /user/hive/warehouse/web.db/people_access_log \
--num-mappers 1
注意:mysql
表people_access_log_out
需要提前建好,否則報錯:ErrorException: Table 'test.people_access_log_out' doesn't exist
。如果有id
自增列,hive
表也需要有,hive
表與mysql
表字段必須完全相同。
create table people_access_log_out like people_access_log;
執行完一個mr
任務后,成功導入到mysql
表people_access_log_out
中.
四. mysql數據增量導入hive
實際中mysql
數據會不斷增加,這時候需要用sqoop
將數據增量導入hive
,然后進行海量數據分析統計。增量數據導入分兩種,一是基於遞增列的增量數據導入(Append
方式)。二是基於時間列的增量數據導入(LastModified
方式)。有幾個核心參數:
–check-column
:用來指定一些列,這些列在增量導入時用來檢查這些數據是否作為增量數據進行導入,和關系型數據庫中的自增字段及時間戳類似.注意:這些被指定的列的類型不能使任意字符類型,如char、varchar等類型都是不可以的,同時–check-column
可以去指定多個列–incremental
:用來指定增量導入的模式,兩種模式分別為Append
和Lastmodified
–last-value
:指定上一次導入中檢查列指定字段最大值
1. 基於遞增列Append導入
接着前面的日志表,里面每行有一個唯一標識自增列ID
,在關系型數據庫中以主鍵形式存在。之前已經將id在0~6
之間的編號的訂單導入到Hadoop
中了(這里為HDFS
),現在一段時間后我們需要將近期產生的新的訂 單數據導入Hadoop
中(這里為HDFS
),以供后續數倉進行分析。此時我們只需要指定–incremental
參數為append
,–last-value
參數為6
即可。表示只從id
大於6
后即7
開始導入。
1). 創建hive
表
首先我們需要創建一張與mysql
結構相同的hive
表,假設指定字段分隔符為\t
,后面導入數據時候分隔符也需要保持一致。
2). 創建job
增量導入肯定是多次進行的,可能每隔一個小時、一天等,所以需要創建計划任務,然后定時執行即可。我們都知道hive
的數據是存在hdfs
上面的,我們創建sqoop job
的時候需要指定hive
的數據表對應的hdfs
目錄,然后定時執行這個job
即可。
當前mysql
中數據,hive
中數據與mysql
一樣也有6條:
id |
user_id |
access_time |
ip |
url |
---|---|---|---|---|
1 | 15110101010 | 1577003281739 | 112.168.1.2 | https://www.baidu.com |
2 | 15110101011 | 1577003281749 | 112.16.1.23 | https://www.baidu.com |
3 | 15110101012 | 1577003281759 | 193.168.1.2 | https://www.taobao.com |
4 | 15110101013 | 1577003281769 | 112.18.1.2 | https://www.baidu.com |
5 | 15110101014 | 1577003281779 | 112.168.10.2 | https://www.baidu.com |
6 | 15110101015 | 1577003281789 | 11.168.1.2 | https://www.taobao.com |
增量導入有幾個參數,保證下次同步的時候可以接着上次繼續同步.
sqoop job --create mysql2hive_job -- import \
--connect jdbc:mysql://master1.hadoop:3306/test \
--username root \
--password 123456 \
--table people_access_log \
--target-dir /user/hive/warehouse/web.db/people_access_log \
--check-column id \
--incremental append \
--fields-terminated-by '\t' \
--last-value 6 \
-m 1
這里通過sqoop job --create job_name
命令創建了一個名為mysql2hive_job
的sqoop job
。
3). 執行job
創建好了job
,后面只需要定時周期執行這個提前定義好的job
即可。我們先往mysql
里面插入2條數據。
INSERT INTO `people_access_log` (`id`,`user_id`,`access_time`,`ip`,`url`) VALUES
(7,15110101016,1577003281790,'112.168.1.3','https://www.qq.com'),
(8,15110101017,1577003281791,'112.1.1.3','https://www.microsoft.com');
這樣mysql
里面就會多了2條數據。此時hive
里面只有id
為1 ~ 6
的數據,執行同步job
使用以下命令。
sqoop job -exec mysql2hive_job
執行完成后,發現剛才mysql
新加入的id
為7 ~ 8
的兩條數據已經同步到hive
。
hive> select * from web.people_access_log;
OK
1 15110101010 1577003281739 112.168.1.2 https://www.baidu.com
2 15110101011 1577003281749 112.16.1.23 https://www.baidu.com
3 15110101012 1577003281759 193.168.1.2 https://www.taobao.com
4 15110101013 1577003281769 112.18.1.2 https://www.baidu.com
5 15110101014 1577003281779 112.168.10.2 https://www.baidu.com
6 15110101015 1577003281789 11.168.1.2 https://www.taobao.com
7 15110101016 1577003281790 112.168.1.3 https://www.qq.com
8 15110101017 1577003281791 112.1.1.3 https://www.microsoft.com
由於實際場景中,mysql
表中的數據,比如訂單表等,通常是一致有數據進入的,這時候只需要將sqoop job -exec mysql2hive_job
這個命令定時(比如說10分鍾頻率)執行一次,就能將數據10分鍾同步一次到hive
數據倉庫。
2. Lastmodified
導入實戰
append
適合業務系統庫,一般業務系統表會通過自增ID作為主鍵標識唯一性。Lastmodified
適合ETL
的數據根據時間戳字段導入,表示只導入比這個時間戳大,即比這個時間晚的數據。
1). 新建一張表
在mysql
中新建一張表people_access_log2
,並且初始化幾條數據:
CREATE TABLE `people_access_log2` (
`id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'id',
`user_id` bigint(20) unsigned NOT NULL COMMENT '用戶id',
`access_time` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
`ip` varchar(15) NOT NULL COMMENT '訪客ip',
`url` varchar(200) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
插入數據:
insert into people_access_log2(id,user_id, ip, url) values(1,15110101010,'112.168.1.200','https://www.baidu.com');
insert into people_access_log2(id,user_id, ip, url) values(2,15110101011,'112.16.1.2','https://www.baidu.com');
insert into people_access_log2(id,user_id, ip, url) values(3,15110101012,'112.168.1.2','https://www.taobao.com');
insert into people_access_log2(id,user_id, ip, url) values(4,15110101013,'112.168.10.2','https://www.baidu.com');
insert into people_access_log2(id,user_id, ip, url) values(5,15110101014,'112.168.1.2','https://www.jd.com');
insert into people_access_log2(id,user_id, ip, url) values(6,15110101015,'112.168.12.4','https://www.qq.com');
mysql
里面的數據就是這樣:
id | user_id | access_time | ip | url |
---|---|---|---|---|
1 |
15110101010 |
2019-12-28 16:23:10 |
112.168.1.200 |
https://www.baidu.com |
2 |
15110101011 |
2019-12-28 16:23:33 |
112.16.1.2 |
https://www.baidu.com |
3 |
15110101012 |
2019-12-28 16:23:41 |
112.168.1.2 |
https://www.taobao.com |
4 |
15110101013 |
2019-12-28 16:23:46 |
112.168.10.2 |
https://www.baidu.com |
5 |
15110101014 |
2019-12-28 16:23:52 |
112.168.1.2 |
https://www.jd.com |
6 |
15110101015 |
2019-12-28 16:23:56 |
112.168.12.4 |
https://www.qq. |
2). 初始化hive
表:
初始化hive
數據,將mysql
里面的6
條數據導入hive
中,並且可以自動幫助我們創建對應hive
表,何樂而不為,否則我們需要自己手動創建,完成初始化工作。
sqoop import \
--connect jdbc:mysql://master1.hadoop:3306/test \
--username root \
--password 123456 \
--table people_access_log2 \
--hive-import \
--create-hive-table \
--fields-terminated-by ',' \
--hive-table web.people_access_log2
可以看到執行該命令后,啟動了二一個mapreduce
任務,這樣6條數據就進入hive
表web.people_access_log2
了:
hive> select * from web.people_access_log2;
OK
1 15110101010 2019-12-28 16:23:10.0 112.168.1.200 https://www.baidu.com
2 15110101011 2019-12-28 16:23:33.0 112.16.1.2 https://www.baidu.com
3 15110101012 2019-12-28 16:23:41.0 112.168.1.2 https://www.taobao.com
4 15110101013 2019-12-28 16:23:46.0 112.168.10.2 https://www.baidu.com
5 15110101014 2019-12-28 16:23:52.0 112.168.1.2 https://www.jd.com
6 15110101015 2019-12-28 16:23:56.0 112.168.12.4 https://www.qq.com
Time taken: 0.326 seconds, Fetched: 6 row(s)
3). 增量導入數據:
我們再次插入一條數據進入mysql
的people_access_log2
表:
insert into people_access_log2(id,user_id, ip, url) values(7,15110101016,'112.168.12.45','https://www.qq.com');
此時,mysql
表里面已經有7
條數據了,我們使用incremental
的方式進行增量的導入到hive
:
sqoop import \
--connect jdbc:mysql://master1.hadoop:3306/test \
--username root \
--password 123456 \
--table people_access_log2 \
--hive-import \
--hive-table people_access_log2 \
-m 1 \
--check-column access_time \
--incremental lastmodified \
--last-value "2019-12-28 16:23:56" \
2019-12-28 16:23:56
就是第6條數據的時間,這里需要指定。報錯了:
19/12/28 16:17:25 ERROR tool.ImportTool: Error during import: --merge-key or --append is required when using --incremental lastmodified and the output directory exists.
注意:可以看到--merge-key or --append is required when using --incremental lastmodified
意思是,這種基於時間導入模式,需要指定--merge-key
或者--append
參數,表示根據時間戳導入,數據是直接在末尾追加(append)還是合並(merge),這里使用merge
方式,根據id
合並:
sqoop import \
--connect jdbc:mysql://master1.hadoop:3306/test \
--username root \
--password 123456 \
--table people_access_log2 \
--hive-import \
--hive-table web.people_access_log2 \
--check-column access_time \
--incremental lastmodified \
--last-value "2019-12-28 16:23:56" \
--fields-terminated-by ',' \
--merge-key id
執行該命令后,與直接導入不同,該命令啟動了2個mapreduce
任務,這樣就把數據增量merge
導入hive
表了.
hive> select * from web.people_access_log2 order by id;
OK
1 15110101010 2019-12-28 16:23:10.0 112.168.1.200 https://www.baidu.com
2 15110101011 2019-12-28 16:23:33.0 112.16.1.2 https://www.baidu.com
3 15110101012 2019-12-28 16:23:41.0 112.168.1.2 https://www.taobao.com
4 15110101013 2019-12-28 16:23:46.0 112.168.10.2 https://www.baidu.com
5 15110101014 2019-12-28 16:23:52.0 112.168.1.2 https://www.jd.com
6 15110101015 2019-12-28 16:23:56.0 112.168.12.4 https://www.qq.com
6 15110101015 2019-12-28 16:23:56.0 112.168.12.4 https://www.qq.com
7 15110101016 2019-12-28 16:28:24.0 112.168.12.45 https://www.qq.com
Time taken: 0.241 seconds, Fetched: 8 row(s)
可以看到id=6
的數據,有2條,它的時間剛好是--last-value
指定的時間,則會導入大於等於--last-value
指定時間的數據,這點需要注意。
Hive系列文章
Hive表的基本操作
Hive中的集合數據類型
Hive動態分區詳解
hive中orc格式表的數據導入
Java通過jdbc連接hive
通過HiveServer2訪問Hive
SpringBoot連接Hive實現自助取數
hive關聯hbase表
Hive udf 使用方法
Hive基於UDF進行文本分詞
Hive窗口函數row number的用法
數據倉庫之拉鏈表