十分钟掌握 Flink CDC，实现Mysql数据增量备份到Clickhouse [纯干货，建议收藏]

Clickhouse的优点.

1. 真正的面向列的 DBMS

ClickHouse 是一个 DBMS，而不是一个单一的数据库。它允许在运行时创建表和数据库、加载数据和运行

查询，而无需重新配置和重新启动服务器。

2. 数据压缩

一些面向列的 DBMS（InfiniDB CE 和 MonetDB）不使用数据压缩。但是，数据压缩确实提高了性能。

3. 磁盘存储的数据

4. 在多个服务器上分布式处理

5. SQL支持

6. 数据不仅按列存储，而且由矢量 - 列的部分进行处理，这使开发者能够实现高 CPU 性能

Clickhouse的缺点

1. 没有完整的事务支持，

2. 缺少完整的Update/Delete操作，缺少高频率、低延迟的修改或删除已存在数据的能力，仅能用于批量删

除或修改数据

3. 聚合结果必须小于一台机器的内存大小：

4. 不适合key-value存储，

什么时候不可以用Clickhouse?

1. 事物性工作（OLTP）

2. 高并发的键值访问

3. Blob或者文档存储

4. 超标准化的数据

Flink CDC

=========

Flink cdc connector 消费 Debezium 里的数据，经过处理再sink出来，这个流程还是相对比较简单的

首先创建 Source 和 Sink（对应的依赖引用，在文末）

SourceFunction sourceFunction = MySQLSource.builder()

.hostname(“localhost”)

.port(3306)

.databaseList(“test”)

.username(“flinkcdc”)

.password(“dafei1288”)

.deserializer(new JsonDebeziumDeserializationSchema())

.build();

// 添加 source

env.addSource(sourceFunction)

// 添加 sink

.addSink(new ClickhouseSink());

这里用到的JsonDebeziumDeserializationSchema，是我们自定义的一个序列化类，用于将Debezium输出的数据，序列化

public static class JsonDebeziumDeserializationSchema implements DebeziumDeserializationSchema {

@Override

public void deserialize(SourceRecord sourceRecord, Collector collector) throws Exception {

Gson jsstr = new Gson();

HashMap<String, Object> hs = new HashMap<>();

String topic = sourceRecord.topic();

String[] split = topic.split(“[.]”);

String database = split[1];

String table = split[2];

hs.put(“database”,database);

hs.put(“table”,table);

//获取操作类型

Envelope.Operation operation = Envelope.operationFor(sourceRecord);

//获取数据本身

Struct struct = (Struct)sourceRecord.value();

Struct after = struct.getStruct(“after”);

if (after != null) {

Schema schema = after.schema();

HashMap<String, Object> afhs = new HashMap<>();

for (Field field : schema.fields()) {

afhs.put(field.name(), after.get(field.name()));

}

hs.put(“data”,afhs);

}

String type = operation.toString().toLowerCase();

if (“create”.equals(type)) {

type = “insert”;

}

hs.put(“type”,type);

collector.collect(jsstr.toJson(hs));

}

@Override

public TypeInformation getProducedType() {

return BasicTypeInfo.STRING_TYPE_INFO;

}

}

这里是将数据序列化成如下Json格式

{“database”:“test”,“data”:{“name”:“jacky”,“description”:“fffff”,“id”:8},“type”:“insert”,“table”:“test_cdc”}

接下来就是要创建Sink，将数据变化存入Clickhouse中，这里我们仅以insert为例

public static class ClickhouseSink extends RichSinkFunction{

Connection connection;

PreparedStatement pstmt;

private Connection getConnection() {

Connection conn = null;

try {

Class.forName(“ru.yandex.clickhouse.ClickHouseDriver”);

String url = “jdbc:clickhouse://localhost:8123/default”;

conn = DriverManager.getConnection(url,“default”,“dafei1288”);

} catch (Exception e) {

e.printStackTrace();

}

return conn;

}

@Override

public void open(Configuration parameters) throws Exception {

super.open(parameters);

connection = getConnection();

String sql = “insert into sink_ch_test(id,name,description) values (?,?,?)”;

pstmt = connection.prepareStatement(sql);

}

// 每条记录插入时调用一次

public void invoke(String value, Context context) throws Exception {

//{“database”:“test”,“data”:{“name”:“jacky”,“description”:“fffff”,“id”:8},“type”:“insert”,“table”:“test_cdc”}

Gson t = new Gson();

HashMap<String,Object> hs = t.fromJson(value,HashMap.class);

String database = (String)hs.get(“database”);

String table = (String)hs.get(“table”);

String type = (String)hs.get(“type”);

if(“test”.equals(database) && “test_cdc”.equals(table)){

if(“insert”.equals(type)){

System.out.println("insert => "+value);

LinkedTreeMap<String,Object> data = (LinkedTreeMap<String,Object>)hs.get(“data”);

String name = (String)data.get(“name”);

String description = (String)data.get(“description”);

Double id = (Double)data.get(“id”);

// 未前面的占位符赋值

pstmt.setInt(1, id.intValue());

pstmt.setString(2, name);

pstmt.setString(3, description);

pstmt.executeUpdate();

}

}

}

@Override

public void close() throws Exception {

super.close();

if(pstmt != null) {

pstmt.close();

}

if(connection != null) {

connection.close();

}

}

}

