头条项目推荐的相关技术(一): 数据库的迁移与定时迁移技术

1. 写在前面

这里是有关于一个头条推荐项目的学习笔记，主要是整理工业上的推荐系统用到的一些常用技术，这是第一篇，首先是数据库的迁移技术。在实际生产环境中，我们要处理的数据可能来自各个地方，业务数据库，爬虫数据库，日志文件，api网关买入数据等，而我们实际做分析或者推荐的时候，往往不能直接在原数据服务器上做的，而是需要先进行一个数据库的迁移，本篇内容主要介绍业务数据存储在mysql中的时候，我们如何把这个数据库迁移到Hadoop中，用Hive查询，在大数据平台上进行分析。主要包括：

为什么需要数据库的迁移以及需求
如何进行数据库的迁移
应用Sqoop完成头条业务数据到Hive的导入

Ok, let’s go!

2. 为什么需要数据库的迁移及需求

我们的web开发和我们的推荐系统所使用的服务器是不同的，这两块肯定是分开的，各自有各自的任务。因此我们从web开发中获取到的某些业务数据表，往往是没法直接就能在做推荐系统开发的上面直接获取到的。因此我们就需要一个数据库的迁移操作，把web开发中获取到的各种业务数据表导入到推荐系统所在的服务器上去，而且我们还得保证实时性，也就是还有一个增量更新的需求。这就是我们做数据库的迁移的原因。

业务mysql数据库中的数据，会同步到我们hadoop的Hive数据仓库中，两个好处：

为了避免直接连接，操作业务数据，造成一些错误
同步一份数据在集群中方便进行数据分析操作

好了，介绍完为啥这样做之后，就看看怎么做了。

3. 如何进行数据库的迁移

我们是将mysql中的表导入到Hive中，因为Hive非常适合离线的处理和分析，导入的方式会用到Sqoop这个工具，关于具体的导入方式，实际上当时搭建大数据环境的时候，就介绍过，这里还是那样做的，只不过数据量比之前大了些而已。所以想要玩接下来的操作，还是得先有大数据的环境，然后安装好mysql, hive等，还有sqoop等，具体的参考我之前的文章。

关于Sqoop：

Sqoop(发音：skup)是一款开源的工具，主要用于在Hadoop(Hive)与传统的数据库(mysql、postgresql…)间进行数据的传递，可以将一个关系型数据库（例如： MySQL ,Oracle ,Postgres等）中的数据导进到Hadoop的HDFS中，也可以将HDFS的数据导进到关系型数据库中。

3.1 Mysql 业务数据

头条推荐这里用到了5张表：

用户资料信息的两张表：user_profile，user_basic
文章内容基本信息、频道三张表：news_article_basic，news_article_content，news_channel

具体内容这里就先不看了，下面主要是介绍迁移技术，内容的话后面应该会再遇到。

3.2 sqoop迁移Mysql业务数据

用sqoop从mysql中迁移数据，我上面那篇文章里面也介绍过，当然这里还有些不太一样的东西，所以也整理下。

首先，做一个Sqoop连接mysql的一个测试，看看能不能连接成功：

sqoop list-databases --connect jdbc:mysql://192.168.56.101:3306/ --username root -P

如果能连接成功，就会自动的显示mysql里面已经有的数据库：

业务数据的导入问题：

问题：每天会有新的用户，文章内容更新
- 新增的用户和文章
- 修改的用户信息，文章信息
两种导入形式：全量导入和增量导入，我们一般选择增量，定期导入新数据

Sqoop支持两种方式的数据导入，即全量数据导入和增量数据导入，下面介绍两种导入形式：

Sqoop全量导入，就是一次性将所有需要的数据，从关系型数据库一次性导入到Hadoop生态中（可以是HDFS，Hive，Hbase等）。这里使用的shell脚本的方式，格式如下：
```
#!/bin/bash# 把需要导入的表写成一个数组的形式，下面循环导入
array=(user_profile user_basic news_user_channel news_channel user_follow user_blacklist user_search news_collection news_article_basic news_article_content news_read news_article_statistic user_material)for table_name in ${array[@]};
dosqoop import \--connect jdbc:mysql://192.168.56.101/toutiao \     # 数据库--username root \    # 用户名--password password \  # 密码--table $table_name \     # 表名--m 5 \    # 操作线程数量--hive-home /root/bigdata/hive \  # Hive 路径  全量导入这里是不需要创建HIVE表的，自动的会创建HIVE表--hive-import \      # 导入形式--create-hive-table  \ --hive-drop-import-delims \   --warehouse-dir /user/hive/warehouse/toutiao.db \--hive-table toutiao.$table_name
done
```
全量导入的形式比较简单，适用场景一次性离线分析场景，上面我做了一些注释，真正跑这个shell脚本的时候，是不能有这些注释的。这个代码目前我没有测，因为我拿到的数据直接是下面这种形式了

