Flink SelectDB Connector 介绍

概述

随着云基础设施的不断完善，云原生已经成为各行业数字化转型的必选项，越来越多的应用开始进行云原生化架构升级和应用迁移。

而云原生实时数仓的出现，让传统的数据仓库无论是成本、灵活性还是开放性等方面都显露出不足。拥有高性能、高可用性、可伸缩性、高安全性等特征的云原生数据库，正在成为企业的首选。

SelectDB Cloud 作为一款运行于多云之上的云原生实时数据仓库，可以为客户提供极简运维和极致性价比的数仓服务，为用户提供开箱即用的能力。

同时，SelectDB Cloud 结合 Flink 流式计算，可以让用户将 Kafka 中的非结构化数据以及 MySQL 等上游业务库中的变更数据，实时同步到 SelectDB Cloud中，同时 SelectDB Cloud 提供亚秒级分析查询的能力，可以有效地满足实时 OLAP、实时数据看板以及实时数据服务等场景的需求。

Flink SelectDB Connector 原理

架构设计

在实时计算中，通过Flink可以将业务数据库(MySQL、SQLServer、Oracle)或Kafka消息队列等其他上游数据作为Source读取出来，经过FlinkSQL或着DataStream加工计算，最后将清洗后的数据写入。

Sink写入时使用Flink SelectDB Connector，将数据先上传到internal stage，最后通过copy into的方式将文件一次性加载到表中。

Exactly-Once

在实时写入场景中，如何确保端到端的数据一致性是经常遇到的问题，Flink SelectDB Connector基于Flink的两阶段提交，实现了端到端的Exactly-Once。

以Kafka为例，接收到上游的数据后，会持续的发送到Sink端
Sink端会将数据以文件的形式周期性的写入到SelectDB Cloud的InternalStage。
当Checkpoint完成后，会将刚写入Internal Stage的文件列表一次性通过Copy Into的方式导入到Table中。
Flink任务意外挂掉后，从上个Checkpoint重启，会自动Commit上次未完成的Copy任务。

使用 Flink CDC 同步 MySQL 数据到 SelectDB Cloud

1.下载安装Flink

这里以Flink1.15版本为例

wget https://dlcdn.apache.org/flink/flink-1.15.3/flink-1.15.3-bin-scala_2.12.tgz
tar -zxvf flink-1.15.3-bin-scala_2.12.tgz
cd flink-1.15.3
wget https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.3.0/flink-sql-connector-mysql-cdc-2.3.0.jar -P ./lib/
wget https://repo1.maven.org/maven2/com/selectdb/flink-selectdb-connector-1.15/2.0.0/flink-selectdb-connector-1.15-2.0.0.jar  -P ./lib/

2.启动Flink Standalone集群

bin/start-cluster.sh
Starting cluster.
Starting standalonesession daemon on host flink.
Starting taskexecutor daemon on host flink.

3.在MySQL和SelectDBCloud上创建表，并且在MySQL中预先插入数据

-- 创建MySQL表
CREATE TABLE test.employees (
    emp_no      INT             NOT NULL,
    birth_date  DATE            NOT NULL,
    first_name  VARCHAR(14)     NOT NULL,
    last_name   VARCHAR(16)     NOT NULL,
    gender      ENUM ('M','F')  NOT NULL,    
    hire_date   DATE            NOT NULL,
    PRIMARY KEY (emp_no)
);

-- 插入数据
INSERT INTO test.employees VALUES (10001,'1953-09-02','Georgi','Facello','M','1986-06-26'),
(10002,'1964-06-02','Bezalel','Simmel','F','1985-11-21'),
(10003,'1959-12-03','Parto','Bamford','M','1986-08-28'),
(10004,'1954-05-01','Chirstian','Koblick','M','1986-12-01'),
(10005,'1955-01-21','Kyoichi','Maliniak','M','1989-09-12');

-- selectdb cloud中创建表
CREATE TABLE test.employees (
    emp_no       int NOT NULL,
    birth_date   date,
    first_name   varchar(20),
    last_name    varchar(20),
    gender       char(2),
    hire_date    date
)
UNIQUE KEY(`emp_no`)
DISTRIBUTED BY HASH(`emp_no`) BUCKETS 1

4.在Flink SQL Client上提交Flink任务

SET 'execution.checkpointing.interval' = '10s';

CREATE TABLE employees_source (
    emp_no INT,
    birth_date DATE,
    first_name STRING,
    last_name STRING,
    gender STRING,
    hire_date DATE,
    PRIMARY KEY (`emp_no`) NOT ENFORCED
) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = '127.0.0.1',
    'port' = '3306',
    'username' = 'root',
    'password' = '123456',
    'database-name' = 'test',
    'table-name' = 'employees'
);

CREATE TABLE employees_sink (
    emp_no       INT ,
    birth_date   DATE,
    first_name   STRING,
    last_name    STRING,
    gender       STRING,
    hire_date    DATE
) 
WITH (
  'connector' = 'selectdb',
  'load-url' = '127.0.0.1:36252',
  'jdbc-url' = '127.0.0.1:19846',
  'cluster-name' = 'cluster1',
  'table.identifier' = 'test.employees',
  'username' = 'admin',
  'password' = '',
  'sink.enable-delete' = 'true'
);

INSERT INTO employees_sink select * from employees_source

5.在MySQL中执行增删改操作

INSERT INTO test.employees VALUES (10006,'1953-04-20','Anneke','Preusig','F','1989-06-02');

UPDATE test.employees set last_name = 'update' where emp_no = 10001;

DELETE FROM test.employees WHERE emp_no = 10002;

6.在 SelectDB Cloud 中验证

mysql> select * from employees;
+--------+------------+------------+-------------+--------+------------+
| emp_no | birth_date | first_name | last_name   | gender | hire_date  |
+--------+------------+------------+-------------+--------+------------+
|  10001 | 1953-09-02 | Georgi     | update      | M      | 1986-06-26 |
|  10003 | 1959-12-03 | Parto      | Bamford     | M      | 1986-08-28 |
|  10004 | 1954-05-01 | Chirstian  | Koblick     | M      | 1986-12-01 |
|  10005 | 1955-01-21 | Kyoichi    | Maliniak    | M      | 1989-09-12 |
|  10006 | 1953-04-20 | Anneke     | Preusig     | F      | 1989-06-02 |
+--------+------------+------------+-------------+--------+------------+

连接 JDBC 数据源 Kafka SelectDB Connector 介绍