SQLServer CDC数据通过Kafka connect实时同步至分析型数据库 AnalyticDB For PostgreSQL及OSS

简介: 本文主要介绍如何通过消息对接, kafkakafka-connect数据平台以及相关插件将数据同步到分析型数据库 AnalyticDB PostgreSQL
+关注继续查看

背景

SQLServer为实时更新数据同步提供了CDC机制,类似于Mysql的binlog,将数据更新操作维护到一张CDC表中。
开启cdc的源表在插入INSERT、更新UPDATE和删除DELETE活动时会插入数据到日志表中。cdc通过捕获进程将变更数据捕获到变更表中,通过cdc提供的查询函数,可以捕获这部分数据。

CDC的使用条件

1.SQL server 2008及以上的企业版、开发版和评估版;
2.需要开启代理服务(作业)。
3.CDC需要业务库之外的额外的磁盘空间。
4.CDC的表需要主键或者唯一主键。
image
图1:Sqlserver CDC原理

ADB4PG Sink使用条件

  1. 需要提前使用建表语句,在ADB4PG端建表,系统不会自动创建(如果有需要可以加这部分功能)
  2. 每张表需要有主键或唯一主键
  3. 当前支持的数据格式:INTEGER,BIGINT,SMALLINT,NUMERIC,DECIMAL,REAL,DOUBLEPERICISION,BOOLEAN,DATE,TIMESTAMP,VARCHAR

环境准备

SQLServer环境准备

  1. 已有自建SQLServer或云上RDS实例(示例使用云上RDS SQLServer实例)
  2. 已有windows环境,并安装SSMS(SQL Server Management Studio),部分命令需要在SSMS执行

SQLServer环境建表

-- 创建源表
create database connect
GO
use connect
GO  

create table t1
(
    a int NOT NULL PRIMARY KEY,
    b BIGINT,
    c SMALLINT,
    d REAL,
    e FLOAT,
    f DATETIME,
    g VARCHAR
);



-- 开启db级的cdc
exec sp_rds_cdc_enable_db

-- 验证数据库是否开启cdc成功
select * from sys.databases where is_cdc_enabled = 1

-- 对源表开启cdc
exec sp_cdc_enable_table @source_schema='dbo', @source_name='t1', @role_name=null;

ADB4PG端创建目标表

CREATE DATABASE connect;

create table t1
(
    a int NOT NULL PRIMARY KEY,
    b BIGINT,
    c SMALLINT,
    d REAL,
    e FLOAT,
    f TIMESTAMP,
    g VARCHAR
);

Kafka环境准备

安装Kafka Server

1. 下载kafka安装包,并解压

SQL Server Source Connect目前只支持2.1.0及以上版本的Kafka Connect,故需要安装高版本kafka,实例使用kfakf-2.11-2.1.0。 http://kafka.apache.org/downloads?spm=a2c4g.11186623.2.19.7dd34587dwy89h#2.1.0

2. 编辑$KAFKA_HOME/config/server.properties

修改以下参数

...
## 为每台broker配置一个唯一的id号
broker.id=0

...

## log存储地址
log.dirs=/home/gaia/kafka_2.11-2.1.0/logs

## kafka集群使用的zk地址
zookeeper.connect=zk1:2181,zk2:2181,zk3:2181
...
3. 启动kafka server
bin/kafka-server-start.sh config/server.properties

安装Kafka Connect

1. 修改kafka connect配置文件

修改$KAFKA_HOME/config/connect-distributed.properties

## kafka server地址
bootstrap.servers=broker1:9092,broker2:9092,broker3:9092

## 为kafka connector选定一个消费group id
group.id=

## 安装插件的地址,每次kafka connector启动时会动态加载改路径下的jar包,可以将每个插件单独放到一个子路径
plugin.path=

安装需要的kafka-connect插件

1. 将插件jar包放在我们在前面已经配置过的配置的plugin.path路径

sqlserver-source-connector

https://repo1.maven.org/maven2/io/debezium/debezium-connector-sqlserver/?spm=a2c4g.11186623.2.18.7dd34587dwy89h

oss-sink-connector, 需要使用代码自行编译,注意在pom修改依赖的kafka及scala版本号

https://github.com/aliyun/kafka-connect-oss

adb4pg-jdbc-sink-connector,需要下载以下jar包及对应ADB For PG的JDBC驱动
https://yq.aliyun.com/attachment/download/?spm=a2c4e.11153940.0.0.70ed10daVH6ZQO&id=7282