这种形式的数据，并不是mysql数据的格式，这种只需要在Hive中创建表的时候采用指定location就可以导入到Hive了，所以我直接是到了后面的应用那块，但是感觉这两种形式还是应该记录下的。这种一般常用，因为真实场景中一次性离线分析的东西很少，往往是那种长久不变的，数据格式混乱的这种，比如文章的内容啊等，常用的是下面的增量导入。
增量导入
实际生产环境中，不断有业务相关的数据产生到关系型数据库，系统需要定期从数据库向hadoop导入数据，导入数仓后，继续进行离线分析。我们不可能将所有的数据重新导一遍，此时就需要用sqoop的增量数据导入模式。

两种导入方式：
- append：即通过制定一个递增的列导入，--incremental append --check-column num_iid --last-value 0, 这个的意思就是按照num_iid进行增量导入， last_value表示上一次导入的位置，后面导入的时候，会接着这个id号往下导入。
  
  举个栗子，有一个订单表，里面每个订单有一个唯一标识自增列ID，在关系型数据库中以主键形式存在。之前已经将id在0~5201314之间的编号的订单导入到Hadoop中了（这里为HDFS），一段时间后我们需要将近期产生的新的订单数据导入Hadoop中（这里为HDFS），以供后续数仓进行分析。此时我们只需要指定–incremental 参数为append，–last-value参数为5201314即可，表示只从id大于5201314后开始导入。
  
  数据格式：
```
1.  sqoop import \
2.      --connect jdbc:mysql://192.168.15.111:3306/test \  #连接到指定数据库
3.      --username root \
4.      --password 123456 \
5.      --table order_table \     # 指定数据库的指定表
6.      --target-dir /user/mysql_to_hdfs  \
7.      --m 3  \
8.      –-hive import \
9.      –-incremental append \        # 指明模式
10.     –-check-column order_id         # 指明用于增量导入的参考列
11.     –-last-value 5201314 \    # 指定参考列上次导入的最大值
```
  这种方式的缺陷就是万一没有递增的列的话，就不好使了。
- incremental: 时间戳的方式，此方式要求原有表中有time字段，它能指定一个时间戳，让Sqoop把该时间戳之后的数据导入至Hadoop（这里为HDFS）。比如我的头条业务数据库中，某篇文章的点赞数增加或减少了，变成了一个新的数据，在我指定的时间戳后面作为新的业务数据导入到了Hadoop（这里指HDFS），我们可以指定给merge-key参数，例如是article_id，表示将后续的新的记录与原有的记录合并。这里和上面的后3行对比下，前面的一致。
```
--incremental lastmodified \         # 最后的修改时间
--check-column column \          # 时间列
--merge-key key \         # 检查列， 合并(主键， 合并键值相同的记录)
--last-value '2012-02-01 11:0:00'          # 时间阈值
```
  一般会用这种。就是只导入check-column的列比’2012-02-01 11:0:00’更大的数据,按照key合并

导入到Hive之后，最终的结果也会有两种形式：

直接Sqoop导入到Hive，这种形式，是在Hive中直接创建好了表，但注意，-incremental lastmodified模式不支持导入Hive。也就是这种增量模式不能用这种方式。那怎么办呢？下面这种
Sqoop导入到Hdfs，然后建立Hive表关联，导入的时候要指定关联位置 --target-dir /user/hive/warehouse/toutiao.db/，这时候进入Hive交互界面，在数据库中建立表的时候，表名和传上来的数据文件一样，hive就能自动将数据映射到hive表中了。

从下面的迁移案例中，我们就能看到区别，当然下面的案例我没试过，因为我拿的数据并不是SQL表的数据，而是直接放到toutiao.db里面的数据。这种情况下我想用的时候，直接建表就行了。但真实的导入过程依赖的是下面的导入脚本，要看明白下面几个表导入的导入脚本的区别。

