Hadoop-25 Sqoop迁移 增量数据导入 CDC 变化数据捕获 差量同步数据 触发器 快照 日志

章节内容

上节我们完成了如下的内容：

• Sqoop MySQL迁移到Hive
• Sqoop Hive迁移数据到MySQL
• 编写脚本进行数据导入导出测试

背景介绍

这里是三台公网云服务器，每台 2C4G，搭建一个Hadoop的学习环境，供我学习。
之前已经在 VM 虚拟机上搭建过一次，但是没留下笔记，这次趁着前几天薅羊毛的3台机器，赶紧尝试在公网上搭建体验一下。

• 2C4G 编号 h121
• 2C4G 编号 h122
• 2C2G 编号 h123

CDC

全称为：变化数据捕获（Change Data Capture）
我们前面执行的都是全量数据的导入。

• 如果数据量很小，则采取完全源数据抽取
• • 如果源数据量很大，则需要抽取发生变化的数据，这种数据抽取模式叫：“变化数据捕获”，简称 CDC。

如果CDC是侵入式的，那么操作会给源系统带来性能的影响。

基于时间戳

抽取过程可以根据某些属性列来判断哪些数据是增量的，最常见的属性列有以下两种：

• 时间戳：最好有两个列，一个插入时间戳，表示何时创建，一个更新时间戳表示最后一次更新时间。
• 序列：大多数数据库都提供自增功能，表中的列定义成自增的，很容易得根据该列识别新插入的数据

时间戳是最简单且常用的，但是有如下缺点：

• 不能记录删除记录的操作
• 无法识别多次更新
• 不具有实时的能力

基于触发器

当执行：INSERT、UPDATE、DELTE 这些 SQL 语句时，激活数据库的触发器，使用触发器可捕获变更的数据，并把数据保存中间临时表里。
大多数场合下，不允许向操作性数据库里添加触发器，且这种会降低系统性能，基本不会采用。

基于快照

可以通过比较源表和快照表来得到数据的变化，基于快照的CDC可以检测插入、更新、删除等数据，这是相对于时间戳的CDC方案的优点。
缺点就是需要大量的空间。

基于日志

最复杂和没有侵入性的就是基于日志的方式，数据库把每个插入、更新、删除都记录到日志里，解析日志文件，就可以获取相关的信息。
每个关系型数据库：日志格式不一致，没有通用的产品。
阿里巴巴的Canal可以完成MySQL日志文件解析。

Append方式

初始化数据

删除 MySQL 中的数据

-- 删除 MySQL 表中的全部数据
truncate table sqoop.goodtbl;

删除Hive中的数据

-- 删除 Hive 表中的全部数据
truncate table mydb.goodtbl;

重新生成数据

这个SQL是之前章节写的函数方法，如果你第一次看到这里，你可能需要把前边的文章执行一次。

-- 向MySQL的表中插入100条数据
call batchInsertTestData(1, 100);

导入Hive

sqoop import \
--connect jdbc:mysql://h122.wzk.icu:3306/sqoop \
--username hive --password [e-mail] \
--table goodtbl \
--incremental append \
--hive-import \
--fields-terminated-by "\t" \
--hive-table mydb.goodtbl \
--check-column serialNumber \
--last-value 50 \
-m 1

以上参数说明：

• check-column 用来指定一些列，来检查是否可以作为增量数据进行导入，和关系型数据库自增或时间戳类似。
• last-value 制定上一次导入检查列指定字段的最大值

检查Hive

我们通过指令查看 Hive 同步了多少数据过来：

select count(*) from mydb.goodtbl;

继续生成

call batchInsertTestData(200, 1000);

增量导入

sqoop import \
--connect jdbc:mysql://h122.wzk.icu:3306/sqoop \
--username hive --password [e-mail] \
--table goodtbl \
--incremental append \
--hive-import \
--fields-terminated-by "\t" \
--hive-table mydb.goodtbl \
--check-column serialNumber \
--last-value 100 \
-m 1

检查Hive

重新查看Hive，看看目前同步了多少数据过来

select count(*) from mydb.goodtbl;