2. 编辑配置文件
# CDC connector的配置文件 sqlserver-cdc-source.json
▽
{
    "name": "sqlserver-cdc-source",
    "config": {
        "connector.class" : "io.debezium.connector.sqlserver.SqlServerConnector",
        "tasks.max" : "1",
        "database.server.name" : "server1",
        "database.hostname" : "database hostname",
        "database.port" : "1433",
        "database.user" : "xxxx",
        "database.password" : "xxxxxx",
        "database.dbname" : "connect",
        "schemas.enable" : "false",
        "mode":"incrementing",
        "incrementing.column.name":"a",
        "database.history.kafka.bootstrap.servers" : "kafka-broker:9092",
        "database.history.kafka.topic": "server1.dbo.t1",
        "value.converter.schemas.enable":"false",
        "value.converter":"org.apache.kafka.connect.json.JsonConverter"
    }
}
# oss sink的配置文件 oss-sink.json
{
     "name":"oss-sink",
     "config": {
        "name":"oss-sink",
        "topics":"server1.dbo.testdata",
        "connector.class":"com.aliyun.oss.connect.kafka.OSSSinkConnector",
        "format.class":"com.aliyun.oss.connect.kafka.format.json.JsonFormat",
        "flush.size":"1",
        "tasks.max":"4",
        "storage.class":"com.aliyun.oss.connect.kafka.storage.OSSStorage",
        "partitioner.class":"io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
        "timestamp.extractor":"Record",
        "oss.bucket":"traffic-csv",
        "partition.duration.ms":"10000",
        "path.format":"YYYY-MM-dd-HH",
        "locale":"US",
        "timezone":"Asia/Shanghai",
        "rotate.interval.ms":"30000"
        }
}

有关oss sinker更详尽的配置,见文档 https://github.com/aliyun/kafka-connect-oss

## adb4pg-jdbc-sink配置文件
{
     "name":"adb4pg-jdbc-sink",
     "config": {
        "name":"adb4pg-jdbc-sink",
        "topics":"server1.dbo.t1",
        "connector.class":"io.confluent.connect.jdbc.Adb4PgSinkConnector",
        "connection.url":"jdbc:postgresql://gp-8vb8xi62lohhh2777o.gpdb.zhangbei.rds.aliyuncs.com:3432/connect",
        "connection.user":"xxx",
        "connection.password":"xxxxxx",
        "col.names":"a,b,c,d,e,f,g",
        "col.types":"integer,bigint,smallint,real,doublepericision,timestamp,varchar",
        "pk.fields":"a",
        "target.tablename":"t1",
        "tasks.max":"1",
        "auto.create":"false",
        "table.name.format":"t1",
        "batch.size":"1"
        }
}

由于OSS sinker使用了hdfs封装的FileSystem,需要将OSS相关的信息维护到$KAFKA_HOME/config/core-site.xml文件中

<configuration>
    <property>
        <name>fs.oss.endpoint</name>
        <value>xxxxxxx</value>
    </property>
    <property>
        <name>fs.oss.accessKeyId</name>
        <value>xxxxxxx</value>
    </property>
    <property>
        <name>fs.oss.accessKeySecret</name>
        <value>xxxxxxx</value>
    </property>
    <property>
        <name>fs.oss.impl</name>
        <value>org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem</value>
    </property>
    <property>
        <name>fs.oss.buffer.dir</name>
        <value>/tmp/oss</value>
    </property>
    <property>
        <name>fs.oss.connection.secure.enabled</name>
        <value>false</value>
    </property>
    <property>
        <name>fs.oss.connection.maximum</name>
        <value>2048</value>
    </property>
</configuration>
3. 启动已经配置好的kafka-connector插件

启动及删除connect任务命令

## 启动命令
curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @sqlserver-cdc-source.json
curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @adb4pg-jdbc-sink.json
curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://localhost:8083/connectors/ -d @oss-sink.json

## 删除命令
curl -s -X DELETE http://localhost:8083/connectors/sqlserver-cdc-source
curl -s -X DELETE http://localhost:8083/connectors/adb4pg-jdbc-sink
curl -s -X DELETE http://localhost:8083/connectors/oss-sink

在ADB For PG获取更新数据

SQLServer插入赠/更/删数据记录

insert into t1(a,b,c,d,e,f,g) values(1, 2, 3, 4, 5, convert(datetime,'24-12-19 10:34:09 PM',5), 'h');

在kafka topic获取更新结果

先确认是否生成了kafka-connect所需的topic信息

bin/kafka-topics.sh --zookeeper zk_address --list

image
如截图,connect-configs, connect-offsets, connect-status为kafka-connect用来存储任务数据更新状态的topic。schema-changes-inventory是维护sqlserver表结构的topic。
可以通过kafka consloe-consumer上获取到的topic信息,以确认cdc数据正确被采集到kafka topic

bin/kafka-console-consumer.sh --bootstrap-server xx.xx.xx.xx:9092 --topic server1.dbo.t1