4. Sqoop迁移案例

这个在没看视频之前，我就自己在我的大数据平台上玩了一遍，只可惜，创建Hive表的时候，每个字段的类型我不知道，胡乱指定的，所以看完这个视频之后，才发现原来人家说了咋导入了啊，只能删除又按照人家的创建表的格式走了一遍。

这个东西其实还是挺简单的，他们分享的数据文件是下面的这个样子：

头条数据库里面包括这5张表，每张里面都是下面这种形式的，按照我上面那篇文章记录的，先把这5张表传到HDFS上，然后在Hive中创建表：

# 这种方式其实就是先把数据导入到hdfs，然后建立Hive表关联 location
create external table stu(id int, fname string, lname string) row format delimited fields terminated by ',' location '/tmp/u1/';

这种格式创建5张表就OK了，只可惜我这里字段名称都指定错了。这里的一个坑就是要指定分隔符。

导入数据到hive中，需要在创建HIVE表加入 row format delimited fields terminated by ‘,’

原因： sqoop 导出的 hdfs 分片数据，都是使用逗号 , 分割的，由于 hive 默认的分隔符是 /u0001（Ctrl+A）,为了平滑迁移，需要在创建表格时指定数据的分割符号。否则hadoop数据在hive中查询就全是NULL。

下面开始按照人家说的玩了，首先，开启我久违的大数据平台，关于这个的搭建过程，可以参考我之前写的大数据环境搭建系列(也是亲手走了一遍，记录了整个历程，已经把坑趟完了)。开启hadoop集群。

我这里已经把上面的五个表目录传到了HDFS上：

下面就直接按照上面的格式导入了：

首先，用户的两张表

user_profile表:
使用lastmodified模式
mysql数据库中更新时候update_time会修改最近时间，按照user_id合并(增量导入进去会有新的重复的数据，需要合并)
指定last_time时间
user_basic表：
使用lastmodified模式
last_login作为更新时间
指定last_time时间，按照user_id合并

Mysql导入对应hive类型:

MySQL(bigint) --> Hive(bigint)
MySQL(tinyint) --> Hive(tinyint)
MySQL(int) --> Hive(int)
MySQL(double) --> Hive(double)
MySQL(bit) --> Hive(boolean)
MySQL(varchar) --> Hive(string)
MySQL(decimal) --> Hive(double)
MySQL(date/timestamp) --> Hive(string)

先进入Hive，建立一个toutiao数据库，然后开始创建这5张表，这里只给出其中的一个案例代码：

create table user_profile(
user_id BIGINT comment "userID",
gender BOOLEAN comment "gender",
birthday STRING comment "birthday",
real_name STRING comment "real_name",
create_time STRING comment "create_time",
update_time STRING comment "update_time",
register_media_time STRING comment "register_media_time",
id_number STRING comment "id_number",
id_card_front STRING comment "id_card_front",
id_card_back STRING comment "id_card_back",
id_card_handheld STRING comment "id_card_handheld",
area STRING comment "area",
company STRING comment "company",
career STRING comment "career")
COMMENT "toutiao user profile"
row format delimited fields terminated by ','
LOCATION '/tmp/toutiaodata/user_profile';

人家这里直接指明好了数据格式了，我之前自己玩的时候全指定成string，真是醉了。这里的话就是复制代码创建表就行了，SQL语句。这样就可以把用户的两个表导入成功。

这样，初始表就建好了，那么如果想增量导入数据的话，既然把Hive中的表和HDFS进行了关联，此时只需要动态更新HDFS里面的数据就好了。

这里可以写个脚本import_incremental.sh，这个是从mysql表中把数据传入到HDFS，并进行增量更新。

-m：指定同时导入的线程数量
连接地址以及账号密码，表、目录
指定要导入的时间

time=`date +"%Y-%m-%d" -d "-1day"`
declare -A check
check=([user_profile]=update_time [user_basic]=last_login [news_channel]=update_time)
declare -A merge
merge=([user_profile]=user_id [user_basic]=user_id [news_channel]=channel_id)for k in ${!check[@]}
dosqoop import \--connect jdbc:mysql://192.168.56.101/toutiao \--username root \--password password \--table $k \   --m 4 \--target-dir /user/hive/warehouse/toutiao.db/$k \--incremental lastmodified \--check-column ${check[$k]} \       # update_time --merge-key ${merge[$k]} \     # 指定合并的主键，比如用户修改了用户信息， 那么我们导入的时候是要和原来的用户信息合并，然后把用户信息改成最新的，而这里合并的基础就是主键--last-value ${time}      # 从这个时间之后导入新数据
done

