实时计算Flink集成开源连接器TiDB CDC Connector案例实践

TIDB部署（阿里云ECS）

1、系统配置

TIDB官方建议使用CentOS7.3及以上版本：

本次实验我们选择CentOS 7.6 64位，考虑网络连通，需将TIDB ECS实例与Flink集群部署在相同VPC网络。

2、TIDB部署

a、下载并安装 TiUP

curl --proto '=https' --tlsv1.2 -sSf https://tiup-mirrors.pingcap.com/install.sh | sh

b. 安装 TiUP 的 cluster 组件

声明全局的环境变量，不然找不到tiup命令：

source .bash_profile

执行安装cluster命令：

tiup cluster

c. 调大 sshd 服务的连接数限制（可选）

#调大sshd服务的连接数限制
cat /etc/ssh/sshd_config | grep -n MaxSessions
vim +41 /etc/ssh/sshd_config

#改完后重启sshd
systemctl restart sshd.service

d. 创建并启动集群

按配置模板，编辑配置文件（当前目录vim编辑后保存即可），命名为 topo.yaml，其中：

• host：部署TIDB的服务器i（ECS内网IP）
• ssh_port默认是22
• 官方文件的tikv_servers是3个节点，这里设置成了只有1个节点，原因是本地配置多个节点时只有1个节点能启动成功

模板如下：

global:
 user: "tidb"
 ssh_port: 22
 deploy_dir: "/tidb-deploy"
 data_dir: "/tidb-data"

monitored:
 node_exporter_port: 9100
 blackbox_exporter_port: 9115

server_configs:
 tidb:
   log.slow-threshold: 300
 tikv:
   readpool.storage.use-unified-pool: false
   readpool.coprocessor.use-unified-pool: true
 pd:
   replication.enable-placement-rules: true
   replication.location-labels: ["host"]
 tiflash:
   logger.level: "info"

pd_servers:
 - host: 192.168.0.1

tidb_servers:
 - host: 192.168.0.1

tikv_servers:
 - host: 192.168.0.1
   port: 20160
   status_port: 20180
   config:
     server.labels: { host: "laomo-talkservice-tidb-1" }

tiflash_servers:
 - host: 192.168.0.1

部署集群：

# cluster-name:集群名称
# tidb-version:TiDB版本号，可以通过tiup list tidb这个命令来查看
tiup cluster deploy   ./topo.yaml --user root -p 

# 这里使用 v6.0.0，集群名称叫mytidb-cluster来部署
tiup cluster deploy mytidb-cluster v6.0.0 ./topo.yaml --user root -p P...D

启动集群：

#启动刚才部署的集群，集群名称为mytidb-cluster
tiup cluster start mytidb-cluster

e. 开启CDC（重要：CDC不开启无法捕获数据变更。另外，本示例单节点，多节点集群需要配置节点访问免密）

# 创建scale-out.yaml文件添加 TiCDC 节点信息
cdc_servers:
  - host: 192.168.0.1
    gc-ttl: 86400

# TiUP运行scale-out命令
tiup cluster scale-out mytidb-cluster scale-out.yaml

#TiUP 停止和启动 TiCDC
tiup cluster stop -R cdc
tiup cluster start -R cdc
tiup cluster restart -R cdc

f. 访问TiDB

在ECS上安装 MySQL 客户端（通过MySQL客户端访问TiDB）：

yum -y install mysql

访问TiDB，密码默认为空

mysql -h 192.168.0.1 -P 4000 -u root

Flink TiDB CDC

TiDB CDC

TiDB CDC connector 是一个 Flink Source connector，它会先读取数据库快照，然后在发生故障时以exactly-once 处理继续读取更改事件，支持exactly-once语义

兼容版本

参数配置

集成TiDB连接器

实时计算Flink版是一套基于Apache Flink构建的⼀站式实时大数据分析平台，支持作业开发、数据调试、运行与监控、自动调优、智能诊断等全生命周期能力。本文案例基于阿里云实时计算Flink版进行演示。

1、版本准备

兼容版本可以参考Flink TiDB CDC小结介绍，本次实验选型如下：

❏ Flink全托管版本：vvr-6.0.4-flink-1.15

❏ TiDB版本：v6.0.0

❏ TiDB CDC Connector：flink-connector-tidb-cdc-2.3.0

2、集群创建

登录阿里云控制台，创建全托管VVP集群，注意集群部署网络环境与TiDB保持一致（同一VPC）：

3、注册TiDB连接器

4、创建Session集群

创建一个vvr-6.0.4-flink-1.15版本的Session集群，创建后并启动:

5、基于Flink SQL开发作业

注意：代码里没有配置scan.startup.mode参数，默认值initial：在第一次启动时，会先扫描历史全量数据（后续会演示相关效果）

CREATE TEMPORARY TABLE students_tidb_source (
  db_name STRING METADATA FROM 'database_name' VIRTUAL,
  table_name STRING METADATA  FROM 'table_name' VIRTUAL,
  operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
  id INT,
  `name` STRING,
  score INT,
  PRIMARY KEY(id) NOT ENFORCED
) WITH (
  'connector' = 'tidb-cdc',
  'tikv.grpc.timeout_in_ms' = '20000',
  'pd-addresses' = '192.168.0.1:2379',
  'database-name' = 'tidb_source',
  'table-name' = 'student'
);

CREATE TEMPORARY TABLE students_print_sink (
  `db_name` STRING,
  `table_name` STRING,
  operation_ts TIMESTAMP_LTZ(3),
  id INT,
  `name` STRING,
  score INT
) WITH (
  'connector' = 'print',
  'limit' = '100',
  'logger'='true'
);

insert into students_print_sink
select * from students_tidb_source;

6、TiDB数据准备

运行作业前，先添加一条数据：

CREATE DATABASE  tidb_source;
use  tidb_source;
--创建student表
CREATE TABLE student(
  id int NOT NULL ,
  name varchar(100) NULL ,
  score int NULL ,
  primary key(id)
);

--插入数据
insert into student values(1,"lincloud",100);
select * from student;

7、验证测试

启动作业，测试