flink sqlserver cdc实时同步(含sqlserver安装配置等)

本文涉及的产品
实时计算 Flink 版,1000CU*H 3个月
RDS MySQL Serverless 基础系列,0.5-2RCU 50GB
云数据库 RDS SQL Server,基础系列 2核4GB
简介: flink sqlserver cdc实时同步(含sqlserver安装配置等)

01 引言

官方文档:https://github.com/ververica/flink-cdc-connectors/blob/master/docs/content/connectors/sqlserver-cdc.md

如果要使用flink cdc做sqlserver的实时同步,需要满足以下条件:

  1. 需要安装SQLServer(需要支持CDC的功能,SQLServer 2008之后的版本都支持)
  2. 需要开启SQL Server代理;
  3. 启用CDC功能。

ok,接下来开始讲解。

02 SQLServer安装

首先需要先安装SqlServer(使用的是2019版本),有兴趣的同学可以参考博主之前写的《Docker下安装SqlServer2019》

主要就是两个步骤:

## 拉取最新镜像
docker pull mcr.microsoft.com/mssql/server:2019-latest
## 运行 SQL Server 容器(密码必须是8个字符,并包含字母、数字和特殊字符,如:abc@123456 ,下面映射主机端口为30027)
docker run -e 'ACCEPT_EULA=Y' -e 'SA_PASSWORD=abc@123456' -p 30027:1433 --name sql_server_2019 -d mcr.microsoft.com/mssql/server:2019-latest

03 开启SQLServer代理

首先使用root用户进入容器:

docker exec -it --user root sql_server_2019 bash

进入容器后,执行命令启用SqlServeragent:

/opt/mssql/bin/mssql-conf set sqlagent.enabled true

退出,并重启容器:

exit
docker restart sql_server_2019

具体操作如下:

04 开启CDC功能


step1:创建’cdc_test’数据库,并使用连接工具登录该数据库,使用以下 SQL 命令启用 CDC 功能:

-- 创建数据库
CREATE DATABASE cdc_test;
-- 启用CDC功能
EXEC sys.sp_cdc_enable_db;
-- 判断当前数据库是否启用了CDC(如果返回1,表示已启用)
SELECT is_cdc_enabled FROM sys.databases WHERE name = 'cdc_test';