用户的两张表比较简单，用的脚本也是上面这个，因为两个表中都有个更新时间字段，我们就可以根据这个进行增量导入。

下面是文章的3张表的导入，先解释增量更新的类型：

news_article_basic的导入

按照键review_time更新
合并按照article_id
指定时间

create table news_article_basic(
article_id BIGINT comment "article_id",
user_id BIGINT comment "user_id",
channel_id BIGINT comment "channel_id",
title STRING comment "title",
status BIGINT comment "status",
update_time STRING comment "update_time")
COMMENT "toutiao news_article_basic"
row format delimited fields terminated by ','
LOCATION '/user/hive/warehouse/toutiao.db/news_article_basic';

原mysql数据库中某些字段的值存在一些特定的字符(比如title字段)，如","、"\t", "\n"这些字符都会导致导入到hadoop被hive读取失败，解析时会认为另一条数据或者多一个字段。当然我这里是人家导入到hadoop的数据了，不会出现这个问题。

解决方案：

在导入时，加入—query参数，从数据库中选中对应字段，过滤相应内容，使用REPLACE、CHAR(或者CHR)进行替换字符

并且如果mysql表中存在tinyibt数据类型，在导入到hive的时候，会自动的变成bool类型，这种情况也要规避掉，所以必须在connet中加入: ?tinyInt1isBit=false **防止默认到HIVE中，字段默认被转换为boolean类型的数据。

下面是导入脚本，这个表示的就是从MySQL表中导入的时候，把某些字段的特殊字符给过滤掉。

# 方案：导入方式，过滤相关字符
sqoop import \--connect jdbc:mysql://192.168.19.137/toutiao?tinyInt1isBit=false \--username root \--password password \--m 4 \--query 'select article_id, user_id, channel_id, REPLACE(REPLACE(REPLACE(title, CHAR(13),""),CHAR(10),""), ",", " ") title, status, update_time from news_article_basic WHERE $CONDITIONS' \--split-by user_id \--target-dir /user/hive/warehouse/toutiao.db/news_article_basic \--incremental lastmodified \--check-column update_time \--merge-key article_id \--last-value ${time}

这里的CHAR(13)表示的是\n, CHAR(10)表示的是\t，这种直接替换。所以这些特殊字符是在MySQL往Hadoop导入的过程中处理的，而不是在Hive与Hadoop关联的过程中处理的。

news_channel的导入：

跟上面用户的表处理方式相同，按照update_time键更新
按照channel_id合并
更新时间

create table news_channel(
channel_id BIGINT comment "channel_id",
channel_name STRING comment "channel_name",
create_time STRING comment "create_time",
update_time STRING comment "update_time",
sequence BIGINT comment "sequence",
is_visible BOOLEAN comment "is_visible",
is_default BOOLEAN comment "is_default")
COMMENT "toutiao news_channel"
row format delimited fields terminated by ','
LOCATION '/user/hive/warehouse/toutiao.db/news_channel';

这个的导入脚本和用户的那两个一致。

news_article_content文章内容表中含有过多特殊字符，选择直接全量导入。

news_article_content)
create table news_article_content(
article_id BIGINT comment "article_id",
content STRING comment "content")
COMMENT "toutiao news_article_content"
row format delimited fields terminated by ','
LOCATION '/user/hive/warehouse/toutiao.db/news_article_content';

直接导入到HIVE汇总，导入脚本为，这里是全量导入的方式：

# 全量导入(表只是看结构，不需要在HIVE中创建，因为是直接导入HIVE，会自动创建
sqoop import \--connect jdbc:mysql://192.168.19.137/toutiao \--username root \--password password \--table news_article_content \--m 4 \--hive-home /root/bigdata/hive \   # 这个地方要看到和上面那几个增量导入的区别--hive-import \--hive-drop-import-delims \--hive-table toutiao.news_article_content \--hive-overwrite

脚本的这几个都没有测试，不过建表的都已经测试成功。

所以重点就是全量导入和增量导入的区别：