在ADB For PG上查询同步过来的数据

注意:因为是不同数据库之间的同步,时区设置的不同可能会导致同步结果产生时区偏移,需要在两侧数据库做好设置。
image

在OSS查看更新的数据

image

相关实践学习
数据库实验室挑战任务-初级任务
本场景介绍如何开通属于你的免费云数据库,在RDS-MySQL中完成对学生成绩的详情查询,执行指定类型SQL。
阿里云云原生数据仓库AnalyticDB MySQL版 使用教程
云原生数据仓库AnalyticDB MySQL版是一种支持高并发低延时查询的新一代云原生数据仓库,高度兼容MySQL协议以及SQL:92、SQL:99、SQL:2003标准,可以对海量数据进行即时的多维分析透视和业务探索,快速构建企业云上数据仓库。 了解产品 https://www.aliyun.com/product/ApsaraDB/ads
目录
相关文章
|
2月前
|
SQL Cloud Native 关系型数据库
ADBPG(AnalyticDB for PostgreSQL)是阿里云提供的一种云原生的大数据分析型数据库
ADBPG(AnalyticDB for PostgreSQL)是阿里云提供的一种云原生的大数据分析型数据库
235 1
|
3月前
|
运维 Cloud Native 安全
降本50%!阿里云发布AnalyticDB数仓升舱解决方案
降本50%!阿里云发布AnalyticDB数仓升舱解决方案
121 0
|
7月前
|
OLAP
《分析型数据库(AnalyticDB)2.0深度技术解析》电子版地址
分析型数据库(AnalyticDB)2.0深度技术解析
46 0
《分析型数据库(AnalyticDB)2.0深度技术解析》电子版地址
|
9月前
|
运维 Cloud Native 关系型数据库
今日14点开播!云原生AnalyticDB MySQL湖仓版如何支撑「DAS」高吞吐入湖和高性能查询
《数据库风向标》是一档聚焦数据库新趋势与新技术的视频栏目,节目每期会请到几位资深技术大咖,与大家共话数据库热点话题。
今日14点开播!云原生AnalyticDB MySQL湖仓版如何支撑「DAS」高吞吐入湖和高性能查询
|
9月前
|
弹性计算 运维 监控
2022云栖精选—AnalyticDB MySQL 湖仓版在B站的最佳实践
陈浩 哔哩哔哩Bilibili资深数据开发专家
2022云栖精选—AnalyticDB MySQL 湖仓版在B站的最佳实践
|
10月前
《阿里云AnalyticDB+Data Works 企业数仓构》电子版地址
阿里云AnalyticDB+Data Works 企业数仓构建方案解析
54 0
《阿里云AnalyticDB+Data Works 企业数仓构》电子版地址
|
10月前
|
存储 分布式计算 Cloud Native
重磅发布|AnalyticDB MySQL湖仓版公测发布:从湖到仓,打造云原生一站式数据分析平台
云原生数据仓库产品AnalyticDB MySQL数仓版,是阿里云基于集团电商双11业务百亿次在线分析的最佳实践,推出的业界首款兼容MySQL协议,且性能全球第一(TPC-DS 10TB)的云数据仓库产品。企业只需要招一些具备SQL技能的数据分析师,搭配上一个QuickBI/DataV/自研可视化报表,就可以快速将企业的关键指标实时可视化展示,帮助企业转型成数据驱动决策的公司。
重磅发布|AnalyticDB MySQL湖仓版公测发布:从湖到仓,打造云原生一站式数据分析平台
|
11月前
|
SQL 关系型数据库 MySQL
阿里云分析型数据库MySQL版(AnalyticDB)测试初体验(2)
阿里云分析型数据库MySQL版(AnalyticDB)测试初体验
阿里云分析型数据库MySQL版(AnalyticDB)测试初体验(2)
|
11月前
|
SQL 存储 关系型数据库
阿里云分析型数据库MySQL版(AnalyticDB)测试初体验(1)
阿里云分析型数据库MySQL版(AnalyticDB)测试初体验
219 0
阿里云分析型数据库MySQL版(AnalyticDB)测试初体验(1)
|
运维 Cloud Native 安全
重磅 | 阿里云发布AnalyticDB数仓升舱解决方案 可帮客户节省50% IT采购成本
阿里云发布AnalyticDB数据仓库升舱解决方案,可帮助金融、电信等行业客户实现传统数仓向云原生数据仓库的平滑升级。
318 0
重磅 | 阿里云发布AnalyticDB数仓升舱解决方案 可帮客户节省50% IT采购成本
热门文章
最新文章
相关产品
云原生数据仓库 AnalyticDB PostgreSQL版
推荐文章
更多