SQLServer CDC数据通过Kafka connect实时同步至分析型数据库 AnalyticDB For PostgreSQL及OSS-阿里云开发者社区

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

2019-12-26 9966

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

RDS PostgreSQL Serverless，0.5-4RCU 50GB 3个月

阿里云百炼推荐规格 ADB PostgreSQL，4核16GB 100GB 1个月

简介： 本文主要介绍如何通过消息对接, 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的表需要主键或者唯一主键。

图1：Sqlserver CDC原理

ADB4PG Sink使用条件

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

环境准备

SQLServer环境准备

已有自建SQLServer或云上RDS实例（示例使用云上RDS SQLServer实例）
已有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;
        
          
        
        
        
          
          AI 代码解读

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
);
        
          
        
        
        
          
          AI 代码解读

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
...
        
          
        
        
        
          
          AI 代码解读

3. 启动kafka server

bin/kafka-server-start.sh config/server.properties
        
          
        
        
        
          
          AI 代码解读

安装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=
        
          
        
        
        
          
          AI 代码解读

安装需要的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"
    }
}
        
          
        
        
        
          
          AI 代码解读

# 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"
        }
}
        
          
        
        
        
          
          AI 代码解读

有关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"
        }
}
        
          
        
        
        
          
          AI 代码解读

由于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>
        
          
        
        
        
          
          AI 代码解读

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
        
          
        
        
        
          
          AI 代码解读

在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');
        
          
        
        
        
          
          AI 代码解读

在kafka topic获取更新结果

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

bin/kafka-topics.sh --zookeeper zk_address --list
        
          
        
        
        
          
          AI 代码解读

如截图，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
        
          
        
        
        
          
          AI 代码解读

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

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

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

背景

CDC的使用条件

ADB4PG Sink使用条件

环境准备

SQLServer环境准备

SQLServer环境建表

ADB4PG端创建目标表

Kafka环境准备

安装Kafka Server

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

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

3. 启动kafka server

安装Kafka Connect

1. 修改kafka connect配置文件

安装需要的kafka-connect插件

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

2. 编辑配置文件

3. 启动已经配置好的kafka-connector插件

在ADB For PG获取更新数据

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

在kafka topic获取更新结果

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

在OSS查看更新的数据

云原生数据仓库 AnalyticDB PostgreSQL版

热门文章

最新文章

相关产品

相关课程

相关电子书

相关实验场景

推荐镜像

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

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

背景

CDC的使用条件

ADB4PG Sink使用条件

环境准备

SQLServer环境准备

SQLServer环境建表

ADB4PG端创建目标表

Kafka环境准备

安装Kafka Server

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

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

3. 启动kafka server

安装Kafka Connect

1. 修改kafka connect配置文件

安装需要的kafka-connect插件

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

2. 编辑配置文件

3. 启动已经配置好的kafka-connector插件

在ADB For PG获取更新数据

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

在kafka topic获取更新结果

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

在OSS查看更新的数据

云原生数据仓库 AnalyticDB PostgreSQL版

热门文章

最新文章

相关产品

相关课程

相关电子书

相关实验场景

推荐镜像