全量导入是直接导入到HIVE中，但是不需要自己手动创建HIVE表，从MySQL导入的时候，会自动的创建好HIVE表
增量导入是从MySQL中导入到hdfs里面，然后再和HIVE进行关联。看导入脚本就能够看出区别来。
一般表里面有过多的那种特殊字符，比如爬取了一个网页的这种，不是那种结构化的数据的时候，此时可以选择全量导入的方式。

上面的每个表导入的时候，都是给了两串代码，容易看懵，第一串代码这样操作是人家直接把所有的数据表放到了HDFS上，这时候，我们直接导入到HIVE就可以了，此时就是在HIVE中创建表然后直接关联即可。

而下面的shell脚本的方式，是我们如果是从mysql中直接往HIVE导入数据的操作情况，这时候两种方式，一种是全量导入的脚本，对应的是复杂结构的数据，一种是增量导入的方式，对应的是需要实时更新的数据，而这两种的区别就是上面这个了。所以不要懵了，真实场景下，下面这个脚本的方式才是我们要做的。

一般会把上面这些导入脚本放到一个.sh文件中，然后一键运行，就完成导入了啦。

当然，我们还会设置成定时运行这个脚本文件，因为我们导入数据要保证实时性。看看这个怎么玩。

5. crontab-shell 脚本定时运行

crobtab是用来安装, 卸载或者列出定时任务列表的命令。每个用户都可以拥有自己的crontab文件，虽然这些文件都位于 /var/spool/cron/ 目录中，并不需要直接编辑它们。需要通过crontab命令来编辑或者配置你自己的定时任务。这个东西Linux里面自动就有了，可以直接使用。关于这个命令的详解，参考这篇博客，这里只记录简单的使用情况。

首先编辑crontab文件，在这个文件里面制定好需要定时运行的脚本，系统就会自动的按照我们指定的时间自动的运行脚本。输入crontab -e 来编辑这个脚本文件。

然后就是输入的格式：我们可以参考一下crontab的配置文件：

这里举几个例子：

# 每个5分钟运行一次 backupscript脚本
*/5 * * * * /root/backupscript.sh# 每天的凌晨1点运行backupscript脚本
0 1 * * * /root/backupscript.sh# 每个月的第一个凌晨3:15运行backupscript脚本
15 3 1 * * /root/backscript

如果是指定具体的时间点的话，就是直接在*的位置填入相应的时间即可。而如果是每隔多长时间的话，在对应位置的* 后面加/加数字。

我们这里可以选择定时每隔半个小时增量导入，也就是执行写好的增量导入脚本：

*/30 * * * * /home/icss/workspace/toutiao_project/scripts/import_incremental.sh

这样就OK了，这个import_cremental.sh里面就是上面导入5张表的shell脚本，前4张表用的是增量导入的方式，最后一张用的是全量导入的方式。

关于crontab常用的命令：

crontab -e  // 修改crontab文件，如果文件不存在会自动创建
crontab -l // 显示crontab文件
crontab -r  // 删除crontab文件
crontab -ir  // 删除crontab文件前提醒用户service crond status   // 查看crontab服务状态
service crond start    // 启动服务
service crond stop      // 关闭服务
service crond restart   // 重启任务
service crond reload    // 重新载入配置

看完之后，手动走了一遍，发现了个坑，原来我们编辑定时任务的时候，是直接crontab -e然后在里面输入我们要定时执行的任务就完事。这时候点击保存之后，自动的就会在/var/spool/cron目录下生成一个用户名文件， cat下就会看到我们要自动执行的命令。

我一开始没明白，还手动在那个目录下建了一个root目录，结果输入编辑命令的时候，给我创的文件放到了一个临时目录/tmp中，就是第一行的那个文件里面。然后通过百度，我把手动创建的目录删除，然后又创新编辑，才得到了最终的root，原理这玩意是个文件啊。所以我们编辑命令的时候，就crontab -e就行，系统会自动的根据当前用户创建文件，并定期执行自动运行的任务了。如果不需要执行了，可以删掉这个东西，也可以停止crontab服务。还是删掉吧。

所以总结起来就是四步：

创建一个定时运行的脚本 crontab -e
写入定时执行的命令：按照命令的格式
启动服务： service crond start
关闭服务： service crond stop

参考：

B站黑马头条实战课程
黑马头条推荐项目知识点总结