step2:选择要进行 CDC 跟踪的表(这里使用orders表作为演示

-- 创建示例表(orders)
CREATE TABLE orders (
     id int,
     order_date date,
     purchaser int,
     quantity int,
     product_id int,
     PRIMARY KEY ([id])
);
-- schema_name 是表所属的架构(schema)的名称。
-- table_name 是要启用 CDC 跟踪的表的名称。
-- cdc_role 是 CDC 使用的角色的名称。如果没有指定角色名称,系统将创建一个默认角色。
EXEC sys.sp_cdc_enable_table
  @source_schema = 'dbo',
  @source_name   = 'orders',
  @role_name     = 'cdc_role';

执行结果如下:


step3:启用 CDC 后,SQL Server 将自动跟踪启用了 CDC 的表上的数据更改,并将更改信息存储在 CDC 相关的表中,您可以使用这些信息进行数据更改追踪和同步。

-- 查询在当前数据库下所有的表:
SELECT * FROM INFORMATION_SCHEMA.TABLES

05 Flink SQL

ok,现在可以写FlinkSQL了,如下:

-- 创建源表t_source_sqlserver,使用SQL Server Change Data Capture (CDC)连接器从SQL Server数据库读取数据
CREATE TABLE t_source_sqlserver (
    id INT,
    order_date DATE,
    purchaser INT,
    quantity INT,
    product_id INT,
    PRIMARY KEY (id) NOT ENFORCED -- 主键定义(可选)
) WITH (
    'connector' = 'sqlserver-cdc',  -- 使用SQL Server CDC连接器
    'hostname' = '10.194.183.120',  -- SQL Server主机名
    'port' = '30027',               -- SQL Server端口
    'username' = 'sa',              -- SQL Server用户名
    'password' = 'abc@123456',      -- SQL Server密码
    'database-name' = 'cdc_test',   -- 数据库名称
    'schema-name' = 'dbo',          -- 模式名称
    'table-name' = 'orders'         -- 要捕获更改的表名
);
-- 创建目标表table_sink_mysql,使用JDBC连接器将数据写入MySQL数据库
CREATE TABLE table_sink_mysql (
    id INT,
    order_date DATE,
    purchaser INT,
    quantity INT,
    product_id INT,
    PRIMARY KEY (id) NOT ENFORCED  -- 主键定义(可选)
)
WITH (
    'connector' = 'jdbc',                        -- 使用JDBC连接器
    'url' = 'jdbc:mysql://10.194.183.120:30025/test',  -- MySQL的JDBC URL
    'username' = 'root',                        -- MySQL用户名
    'password' = 'root',                        -- MySQL密码
    'table-name' = 'orders'                     -- 要写入的MySQL表名
);
-- 从t_source_sqlserver表中选择数据,并将其插入到table_sink_mysql表中
INSERT INTO table_sink_mysql SELECT * FROM t_source_sqlserver;

启动程序,一切正常:

06 验证

验证新增:


验证修改:


验证删除:

相关实践学习
每个IT人都想学的“Web应用上云经典架构”实战
本实验从Web应用上云这个最基本的、最普遍的需求出发,帮助IT从业者们通过“阿里云Web应用上云解决方案”,了解一个企业级Web应用上云的常见架构,了解如何构建一个高可用、可扩展的企业级应用架构。
MySQL数据库入门学习
本课程通过最流行的开源数据库MySQL带你了解数据库的世界。   相关的阿里云产品:云数据库RDS MySQL 版 阿里云关系型数据库RDS(Relational Database Service)是一种稳定可靠、可弹性伸缩的在线数据库服务,提供容灾、备份、恢复、迁移等方面的全套解决方案,彻底解决数据库运维的烦恼。 了解产品详情: https://www.aliyun.com/product/rds/mysql 
目录
相关文章
|
2月前
|
SQL 关系型数据库 Apache
从 Flink 到 Doris 的实时数据写入实践 —— 基于 Flink CDC 构建更实时高效的数据集成链路
本文将深入解析 Flink-Doris-Connector 三大典型场景中的设计与实现,并结合 Flink CDC 详细介绍了整库同步的解决方案,助力构建更加高效、稳定的实时数据处理体系。
1022 0
从 Flink 到 Doris 的实时数据写入实践 —— 基于 Flink CDC 构建更实时高效的数据集成链路
|
5月前
|
数据采集 SQL canal
Amoro + Flink CDC 数据融合入湖新体验
本文总结了货拉拉高级大数据开发工程师陈政羽在Flink Forward Asia 2024上的分享,聚焦Flink CDC在货拉拉的应用与优化。内容涵盖CDC应用现状、数据入湖新体验、入湖优化及未来规划。文中详细分析了CDC在多业务场景中的实践,包括数据采集平台化、稳定性建设,以及面临的文件碎片化、Schema演进等挑战。同时介绍了基于Apache Amoro的湖仓融合架构,通过自优化服务解决小文件问题,提升数据新鲜度与读写平衡。未来将深化Paimon与Amoro的结合,打造更高效的入湖生态与自动化优化方案。
257 1
Amoro + Flink CDC 数据融合入湖新体验
|
5月前
|
SQL 关系型数据库 MySQL
Flink CDC 3.4 发布, 优化高频 DDL 处理,支持 Batch 模式,新增 Iceberg 支持
Apache Flink CDC 3.4.0 版本正式发布!经过4个月的开发,此版本强化了对高频表结构变更的支持,新增 batch 执行模式和 Apache Iceberg Sink 连接器,可将数据库数据全增量实时写入 Iceberg 数据湖。51位贡献者完成了259次代码提交,优化了 MySQL、MongoDB 等连接器,并修复多个缺陷。未来 3.5 版本将聚焦脏数据处理、数据限流等能力及 AI 生态对接。欢迎下载体验并提出反馈!
882 1
Flink CDC 3.4 发布, 优化高频 DDL 处理,支持 Batch 模式,新增 Iceberg 支持
|
4月前
|
消息中间件 SQL 关系型数据库
Flink CDC + Kafka 加速业务实时化
Flink CDC 是一种支持流批一体的分布式数据集成工具,通过 YAML 配置实现数据传输过程中的路由与转换操作。它已从单一数据源的 CDC 数据流发展为完整的数据同步解决方案,支持 MySQL、Kafka 等多种数据源和目标端(如 Delta Lake、Iceberg)。其核心功能包括多样化数据输入链路、Schema Evolution、Transform 和 Routing 模块,以及丰富的监控指标。相比传统 SQL 和 DataStream 作业,Flink CDC 提供更灵活的 Schema 变更控制和原始 binlog 同步能力。
|
16天前
|
SQL 人工智能 Linux
SQL Server 2025 RC1 发布 - 从本地到云端的 AI 就绪企业数据库
SQL Server 2025 RC1 发布 - 从本地到云端的 AI 就绪企业数据库
172 5
SQL Server 2025 RC1 发布 - 从本地到云端的 AI 就绪企业数据库
|
15天前
|
关系型数据库 MySQL 数据库
阿里云数据库RDS费用价格:MySQL、SQL Server、PostgreSQL和MariaDB引擎收费标准
阿里云RDS数据库支持MySQL、SQL Server、PostgreSQL、MariaDB,多种引擎优惠上线!MySQL倚天版88元/年,SQL Server 2核4G仅299元/年,PostgreSQL 227元/年起。高可用、可弹性伸缩,安全稳定。详情见官网活动页。
|
15天前
|
关系型数据库 分布式数据库 数据库
阿里云数据库收费价格:MySQL、PostgreSQL、SQL Server和MariaDB引擎费用整理
阿里云数据库提供多种类型,包括关系型与NoSQL,主流如PolarDB、RDS MySQL/PostgreSQL、Redis等。价格低至21元/月起,支持按需付费与优惠套餐,适用于各类应用场景。
|
6月前
|
SQL 数据库 数据安全/隐私保护
数据库数据恢复——sql server数据库被加密的数据恢复案例
SQL server数据库数据故障: SQL server数据库被加密,无法使用。 数据库MDF、LDF、log日志文件名字被篡改。 数据库备份被加密,文件名字被篡改。
|
2月前
|
SQL 人工智能 Linux
SQL Server 2025 RC0 发布 - 从本地到云端的 AI 就绪企业数据库
SQL Server 2025 RC0 发布 - 从本地到云端的 AI 就绪企业数据库
142 5
|
21天前
|
关系型数据库 MySQL 数据库
阿里云数据库RDS支持MySQL、SQL Server、PostgreSQL和MariaDB引擎
阿里云数据库RDS支持MySQL、SQL Server、PostgreSQL和MariaDB引擎,提供高性价比、稳定安全的云数据库服务,适用于多种行业与业务场景。

热门文章

最新文章