Clickhouse的优点.
- 真正的面向列的 DBMS
ClickHouse 是一个 DBMS，而不是一个单一的数据库。它允许在运行时创建表和数据库、加载数据和运行
查询，而无需重新配置和重新启动服务器。
- 数据压缩
一些面向列的 DBMS（InfiniDB CE 和 MonetDB）不使用数据压缩。但是，数据压缩确实提高了性能。
-
磁盘存储的数据
-
在多个服务器上分布式处理
-
SQL支持
-
数据不仅按列存储，而且由矢量 - 列的部分进行处理，这使开发者能够实现高 CPU 性能
Clickhouse的缺点
-
没有完整的事务支持，
-
缺少完整的Update/Delete操作，缺少高频率、低延迟的修改或删除已存在数据的能力，仅能用于批量删
除或修改数据
-
聚合结果必须小于一台机器的内存大小：
-
不适合key-value存储，
什么时候不可以用Clickhouse?
-
事物性工作（OLTP）
-
高并发的键值访问
-
Blob或者文档存储
-
超标准化的数据
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高分面试指南”，里面涵盖了25个分类的面试题以及详细的解析：JavaOOP、Java集合/泛型、Java中的IO与NIO、Java反射、Java序列化、Java注解、多线程&并发、JVM、Mysql、Redis、Memcached、MongoDB、Spring、Spring Boot、Spring Cloud、RabbitMQ、Dubbo 、MyBatis 、ZooKeeper 、数据结构、算法、Elasticsearch 、Kafka 、微服务、Linux。
这不，马上就要到招聘季了，很多朋友又开始准备“金三银四”的春招啦，那我想这份“java高分面试指南”应该起到不小的作用，所以今天想给大家分享一下。
请注意：关于这份“java高分面试指南”，每一个方向专题（25个）的题目这里几乎都会列举，在不看答案的情况下，大家可以自行测试一下水平 且由于篇幅原因，这边无法展示所有完整的答案解析
《一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码》，点击传送门即可获取！
dnimg.cn/images/e5c14a7895254671a72faed303032d36.jpg" alt=“img” style=“zoom: 33%;” />
最后
一次偶然，从朋友那里得到一份“java高分面试指南”，里面涵盖了25个分类的面试题以及详细的解析：JavaOOP、Java集合/泛型、Java中的IO与NIO、Java反射、Java序列化、Java注解、多线程&并发、JVM、Mysql、Redis、Memcached、MongoDB、Spring、Spring Boot、Spring Cloud、RabbitMQ、Dubbo 、MyBatis 、ZooKeeper 、数据结构、算法、Elasticsearch 、Kafka 、微服务、Linux。
这不，马上就要到招聘季了，很多朋友又开始准备“金三银四”的春招啦，那我想这份“java高分面试指南”应该起到不小的作用，所以今天想给大家分享一下。
[外链图片转存中…(img-FkTyyqdI-1712406568279)]文章来源：https://www.toymoban.com/news/detail-850954.html
请注意：关于这份“java高分面试指南”，每一个方向专题（25个）的题目这里几乎都会列举，在不看答案的情况下，大家可以自行测试一下水平 且由于篇幅原因，这边无法展示所有完整的答案解析
《一线大厂Java面试题解析+核心总结学习笔记+最新讲解视频+实战项目源码》，点击传送门即可获取！文章来源地址https://www.toymoban.com/news/detail-850954.html
到了这里，关于十分钟掌握 Flink CDC，实现Mysql数据增量备份到Clickhouse [纯干货，建议收藏]的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