完整代码案例：

package name.lijiaqi.cdc;

import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;

import com.google.gson.Gson;

import com.google.gson.internal.LinkedTreeMap;

import io.debezium.data.Envelope;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;

import org.apache.flink.api.common.typeinfo.TypeInformation;

import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;

import org.apache.flink.util.Collector;

import org.apache.kafka.connect.source.SourceRecord;

import org.apache.kafka.connect.data.Field;

import org.apache.kafka.connect.data.Schema;

import org.apache.kafka.connect.data.Struct;

import java.sql.Connection;

import java.sql.DriverManager;

import java.sql.PreparedStatement;

import java.util.HashMap;

public class MySqlBinlogSourceExample {

public static void main(String[] args) throws Exception {

SourceFunction sourceFunction = MySQLSource.builder()

.hostname(“localhost”)

.port(3306)

.databaseList(“test”)

.username(“flinkcdc”)

.password(“dafei1288”)

.deserializer(new JsonDebeziumDeserializationSchema())

.build();

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 添加 source

env.addSource(sourceFunction)

// 添加 sink

.addSink(new ClickhouseSink());

env.execute(“mysql2clickhouse”);

}

// 将cdc数据反序列化

public static class JsonDebeziumDeserializationSchema implements DebeziumDeserializationSchema {

@Override

public void deserialize(SourceRecord sourceRecord, Collector collector) throws Exception {

Gson jsstr = new Gson();

HashMap<String, Object> hs = new HashMap<>();

String topic = sourceRecord.topic();

String[] split = topic.split(“[.]”);

String database = split[1];

String table = split[2];

hs.put(“database”,database);

hs.put(“table”,table);

//获取操作类型

Envelope.Operation operation = Envelope.operationFor(sourceRecord);

//获取数据本身

Struct struct = (Struct)sourceRecord.value();

Struct after = struct.getStruct(“after”);

if (after != null) {

Schema schema = after.schema();

HashMap<String, Object> afhs = new HashMap<>();

for (Field field : schema.fields()) {

afhs.put(field.name(), after.get(field.name()));

}

hs.put(“data”,afhs);

}

String type = operation.toString().toLowerCase();

if (“create”.equals(type)) {

type = “insert”;

}

hs.put(“type”,type);

collector.collect(jsstr.toJson(hs));

}

@Override

public TypeInformation getProducedType() {

return BasicTypeInfo.STRING_TYPE_INFO;

}

}

public static class ClickhouseSink extends RichSinkFunction{

Connection connection;

PreparedStatement pstmt;

private Connection getConnection() {

Connection conn = null;

try {

Class.forName(“ru.yandex.clickhouse.ClickHouseDriver”);

String url = “jdbc:clickhouse://localhost:8123/default”;

conn = DriverManager.getConnection(url,“default”,“dafei1288”);

} catch (Exception e) {

e.printStackTrace();

}

return conn;

}

@Override

public void open(Configuration parameters) throws Exception {

super.open(parameters);

connection = getConnection();

String sql = “insert into sink_ch_test(id,name,description) values (?,?,?)”;

pstmt = connection.prepareStatement(sql);

}

// 每条记录插入时调用一次

public void invoke(String value, Context context) throws Exception {

//{“database”:“test”,“data”:{“name”:“jacky”,“description”:“fffff”,“id”:8},“type”:“insert”,“table”:“test_cdc”}

Gson t = new Gson();

HashMap<String,Object> hs = t.fromJson(value,HashMap.class);

String database = (String)hs.get(“database”);

String table = (String)hs.get(“table”);

String type = (String)hs.get(“type”);

if(“test”.equals(database) && “test_cdc”.equals(table)){

自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、OPPO等大厂，18年进入阿里一直到现在。

深知大多数Java工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年Java开发全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友，同时减轻大家的负担。

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上Java开发知识点，真正体系化！

由于文件比较大，这里只是将部分目录截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且会持续更新！

如果你觉得这些内容对你有帮助，可以扫码获取！！（备注Java获取）

最后

本人也收藏了一份Java面试核心知识点来应付面试，借着这次机会可以送给我的读者朋友们

目录：

Java面试核心知识点

一共有30个专题，足够读者朋友们应付面试啦，也节省朋友们去到处搜刮资料自己整理的时间！

Java面试核心知识点

已经有读者朋友靠着这一份Java面试知识点指导拿到不错的offer了

《一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码》，点击传送门即可获取！
以上Java开发知识点，真正体系化！**

由于文件比较大，这里只是将部分目录截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且会持续更新！

如果你觉得这些内容对你有帮助，可以扫码获取！！（备注Java获取）

最后

本人也收藏了一份Java面试核心知识点来应付面试，借着这次机会可以送给我的读者朋友们

目录：

[外链图片转存中…(img-lki2bJFN-1711904063888)]

Java面试核心知识点

一共有30个专题，足够读者朋友们应付面试啦，也节省朋友们去到处搜刮资料自己整理的时间！

[外链图片转存中…(img-YzJSumVH-1711904063888)]

Java面试核心知识点

已经有读者朋友靠着这一份Java面试知识点指导拿到不错的offer了

[外链图片转存中…(img-Ez0BgOW4-1711904063889)]

《一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码》，点击传送门即可获取！