2. 程式人生 > 實用技巧 >sqoop用法之mysql與hive資料匯入匯出

sqoop用法之mysql與hive資料匯入匯出

阿新 • • 發佈：2020-12-22

目錄

一. 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指定時間的資料，這點需要注意。

相關推薦

sqoop用法之mysql與hive資料匯入匯出

目錄一. Sqoop介紹二. Mysql 資料匯入到 Hive三. Hive資料匯入到Mysql四. mysql資料增量匯入hive1. 基於遞增列Append匯入1). 建立hive表2). 建立job3). 執行job2. Lastmodified 匯入實戰1). 新建一張表2). 初始化hiv

sqoop將hive資料匯入mysql報錯

技術標籤：工作 sqoop export \\ --connect \'jdbc:mysql://192.168.1.135:3306/safe_manager?useUnicode=true&characterEncoding=utf-8\' \\

大資料-hive-記錄一次不用sqoop將mysql庫的資料匯入到hive庫中

技術標籤：大資料大資料hivemysql 一、場景描述 公司需要的一些資料是自己收集的，收集的時候儲存在mysql中，現在需要將mysql中的儲存的資料匯入到hive庫中。

3-1-5 MySQL基礎-事務機制與資料匯入匯出

事務機制 如果資料的寫入直接操作資料檔案是非常危險的事情,所以MySQL利用日誌來間接寫入.

Nebula Exchange 工具 Hive 資料匯入的踩坑之旅

摘要：本文由社群使用者 xrfinbj 貢獻，主要介紹 Exchange 工具從 Hive 數倉匯入資料到 Nebula Graph 的流程及相關的注意事項。

Hive(四)【DML 資料匯入匯出】

目錄一.資料匯入1.1 【load】--向資料中裝載資料案例1.2 【insert】--查詢語句向表中插入資料案例1.3 【as select】--查詢語句中建立表且載入資料案例1.4 【location】--建立表指定location載入資料案例1.5 【import

Docker進入mysql容器執行資料匯入

進入mysql https://www.cnblogs.com/todarcy/p/11283882.html ****@laravel-docker****:~/****/***-api/docker-compose/mysql$ docker ps

MySQL 文字檔案的匯入匯出資料的方法

有時為了更快速地插入大批量資料或交換資料，需要從文字中匯入資料或匯出資料到文字。下面的具體的方法大家可以參考下。多測試。

Canal+Kafka實現MySQL與Redis資料同步

思維導圖 前言 在很多業務情況下，我們都會在系統中加入redis快取做查詢優化。

SQL與Excel的資料匯入匯出

本篇內容來自https://jingyan.baidu.com/article/73c3ce28c839b7e50243d950.html   工具/原料   SQL Server 2005

SparkSQL之載入與儲存資料

技術標籤：Spark大資料SparkSQLspark大資料scala SparkSQL之載入與儲存資料 文章目錄 前言一、使用步驟1.載入資料2.儲存資料3.讀取mysql和寫入myslq4.Spark連線hive資料庫

binlog redis 驗證訂單_Canal 詳解 Mysql與Redis資料同步 解決方案

技術標籤：binlog redis 驗證訂單 要保證資料庫和redis強一致性是不可能的，肯定有少許時間的不一致。canal是阿里的一套元件，用來監聽mysql master傳送的類似binary log的資料，然後讓訊息費去消費。

Java之JSON與SpringMVC資料處理

SrpingMVC 資料處理 在JavaWeb中我們用的是request.getParam()來獲取前端傳遞的引數。

MySQL資料匯入匯出牛刀小試(r5筆記第3天)

最近學習了下MySQL中資料的匯入匯出，發現功能點真是豐富，很方便很快捷。

Linux下mysql資料庫的建立匯入匯出 及一些基本指令

首先linux 下檢視mysql相關目錄 檢視 mysql 的安裝路徑 執行查詢 SQL mysql>show variables like \'%dir%\';

Redis資料匯入匯出以及資料遷移的4種方法詳解

1、aof 匯入方式。 因為這種方式比較簡單，所以我就先介紹它。 分兩步來實現，第一步先讓源 Redis 生成 AOF 資料檔案。

淺談入門級oracle資料庫資料匯入匯出步驟

oracle資料庫資料匯入匯出步驟（入門） 說明： 1.資料庫資料匯入匯出方法有多種，可以通過exp/imp命令匯入匯出，也可以用第三方工具匯出，如：PLSQL

資料匯入匯出

1.從外部檔案系統向表中載入資料 load [overwrite] into load data [local] inpath \"\" [overwrite] into table table_name [partition(col_name=\"\")]

【技術筆記】neo4j 資料匯入匯出

技術標籤：技術筆記neo4j 最近頻繁使用neo4j，提純資料等操作。在此記錄一下常見命令之neo4j資料匯入。註明一下：本人系統採用Linux系統，版本是Ubuntu 16.04/18.04，neo4j版本：社群版 3.5.16/3.5.19。另外註明

Linux下mysql建立資料庫和匯入匯出.sql檔案

先登陸資料庫： mysql -u root -p; 輸入密碼。 建庫語句，建立資料庫test_db。 create database test_db default character set utf8mb4 default collate utf8mb4_general_ci;