用Sqoop進行Hive和MySQL之間的資料互導
阿新 • • 發佈：2021-07-10
Hive導資料入MySQL
建立mysql表
use anticheat;
create table anticheat_blacklist(
userid varchar(30) primary key ,
dt int,
update_time timestamp,
delete_flag int,
operator varchar(30)
);
全量匯出
用sqoop export全量匯出hive表資料入mysql,具體命令如下：
sqoop export -D mapred.job.queue.name=datacenter --connect jdbc:mysql://localhost:3306/anticheat?tinyInt1isBit=false --username root --password ^qn9DFYPm --table anticheat_blacklist --input-fields-terminated-by '\t' --input-null-string '\\N' --input-null-non-string '\\N' --num-mappers 10 --export-dir hdfs://dc5/user/test/hive/online/anticheat_blacklist_mysql
增量匯出
sqoop export -D mapred.job.queue.name=datacenter --connect jdbc:mysql://localhost:3306/anticheat?tinyInt1isBit=false --username root --password ^qn9DFYPm --table anticheat_blacklist2 --input-fields-terminated-by '\t' --input-null-string '\\N' --input-null-non-string '\\N' --num-mappers 10 --update-key update_time --update-mode allowinsert --export-dir hdfs://dc5/user/test/hive/online/anticheat_blacklist_mysql2
MySQL導資料入Hive
建立Hive表
建立同步mysql表的hive表
CREATE TABLE test.anticheat_blacklist_mysql(
key string,
dt int,
update_time timestamp,
delete_flag int,
operator string
)
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
STORED AS TEXTFILE
LOCATION 'hdfs://dc5/user/test/hive/online/anticheat_blacklist_mysql';
全量匯入
用sqoop import全量匯出mysql表資料入hive表,具體命令如下：
sqoop import -D mapred.job.queue.name=datacenter
--connect jdbc:mysql://localhost:3306/anticheat?tinyInt1isBit=false
--username root
--password ^qn9DFYPm
--table anticheat_blacklist
--delete-target-dir
--beeline "jdbc:hive2://dsrv2.heracles.sohuno.com:10000/test;principal=hive/[email protected];"
--hive-import --fields-terminated-by '\t'
--hive-database test
--hive-table anticheat_blacklist_mysql
--null-string '\\N'
--null-non-string '\\N'
--hive-overwrite
--num-mappers 1
--outdir /home/test/data/anticheat/mysql2hive
null字串轉為NULL,新增下面兩條引數可以實現：
- –null-string如果指定列為字串型別，使用指定字串替換值為null的該類列的值
- –null-non-string如果指定列為非字串型別，使用指定字串替換值為null的該類列的值
增量匯入
增量匯入：（根據時間來匯入，如果表中沒有時間屬性，可以增加一列時間簇）
核心引數：
- –check-column 用來指定一些列，這些列在增量匯入時用來檢查這些資料是否作為增量資料進行匯入，和關係型資料庫中的自增欄位及時間戳類似. 注意:這些被指定的列的型別不能使任意字元型別（在關係資料庫中），如char、varchar等型別都是不可以的，同時–check-column可以去指定多個列
- –incremental 用來指定增量匯入的模式，兩種模式分別為Append和Lastmodified
- –last-value 指定上一次匯入中檢查列指定欄位最大值，即會匯入比lastvalue指定值大的資料記錄
注意：上面三個引數都必須新增！
執行語句：
sqoop import -D mapred.job.queue.name=datacenter
--connect jdbc:mysql://localhost:3306/anticheat?tinyInt1isBit=false
--username root
--password ^qn9DFYPm
--table anticheat_blacklist
--delete-target-dir
--hive-import --fields-terminated-by '\t'
--beeline "jdbc:hive2://dsrv2.heracles.sohuno.com:10000/test;principal=hive/[email protected];"
--hive-database test
--hive-table anticheat_blacklist_mysql
--null-string '\\N'
--hive-overwrite
--num-mappers 1
--check-column update_time
--incremental lastmodified
--last-value "2019-04-12 14:31:34"
--outdir /home/test/data/anticheat/mysql2hive
以上語句使用 lastmodified 模式進行增量匯入，結果報錯：
錯誤資訊：--incremental lastmodified option for hive imports is not supported. Please remove the parameter --incremental lastmodified
錯誤原因：Sqoop 不支援 mysql轉hive時使用 lastmodified 模式進行增量匯入，但mysql轉HDFS時可以支援該方式！
我們使用append方式匯入：
sqoop import -D mapred.job.queue.name=datacenter
--connect jdbc:mysql://localhost:3306/anticheat?tinyInt1isBit=false
--username root
--password ^qn9DFYPm
--table anticheat_blacklist
--delete-target-dir
--hive-import --fields-terminated-by '\t'
--hive-database test
--hive-table anticheat_blacklist_mysql
--null-string '\\N'
--null-non-string '\\N'
--num-mappers 1
--check-column update_time
--incremental append
--last-value "2019-04-12 14:31:34"
--outdir /home/test/data/anticheat/mysql2hive
增量匯入成功！
轉自：https://www.cnblogs.com/xiaodf/p/10712766.html