一、FlinkX实时采集功能的基本介绍

首先为大家介绍下FlinkX实时模块的分类，如下图所示：

1、实时采集模块(CDC)

1）MySQL Binlog插件

利用阿里开源的Canal组件实时从MySQL中捕获变更数据。

2）PostgreSQL Wal插件

PostgreSQL 实时采集是基于 PostgreSQL的逻辑复制以及逻辑解码功能来完成的。逻辑复制同步数据的原理是，在Wal日志产生的数据库上，由逻辑解析模块对Wal日志进行初步的解析,它的解析结果为ReorderBufferChange（可以简单理解为HeapTupleData），再由Pgoutput Plugin对中间结果进行过滤和消息化拼接后，然后将其发送到订阅端，订阅端通过逻辑解码功能进行解析。

2、消息队列

1）Kafka：Kafka插件存在四个版本，根据Kafka版本的不同，插件名称也略有不同。具体对应关系如下表所示:

2）EMQX：EMQX 是一款完全开源，高度可伸缩，高可用的分布式MQTT消息服务器，适用于IoT、M2M 和移动应用程序，可处理千万级别的并发客户端。

3、间隔轮询

RDB类型插件的使用限制：

    只有RDB类型的reader插件支持间隔轮询
    轮询字段只能为数值类型或者时间类型
    轮询字段只能为连续递增且不重复

4、其他

Hive插件: Hive插件只有写入插件，功能基于HDFS的写入插件实现，也就是说从实时采集插件读取，写入Hive也支持失败恢复的功能。

二、Binlog实时采集原理

1、什么是Binlog

MySQL 的二进制日志 Binlog 可以说是 MySQL 最重要的日志，它记录了所有的 DDL 和 DML 语句（除了数据查询语句Select、Show等），以事件形式记录，还包含语句所执行的消耗的时间，MySQL的二进制日志是事务安全型的，Binlog 的主要目的是复制和恢复。

2、Binlog插件基本原理

实时采集插件的核心是如何实时捕获数据库数据的变更，对于MySQL数据库而言，阿里开源的Canal已经很好的帮我们实现了基于MySQL数据库增量日志解析，提供增量数据订阅和消费功能。因此这里我们直接用Canal捕获MySQL数据库数据的变更信息，基于FlinkX框架将任务简化成脚本的配置，基于Flink的Checkpoint机制提供了任务的故障恢复，提高了任务的容错性。

基本步骤如下：

    任务启动时启动Canal线程
    Canal模拟MySQL slave的交互协议，伪装自己为MySQL slave，向MySQL master发送dump协议
    MySQL master收到dump请求，开始推送Binary Log给slave(即Canal)
    Canal解析Binary Log 对象(原始为Byte流)
    FlinkX获取Canal解析后的对象做二次解析，封装后发送至下游数据源

三、Binlog到Hive实战

1、环境准备：确认数据库开启了Binlog

show variables like '%log_bin%';

2、建表

    CREATE TABLE `kudu` (  
      `id` bigint(11) NOT NULL AUTO_INCREMENT,  
      `user_id` bigint(11) DEFAULT NULL,  
      `name` varchar(255) DEFAULT NULL,  
      PRIMARY KEY (`id`)  
    ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4

本文首发于：数栈研习社

数栈是一站式的数据中台PaaS，Flinkx这个项目我们也在github社区上开源了，大家如果有想法和兴趣欢迎来撩~