- 浏览: 79243 次
文章分类
最新评论
debezium关于cdc的使用(上)
博文原址:debezium关于cdc的使用(上)
简介
debezium是一个为了捕获数据变更(cdc)的开源的分布式平台。启动并指向数据库,当其他应用对此数据库执行inserts
、updates
、delete
操作时,此应用快速得到响应。debezium是持久化和快速响应的,因此你的应用可以快速响应且不会丢失任意一条事件。debezium记录是数据库表的行级别的变更事件。同时debezium是构建在kafka之上的,同时与kafka深度耦合,所以提供kafka connector来使用,debezium sink。支持的数据库有mysql、MongoDB、PostgreSQL、Oracle、SQL server。本篇以mysql作为数据源来实现功能,监听msyql的binlog,还需要修改。当前版本是0.9.5.Final,0.10版本正在开发中。
配置
本篇文章主要使用Embedding
形式监听事件,并同步更新到数据库。
下篇主要使用kafka connector来同步更新到数据库。
mysql需要如下开启binlog。但是如果使用的是debezium/mysql镜像,自动已经配置好了。
log-bin=mysql-bin #添加这一行就ok
binlog-format=ROW #选择row模式
server_id=1 #配置mysql replaction需要定义,不能和canal的slaveId重复
Tutorial
先来一个效果,主要是配置kafka connector来获取debezium事件记录。需要3个服务,zookeeper、kakfa和debezium connector。这里使用docker来启动的,所以需要先按照docker。
启动zookeeper
docker run -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/zookeeper:0.9
启动kafka
docker run -it --rm --name kafka -p 9092:9092 --link zookeeper:zookeeper debezium/kafka:0.9
启动mysql
docker run -it --rm --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw debezium/example-mysql:0.9
启动kafka connect
docker run -it --rm --name connect -p 8083:8083 -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses --link zookeeper:zookeeper --link kafka:kafka --link mysql:mysql debezium/connect:0.9
通过connect的http请求创建debezium connector
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{ "name": "inventory-connector", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "mysql", "database.port": "3306", "database.user": "debezium", "database.password": "dbz", "database.server.id": "184054", "database.server.name": "dbserver1", "database.whitelist": "inventory", "database.history.kafka.bootstrap.servers": "kafka:9092", "database.history.kafka.topic": "dbhistory.inventory" } }'
mysql客户端操作
通过invertory
数据库了的任一表的数据
创建监听可以查看debezium事件记录
docker run -it --name watcher --rm --link zookeeper:zookeeper --link kafka:kafka debezium/kafka:0.9 watch-topic -a -k dbserver1.inventory.customers
内嵌式
这里主要使用内嵌式的方式获取cdc事件而不需要使用kafka,直接消费debezium事件流。场景是在某一个mysql数据库里的table发生变更,把变更同步到另一mysql数据库。本次使用的是监听inventory
数据库并将数据同步到inventory_back
。
debezium配置
connector.class=io.debezium.connector.mysql.MySqlConnector
offset.storage=org.apache.kafka.connect.storage.FileOffsetBackingStore
offset.storage.file.filename=offset.dat
offset.flush.interval.ms=60000
name=debezium-kafka-source
database.hostname=localhost
database.port=3306
database.user=debezium
database.password=dbz
#database.dbname=inventory
database.whitelist=inventory
#database.whitelist=inventory,inventory_back
server.id=184054
database.server.name=dbserver1
#transforms=unwrap
#transforms.unwrap.type=io.debezium.transforms.UnwrapFromEnvelope
#transforms.unwrap.drop.tombstones=false
database.history=io.debezium.relational.history.FileDatabaseHistory
database.history.file.filename=dbhistory.dat
属性和convert配置
@Slf4j
@Configuration
public class DebeziumEmbeddedAutoConfiguration {
@Bean
public Properties embeddedProperties() {
Properties propConfig = new Properties();
try(InputStream propsInputStream = getClass().getClassLoader().getResourceAsStream("config.properties")) {
propConfig.load(propsInputStream);
} catch (IOException e) {
log.error("Couldn't load properties", e);
}
PropertyLoader.loadEnvironmentValues(propConfig);
return propConfig;
}
@Bean
public io.debezium.config.Configuration embeddedConfig(Properties embeddedProperties) {
return io.debezium.config.Configuration.from(embeddedProperties);
}
@Bean
public JsonConverter keyConverter(io.debezium.config.Configuration embeddedConfig) {
JsonConverter converter = new JsonConverter();
converter.configure(embeddedConfig.asMap(), true);
return converter;
}
@Bean
public JsonConverter valueConverter(io.debezium.config.Configuration embeddedConfig) {
JsonConverter converter = new JsonConverter();
converter.configure(embeddedConfig.asMap(), false);
return converter;
}
}
同步DDL和DML
这里主要是利用CommandLineRunner特性,启动debezium的EmbeddedEngine引擎,获取到cdc事件后由handleRecord
处理DDL和DML,需要去解析cdc的事件SourceRecord
的key和value。
@Slf4j
@Order(2)
@Component
public class DebeziumEmbeddedRunner implements CommandLineRunner {
@Autowired
private io.debezium.config.Configuration embeddedConfig;
@Autowired
private JdbcTemplate jdbcTemplate;
@Autowired
private NamedParameterJdbcTemplate namedTemplate;
@Autowired
private JsonConverter keyConverter;
@Autowired
private JsonConverter valueConverter;
@Override
public void run(String... args) throws Exception {
EmbeddedEngine engine = EmbeddedEngine.create()
.using(embeddedConfig)
.using(this.getClass().getClassLoader())
.using(Clock.SYSTEM)
.notifying(this::handleRecord)
.build();
ExecutorService executor = Executors.newSingleThreadExecutor();
executor.execute(engine);
shutdownHook(engine);
awaitTermination(executor);
}
/**
* For every record this method will be invoked.
*/
private void handleRecord(SourceRecord record) {
logRecord(record);
Struct payload = (Struct) record.value();
if (Objects.isNull(payload)) {
return;
}
String table = Optional.ofNullable(DebeziumRecordUtils.getRecordStructValue(payload, "source"))
.map(s->s.getString("table")).orElse(null);
// // 处理数据DML
Envelope.Operation operation = DebeziumRecordUtils.getOperation(payload);
if (Objects.nonNull(operation)) {
Struct key = (Struct) record.key();
handleDML(key, payload, table, operation);
return;
}
//
// // 处理结构DDL
String ddl = getDDL(payload);
if (StringUtils.isNotBlank(ddl)) {
handleDDL(ddl);
}
}
private String getDDL(Struct payload) {
String ddl = DebeziumRecordUtils.getDDL(payload);
if (StringUtils.isBlank(ddl)) {
return null;
}
String db = DebeziumRecordUtils.getDatabaseName(payload);
if (StringUtils.isBlank(db)) {
db = embeddedConfig.getString(MySqlConnectorConfig.DATABASE_WHITELIST);
}
ddl = ddl.replace(db + ".", "");
ddl = ddl.replace("`" + db + "`.", "");
return ddl;
}
/**
* 执行数据库ddl语句
*
* @param ddl
*/
private void handleDDL(String ddl) {
log.info("ddl语句 : {}", ddl);
try {
jdbcTemplate.execute(ddl);
} catch (Exception e) {
log.error("数据库操作DDL语句失败,", e);
}
}
/**
* 处理insert,update,delete等DML语句
*
* @param key 表主键修改事件结构
* @param payload 表正文响应
* @param table 表名
* @param operation DML操作类型
*/
private void handleDML(Struct key, Struct payload, String table, Envelope.Operation operation) {
AbstractDebeziumSqlProvider provider = DebeziumSqlProviderFactory.getProvider(operation);
if (Objects.isNull(provider)) {
log.error("没有找到sql处理器提供者.");
return;
}
String sql = provider.getSql(key, payload, table);
if (StringUtils.isBlank(sql)) {
log.error("找不到sql.");
return;
}
try {
log.info("dml语句 : {}", sql);
namedTemplate.update(sql, provider.getSqlParameterMap());
} catch (Exception e) {
log.error("数据库DML操作失败,", e);
}
}
/**
* 打印消息
*
* @param record
*/
private void logRecord(SourceRecord record) {
final byte[] payload = valueConverter.fromConnectData("dummy", record.valueSchema(), record.value());
final byte[] key = keyConverter.fromConnectData("dummy", record.keySchema(), record.key());
log.info("Publishing Topic --> {}", record.topic());
log.info("Key --> {}", new String(key));
log.info("Payload --> {}", new String(payload));
}
private void shutdownHook(EmbeddedEngine engine) {
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
log.info("Requesting embedded engine to shut down");
engine.stop();
}));
}
private void awaitTermination(ExecutorService executor) {
try {
while (!executor.awaitTermination(10L, TimeUnit.SECONDS)) {
log.info("Waiting another 10 seconds for the embedded engine to shut down");
}
} catch (InterruptedException e) {
Thread.interrupted();
}
}
}
provider和table字段解析器太多,这里就不在一一列出来了,如下图所示,支持mysql大部分字段类型。如果有需要的可以关注微信公众号或者邮件以及评论回复。
测试表结构
CREATE TABLE `demo` (
`id` int(10) NOT NULL AUTO_INCREMENT,
`bigint_id` bigint(20) NOT NULL,
`var_name` varchar(255) NOT NULL,
`ex_tinyint` tinyint(4) DEFAULT NULL,
`ex_char` char(255) DEFAULT NULL,
`ex_json` json DEFAULT NULL COMMENT '水电费',
`ex_text` text,
`ex_year` year(4) DEFAULT NULL,
`ex_time` time DEFAULT NULL,
`ex_date` date DEFAULT NULL,
`ex_datetime` datetime DEFAULT NULL,
`ex_timestamp` timestamp NULL DEFAULT NULL,
`ex_blob` blob,
`ex_tinyblob` tinyblob,
`ex_binary` binary(255) DEFAULT NULL,
`ex_double` double(10,4) DEFAULT NULL,
`ex_float` float(10,2) DEFAULT NULL,
`ex_decimal` decimal(10,2) DEFAULT NULL,
`ex_numeric` decimal(10,4) DEFAULT NULL,
`ex_real` double(10,4) DEFAULT NULL,
`ex_bit` bit(1) DEFAULT NULL,
`ex_enum` enum('123','@@','22','水电费') DEFAULT '123',
`ex_set` set('a','b','c','d') DEFAULT NULL,
`ex_geometry` geometry DEFAULT NULL,
`ex_point` point DEFAULT NULL,
`ex_linestring` linestring DEFAULT NULL,
`ex_polygon` polygon DEFAULT NULL,
`ex_geometrycollection` geometrycollection DEFAULT NULL,
`ex_multipoint` multipoint DEFAULT NULL,
PRIMARY KEY (`id`,`bigint_id`,`var_name`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;
结果
DDL事件
可以看出将数据库表的bigint_id
字段长度改为21,监听到事件后:执行了ddl语句,inventory_back
库中的demo
表的bigint_id
字段长度改为21了。
Publishing Topic --> dbserver1
2019-06-24 16:22:21.230 INFO 14995 --- [pool-1-thread-1] c.e.embedded.DebeziumEmbeddedRunner : Key --> {"schema":{"type":"struct","fields":[{"type":"string","optional":false,"field":"databaseName"}],"optional":false,"name":"io.debezium.connector.mysql.SchemaChangeKey"},"payload":{"databaseName":"inventory"}}
2019-06-24 16:22:21.230 INFO 14995 --- [pool-1-thread-1] c.e.embedded.DebeziumEmbeddedRunner : Payload --> {"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"string","optional":true,"field":"version"},{"type":"string","optional":true,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"server_id"},{"type":"int64","optional":false,"field":"ts_sec"},{"type":"string","optional":true,"field":"gtid"},{"type":"string","optional":false,"field":"file"},{"type":"int64","optional":false,"field":"pos"},{"type":"int32","optional":false,"field":"row"},{"type":"boolean","optional":true,"default":false,"field":"snapshot"},{"type":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"db"},{"type":"string","optional":true,"field":"table"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"databaseName"},{"type":"string","optional":false,"field":"ddl"}],"optional":false,"name":"io.debezium.connector.mysql.SchemaChangeValue"},"payload":{"source":{"version":"0.9.3.Final","connector":"mysql","name":"dbserver1","server_id":223344,"ts_sec":1561364540,"gtid":null,"file":"mysql-bin.000006","pos":22530,"row":0,"snapshot":false,"thread":null,"db":null,"table":null,"query":null},"databaseName":"inventory","ddl":"ALTER TABLE `inventory`.`demo` \nMODIFY COLUMN `bigint_id` bigint(21) NOT NULL AFTER `id`"}}
2019-06-24 16:22:21.230 ERROR 14995 --- [pool-1-thread-1] c.example.embedded.DebeziumRecordUtils : not find op field.
2019-06-24 16:22:21.231 INFO 14995 --- [pool-1-thread-1] c.e.embedded.DebeziumEmbeddedRunner : ddl语句 : ALTER TABLE `demo`
MODIFY COLUMN `bigint_id` bigint(21) NOT NULL AFTER `id`
DML的insert事件
在inventory
库中的demo
新增一条记录后有如下日志记录,能查看到topic,key,payload以及dml的insert语句。结果会把数据同步到inventory_back
库中的demo
。
2019-06-24 16:27:14.735 INFO 14995 --- [pool-1-thread-1] i.debezium.connector.mysql.BinlogReader : 1 records sent during previous 00:04:53.506, last recorded offset: {ts_sec=1561364834, file=mysql-bin.000006, pos=23002, row=1, server_id=223344, event=2}
2019-06-24 16:27:14.737 INFO 14995 --- [pool-1-thread-1] c.e.embedded.DebeziumEmbeddedRunner : Publishing Topic --> dbserver1.inventory.demo
2019-06-24 16:27:14.737 INFO 14995 --- [pool-1-thread-1] c.e.embedded.DebeziumEmbeddedRunner : Key --> {"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"bigint_id"},{"type":"string","optional":false,"field":"var_name"}],"optional":false,"name":"dbserver1.inventory.demo.Key"},"payload":{"id":2,"bigint_id":1,"var_name":"老王"}}
2019-06-24 16:27:14.738 INFO 14995 --- [pool-1-thread-1] c.e.embedded.DebeziumEmbeddedRunner : Payload --> {"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"bigint_id"},{"type":"string","optional":false,"field":"var_name"},{"type":"int16","optional":true,"field":"ex_tinyint"},{"type":"string","optional":true,"field":"ex_char"},{"type":"string","optional":true,"name":"io.debezium.data.Json","version":1,"field":"ex_json"},{"type":"string","optional":true,"field":"ex_text"},{"type":"int32","optional":true,"name":"io.debezium.time.Year","version":1,"field":"ex_year"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTime","version":1,"field":"ex_time"},{"type":"int32","optional":true,"name":"io.debezium.time.Date","version":1,"field":"ex_date"},{"type":"int64","optional":true,"name":"io.debezium.time.Timestamp","version":1,"field":"ex_datetime"},{"type":"string","optional":true,"name":"io.debezium.time.ZonedTimestamp","version":1,"field":"ex_timestamp"},{"type":"bytes","optional":true,"field":"ex_blob"},{"type":"bytes","optional":true,"field":"ex_tinyblob"},{"type":"bytes","optional":true,"field":"ex_binary"},{"type":"double","optional":true,"field":"ex_double"},{"type":"double","optional":true,"field":"ex_float"},{"type":"bytes","optional":true,"name":"org.apache.kafka.connect.data.Decimal","version":1,"parameters":{"scale":"2","connect.decimal.precision":"10"},"field":"ex_decimal"},{"type":"bytes","optional":true,"name":"org.apache.kafka.connect.data.Decimal","version":1,"parameters":{"scale":"4","connect.decimal.precision":"10"},"field":"ex_numeric"},{"type":"double","optional":true,"field":"ex_real"},{"type":"boolean","optional":true,"field":"ex_bit"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"123,22"},"default":"123","field":"ex_enum"},{"type":"string","optional":true,"name":"io.debezium.data.EnumSet","version":1,"parameters":{"allowed":"a,b,c,d"},"field":"ex_set"},{"type":"struct","fields":[{"type":"bytes","optional":false,"field":"wkb"},{"type":"int32","optional":true,"field":"srid"}],"optional":true,"name":"io.debezium.data.geometry.Geometry","version":1,"doc":"Geometry","field":"ex_geometry"},{"type":"struct","fields":[{"type":"double","optional":false,"field":"x"},{"type":"double","optional":false,"field":"y"},{"type":"bytes","optional":true,"field":"wkb"},{"type":"int32","optional":true,"field":"srid"}],"optional":true,"name":"io.debezium.data.geometry.Point","version":1,"doc":"Geometry (POINT)","field":"ex_point"},{"type":"struct","fields":[{"type":"bytes","optional":false,"field":"wkb"},{"type":"int32","optional":true,"field":"srid"}],"optional":true,"name":"io.debezium.data.geometry.Geometry","version":1,"doc":"Geometry","field":"ex_linestring"},{"type":"struct","fields":[{"type":"bytes","optional":false,"field":"wkb"},{"type":"int32","optional":true,"field":"srid"}],"optional":true,"name":"io.debezium.data.geometry.Geometry","version":1,"doc":"Geometry","field":"ex_polygon"},{"type":"struct","fields":[{"type":"bytes","optional":false,"field":"wkb"},{"type":"int32","optional":true,"field":"srid"}],"optional":true,"name":"io.debezium.data.geometry.Geometry","version":1,"doc":"Geometry","field":"ex_geometrycollection"},{"type":"struct","fields":[{"type":"bytes","optional":false,"field":"wkb"},{"type":"int32","optional":true,"field":"srid"}],"optional":true,"name":"io.debezium.data.geometry.Geometry","version":1,"doc":"Geometry","field":"ex_multipoint"}],"optional":true,"name":"dbserver1.inventory.demo.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"bigint_id"},{"type":"string","optional":false,"field":"var_name"},{"type":"int16","optional":true,"field":"ex_tinyint"},{"type":"string","optional":true,"field":"ex_char"},{"type":"string","optional":true,"name":"io.debezium.data.Json","version":1,"field":"ex_json"},{"type":"string","optional":true,"field":"ex_text"},{"type":"int32","optional":true,"name":"io.debezium.time.Year","version":1,"field":"ex_year"},{"type":"int64","optional":true,"name":"io.debezium.time.MicroTime","version":1,"field":"ex_time"},{"type":"int32","optional":true,"name":"io.debezium.time.Date","version":1,"field":"ex_date"},{"type":"int64","optional":true,"name":"io.debezium.time.Timestamp","version":1,"field":"ex_datetime"},{"type":"string","optional":true,"name":"io.debezium.time.ZonedTimestamp","version":1,"field":"ex_timestamp"},{"type":"bytes","optional":true,"field":"ex_blob"},{"type":"bytes","optional":true,"field":"ex_tinyblob"},{"type":"bytes","optional":true,"field":"ex_binary"},{"type":"double","optional":true,"field":"ex_double"},{"type":"double","optional":true,"field":"ex_float"},{"type":"bytes","optional":true,"name":"org.apache.kafka.connect.data.Decimal","version":1,"parameters":{"scale":"2","connect.decimal.precision":"10"},"field":"ex_decimal"},{"type":"bytes","optional":true,"name":"org.apache.kafka.connect.data.Decimal","version":1,"parameters":{"scale":"4","connect.decimal.precision":"10"},"field":"ex_numeric"},{"type":"double","optional":true,"field":"ex_real"},{"type":"boolean","optional":true,"field":"ex_bit"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"123,22"},"default":"123","field":"ex_enum"},{"type":"string","optional":true,"name":"io.debezium.data.EnumSet","version":1,"parameters":{"allowed":"a,b,c,d"},"field":"ex_set"},{"type":"struct","fields":[{"type":"bytes","optional":false,"field":"wkb"},{"type":"int32","optional":true,"field":"srid"}],"optional":true,"name":"io.debezium.data.geometry.Geometry","version":1,"doc":"Geometry","field":"ex_geometry"},{"type":"struct","fields":[{"type":"double","optional":false,"field":"x"},{"type":"double","optional":false,"field":"y"},{"type":"bytes","optional":true,"field":"wkb"},{"type":"int32","optional":true,"field":"srid"}],"optional":true,"name":"io.debezium.data.geometry.Point","version":1,"doc":"Geometry (POINT)","field":"ex_point"},{"type":"struct","fields":[{"type":"bytes","optional":false,"field":"wkb"},{"type":"int32","optional":true,"field":"srid"}],"optional":true,"name":"io.debezium.data.geometry.Geometry","version":1,"doc":"Geometry","field":"ex_linestring"},{"type":"struct","fields":[{"type":"bytes","optional":false,"field":"wkb"},{"type":"int32","optional":true,"field":"srid"}],"optional":true,"name":"io.debezium.data.geometry.Geometry","version":1,"doc":"Geometry","field":"ex_polygon"},{"type":"struct","fields":[{"type":"bytes","optional":false,"field":"wkb"},{"type":"int32","optional":true,"field":"srid"}],"optional":true,"name":"io.debezium.data.geometry.Geometry","version":1,"doc":"Geometry","field":"ex_geometrycollection"},{"type":"struct","fields":[{"type":"bytes","optional":false,"field":"wkb"},{"type":"int32","optional":true,"field":"srid"}],"optional":true,"name":"io.debezium.data.geometry.Geometry","version":1,"doc":"Geometry","field":"ex_multipoint"}],"optional":true,"name":"dbserver1.inventory.demo.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":true,"field":"version"},{"type":"string","optional":true,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"server_id"},{"type":"int64","optional":false,"field":"ts_sec"},{"type":"string","optional":true,"field":"gtid"},{"type":"string","optional":false,"field":"file"},{"type":"int64","optional":false,"field":"pos"},{"type":"int32","optional":false,"field":"row"},{"type":"boolean","optional":true,"default":false,"field":"snapshot"},{"type":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"db"},{"type":"string","optional":true,"field":"table"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"}],"optional":false,"name":"dbserver1.inventory.demo.Envelope"},"payload":{"before":null,"after":{"id":2,"bigint_id":1,"var_name":"老王","ex_tinyint":1,"ex_char":"a","ex_json":"{\"abc\":123}","ex_text":"ert","ex_year":2019,"ex_time":59224000000,"ex_date":null,"ex_datetime":null,"ex_timestamp":null,"ex_blob":null,"ex_tinyblob":null,"ex_binary":null,"ex_double":null,"ex_float":null,"ex_decimal":null,"ex_numeric":null,"ex_real":null,"ex_bit":null,"ex_enum":"123","ex_set":null,"ex_geometry":null,"ex_point":null,"ex_linestring":null,"ex_polygon":null,"ex_geometrycollection":null,"ex_multipoint":null},"source":{"version":"0.9.3.Final","connector":"mysql","name":"dbserver1","server_id":223344,"ts_sec":1561364834,"gtid":null,"file":"mysql-bin.000006","pos":23194,"row":0,"snapshot":false,"thread":9,"db":"inventory","table":"demo","query":null},"op":"c","ts_ms":1561364834477}}
2019-06-24 16:27:14.738 INFO 14995 --- [pool-1-thread-1] c.e.embedded.DebeziumEmbeddedRunner : dml语句 : insert into demo values (:id,:bigint_id,:var_name,:ex_tinyint,:ex_char,:ex_json,:ex_text,:ex_year,:ex_time,:ex_date,:ex_datetime,:ex_timestamp,:ex_blob,:ex_tinyblob,:ex_binary,:ex_double,:ex_float,:ex_decimal,:ex_numeric,:ex_real,:ex_bit,:ex_enum,:ex_set,:ex_geometry,:ex_point,:ex_linestring,:ex_polygon,:ex_geometrycollection,:ex_multipoint)
DML的update事件
在inventory
库中的demo
修改刚刚新增的记录后有如下日志记录,能查看到topic,key,payload以及先delete再insert语句。结果会把数据同步到inventory_back
库中的demo
。
DML的delete事件
在inventory
库中的demo
修改刚刚修改的记录给删除掉后有如下日志记录,能查看到topic,key,payload以及先delete语句。结果会把数据同步到inventory_back
库中的demo
将其删掉。这里有2个事件,第二条事件是一种标致,这里不处理。
日志:
参考
欢迎
关注了解最新动态更新
转载于:https://my.oschina.net/damonchow/blog/3065797
相关推荐
参考在您的机器/集群上设置Apache Kafka(使用Kafka Connect) 从安装Debezium PostgreSQL连接器运行Apache Kafka和Kafka Connect 在PostgreSQL中创建表transactions和customers (SQL文件) 使用请求主体向您的...
通过以上步骤,DB2数据库中的表结构变更可以在不影响数据完整性和一致性的情况下,顺利地与CDC功能配合使用。这个过程需要谨慎操作,以避免可能出现的数据丢失或不一致问题。在实际操作中,应确保遵循最佳实践,并在...
本话题将详细讲解如何利用Flink的SQL Server Change Data Capture (CDC) 连接器版本2.3.0,将SQL Server中的数据实时同步到MySQL数据库。 首先,让我们了解什么是CDC。CDC是一种数据库技术,它能够捕获数据库中的...
sql server2008 cdc 数据实时同步到kafka,Debezium是捕获数据实时动态变化的开源的分布式同步平台。能实时捕获到数据源(Mysql、Mongo、PostgreSql)的:新增(inserts)、更新(updates)、删除(deletes)操作,实时...
Flink CDC 实时数据集成方案 ...2. Canal / Debezium:Flink CDC 可以与 Canal / Debezium 一起使用,提供实时数据集成方案。 3. DataX / Sqoop:Flink CDC 可以与 DataX / Sqoop 一起使用,提供实时数据集成方案。
在MySQL中,可以使用开源工具如Debezium、Binlog Server或者基于binlog的解析来实现CDC。 **二、PyFlink与MySQL CDC的结合** 1. **设置MySQL CDC环境** - 首先,你需要启用MySQL的binlog日志,并设置合适的格式...
以及基于日志的CDC,如Debezium、Canal和Flink-CDC,后者通过解析数据库的日志来实时获取变更。Flink-CDC因其全增量一体化同步、分布式架构和强大的数据加工能力而备受青睐。 **二、为何使用CDC及适用场景** 随着...
NiFi 集成 Debezium 接收 MySQL CDC
Flink CDC连接器是Apache Flink的一组源连接器,使用更改数据捕获(CDC)从不同的数据库中提取更改。 Flink CDC连接器将Debezium集成为引擎来捕获数据更改。 因此,它可以充分利用Debezium的功能。 进一步了解什么是...
Debezium是一个开源项目,为捕获数据更改(change data capture,CDC)提供了一个低延迟的流式处理平台。你可以安装并且配置Debezium去监控你的数据库,然后你的应用就可以消费对数据库的每一个行级别(row-level)的更改...
而Debezium是一款强大的分布式平台,专注于数据复制和变更数据捕获(CDC,Change Data Capture),它允许应用程序实时跟踪数据库中的数据变化。在本案例中,"debezium-debezium-connector-mongodb-2.0.0.Final" 是...
在这个场景下,我们将深入探讨如何使用Flink CDC来监测MySQL数据库,并且实现自定义反序列化,以及如何通过Flink API和Flink SQL两种方式进行数据处理。 首先,让我们理解什么是CDC。CDC是一种数据库复制技术,它...
用于debezium CDC框架的重力适配器。 故障排除 快照不起作用 Debezium具有实现称为快照的初始负载的实现,但是适配器启动时以某种方式快照不起作用。 此问题的根本原因是该主题已存在或使用者组的偏移量不为零。 这...
【描述】:“Debezium-k8s是Debezium项目在Kubernetes环境中的实现,它结合了数据库变更数据捕获(CDC)的能力以及对MySQL数据库的支持。这允许在分布式系统中实时跟踪和处理数据库的变更事件,为微服务架构提供了...
Apache Kafka是一个分布式流处理平台,而Debezium是一个开源的变更数据捕获(Change Data Capture, CDC)工具,能够捕获数据库变更事件并发布到Kafka上。本文由Gunnar Morling撰写,他是一名开源软件工程师,同时也...
例如:一个订单系统刚刚开始只需要写入数据库即可完成业务使用。某天BI团队期望对数据库做全文索引,于是我们同时要写多一份数据到ES中,改造后一段时间,又有需求需要写入到Redis缓存中。很明显这种模式是不可持续...
**ChangeDataCapturePOC** 是一个专注于学习和研究如何使用 **Debezium** 和 **Kafka** 实现 **Change Data Capture (CDC)** 的项目。在IT领域,CDC是一种技术,它能够跟踪和捕获数据库中的数据变化,这些变化可以...
对于MySQL,Flink使用Debezium作为其CDC连接器。Debezium是一个分布式平台,可以捕获和传递数据库的变更事件。在MySQL中,它是通过binlog(二进制日志)来跟踪和捕获数据变更的。通过配置Flink与Debezium,开发者...
3. **数据源集成**:讨论Flink与各种数据源的集成,如MySQL、PostgreSQL、Oracle等,以及如何配置和使用Debezium等连接器来实现CDC。 4. **数据转换与处理**:介绍如何在Flink中对捕获的数据进行清洗、转换和聚合,...
该 Demo 展示了如何使用 Flink CDC 将海量数据从数据库实时同步到数据仓库中。该 Demo 表明了 Flink CDC 技术的高效性和可靠性。 Flink CDC 的优点 Flink CDC 技术有很多优点,例如高效的数据同步、实时的数据分析...