sqoop用法之mysql與hive數據導入導出

本文目錄

一. Sqoop介紹
二. Mysql 數據導入到 Hive
三. Hive數據導入到Mysql
四. mysql數據增量導入hive

  • 1 基于遞增列Append導入
    • 1). 創建hive
    • 2). 創建job
    • 3) 執行job
  • 2 Lastmodified 導入實戰
    • 1). 新建一張表
    • 2). 初始化hive表:

一. Sqoop介紹

Sqoop是一個用來將Hadoop和關系型數據庫中的數據相互轉移的工具,可以將一個關系型數據庫(例如:MySQL、Oracle、Postgres等)中的數據導進到HadoopHDFS中,也可以將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). 將mysqlpeople_access_log表導入到hiveweb.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'
Java

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
Bash

該命令會啟用一個mapreduce任務,將mysql數據導入到hive表,并且指定了hive表的分隔符為\t,如果不指定則為默認分隔符^A(ctrl+A)。

參數說明


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

三. Hive數據導入到Mysql

還是使用上面的hiveweb.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
Bash

注意:mysqlpeople_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任務后,成功導入到mysqlpeople_access_log_out中.

四. mysql數據增量導入hive

實際中mysql數據會不斷增加,這時候需要用sqoop將數據增量導入hive,然后進行海量數據分析統計。增量數據導入分兩種,一是基于遞增列的增量數據導入(Append方式)。二是基于時間列的增量數據導入(LastModified方式)。有幾個核心參數:

  • –check-column:用來指定一些列,這些列在增量導入時用來檢查這些數據是否作為增量數據進行導入,和關系型數據庫中的自增字段及時間戳類似.注意:這些被指定的列的類型不能使任意字符類型,如char、varchar等類型都是不可以的,同時–check-column可以去指定多個列
  • –incremental:用來指定增量導入的模式,兩種模式分別為AppendLastmodified
  • –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條:


增量導入有幾個參數,保證下次同步的時候可以接著上次繼續同步.

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
Bash

這里通過sqoop job --create job_name命令創建了一個名為mysql2hive_jobsqoop 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');
SQL

這樣mysql里面就會多了2條數據。此時hive里面只有id1 ~ 6的數據,執行同步job使用以下命令。

sqoop job -exec mysql2hive_job

執行完成后,發現剛才mysql新加入的id7 ~ 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
SQL

由于實際場景中,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;
SQL

插入數據:

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');
SQL

mysql里面的數據就是這樣:


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
Bash

可以看到執行該命令后,啟動了二一個mapreduce任務,這樣6條數據就進入hiveweb.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)
SQL

3). 增量導入數據:

我們再次插入一條數據進入mysqlpeople_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" \
Bash

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
Bash

執行該命令后,與直接導入不同,該命令啟動了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)
SQL

可以看到id=6的數據,有2條,它的時間剛好是--last-value指定的時間,則會導入大于等于--last-value指定時間的數據,這點需要注意。










作者:柯廣的網絡日志

微信公眾號:Java大數據與數據倉庫