记一次sqoop同步到mysql

工作中需要用到将hive的数据导一份到mysql中，需求是这样的：hive每天会产生一份用户标签(tag)发生变化的结果表user_tag，这份结果同步到mysql中，并且保持一份全量表，存储当前用户的标签。最初打算是在mysql建立一个分区表，按照天存储user_tag，研究了一下mysql的分区表，最终没有使用，既然学习了就做一下笔记。

mysql按照时间分区

mysql的分区方法分为好几种range，list，hash，key等方法，具体可以自行搜索一下，这里要讲的是range方法，下面是建表语句

CREATE TABLE `histdata` (
  `uid` char(36) NOT NULL,
  `date` date NOT NULL,
  `cate` int(11) DEFAULT NULL
) 
PARTITION BY RANGE  COLUMNS(`date`)
(PARTITION histdata20180101 VALUES LESS THAN ('2018-01-02'),
PARTITION histdata20180102 VALUES LESS THAN ('2018-01-03')) 

mysql分区表建表的时候需要指定好分区，与hive的分区有很大的不同，并且分区的名字也需要注意，因为这里面使用的是LESS THAN，那么如果每天都需要添加新的分区呢？那就需要修改表了

alter table histdata add PARTITION (PARTITION histdata20180103 VALUES LESS THAN ('2018-01-04'));

手动添加分区太麻烦了，那么能不能自动添加分区呢？ 当然可以，使用存储过程。上学那会经常写java web，一些业务就直接写成了触发器、能够省下不少的代码，下面是存储过程代码：

DELIMITER $$
USE `test_db`$$ -- database name
DROP PROCEDURE IF EXISTS `histdata_add_partition`$$
CREATE DEFINER=`root`@`%` PROCEDURE `histdata_add_partition`()
BEGIN
    SELECT REPLACE(partition_name,'histdata','') INTO @partition_name FROM INFORMATION_SCHEMA.PARTITIONS 
    WHERE table_name='histdata' ORDER BY partition_ordinal_position DESC LIMIT 1;
    SET @head_date= DATE(DATE_ADD(@partition_name, INTERVAL 1 DAY))+0;
        SET @rear_date= DATE(DATE_ADD(@partition_name, INTERVAL 2 DAY));
    SET @s1=CONCAT('ALTER TABLE histdata ADD PARTITION (PARTITION histdata',@head_date,' VALUES LESS THAN (''',DATE(@rear_date),'''))');
    SELECT @s1;
    PREPARE stmt2 FROM @s1;
    EXECUTE stmt2;
    DEALLOCATE PREPARE stmt2;
    COMMIT ;
END$$
DELIMITER ;

上面的作用是从当前分区中找到最大的一个，然后进行时间累加，再创建新的分区，再定义一个事件触发器

DELIMITER $$
CREATE EVENT histdata_event  
          ON SCHEDULE  
          EVERY 1 day STARTS '201x-0x-2x 23:59:59'  
          DO  
      BEGIN  
          CALL `test_db`.`histdata_add_partition`;  
            END $$
DELIMITER ;

show create table 看一下效果：

CREATE TABLE `histdata` (
  `uid` char(36) NOT NULL,
  `date` date NOT NULL,
  `cate` int(11) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8
PARTITION BY RANGE  COLUMNS(`date`)
(PARTITION histdata20180101 VALUES LESS THAN ('2018-01-02') ENGINE = InnoDB,
PARTITION histdata20180102 VALUES LESS THAN ('2018-01-03') ENGINE = InnoDB,
PARTITION histdata20180103 VALUES LESS THAN ('2018-01-04') ENGINE = InnoDB,
PARTITION histdata20180104 VALUES LESS THAN ('2018-01-05') ENGINE = InnoDB,
PARTITION histdata20180105 VALUES LESS THAN ('2018-01-06') ENGINE = InnoDB,
PARTITION histdata20180106 VALUES LESS THAN ('2018-01-07') ENGINE = InnoDB)

可以看到效果很不错，添加了分区。不过由于分区表需要修改表添加分区，涉及到权限问题，最终并没有使用。

使用sqoop同步

下面是所修要用到表的信息，需要先将hive中的数据同步到user_tag中，然后再讲user_tag的内容insert or update 到user_tag_all中，注意到user_tag_all这个表是有主键的

全量表：user_tag_all

用于存储最新的用户Tag,mid为用户id，tag为用户标签，dt为更新日期

CREATE TABLE `user_tag_all` (
  `mid` int(15) NOT NULL,
  `tag` tinyint(4) NOT NULL,
  `dt` date NOT NULL,
  PRIMARY KEY (`mid`)
) ;

每日同步表：user_tag

每日从hive中同步到mysql中，为用户的历史标签信息

CREATE TABLE `user_tag` (
  `mid` int(15) NOT NULL,
  `tag` tinyint(4) NOT NULL,
  `dt` date NOT NULL
);

容错表：user_tag_staging

用于sqoop容错

CREATE TABLE `user_tag_staging` (
  `mid` int(15) NOT NULL,
  `tag` tinyint(4) NOT NULL,
  `dt` date NOT NULL
);

使用的sqoop 导出命令，相关教程网上较多请自行查找

sqoop export -D mapreduce.job.queuename=sqoop
--connect jdbc:mysql://localhost:3306/dbname                        
--username root                        
--password root_pwd                        
--table user_tag                        
--staging-table user_tag_staging                        
--clear-staging-table                        
-m 1                        
--export-dir /user/hadoop/user_tag/dt=2018-03-27                        
--null-string '\N'                        
--null-non-string '\N'                        
--fields-terminated-by 01  

值得注意的是staging-table这个是一个与目标表结构一样的表，是一个中间表，目的是用于对sqoop export进行容错，clear-staging-table将表清空。在操作的过程中自己遇到的的一个问题是mysql表比hive中的表多了dt字段，导致同步的时候一直报错，最终在hive中添加了一个字段用于存储时间(名字任意)，实际上sqoop同步中它先从mysql中查询一条数据，解析出数据的类型，生成一个java文件，再对hdfs文件进行解析，并将数据逐条插入到mysql中，全程与hive无关。接下来就就是插入或者更新user_tag_all这个表了

insert into user_tag_all(mid,tag,dt) select mid,tag,dt from user_tag where dt='2018-03-16' ON DUPLICATE KEY UPDATE tag=VALUES(tag),dt=VALUES(dt);

总结####

本篇文章虽然任务不大，但是从中有许多值得借鉴的知识点

• mysql分区表的使用，以及存储过程的编写
• mysql的不存在则插入、存在则更新的使用
• sqoop命令的使用，中间表的作用

ref###

MySQL每天自动增加分区