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

本文涉及的产品
云数据库 RDS SQL Server,独享型 2核4GB
实时计算 Flink 版,5000CU*H 3个月
RDS SQL Server Serverless,2-4RCU 50GB 3个月
推荐场景:
简介: 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 验证

验证新增:


验证修改:


验证删除:

相关实践学习
如何在云端创建MySQL数据库
开始实验后,系统会自动创建一台自建MySQL的 源数据库 ECS 实例和一台 目标数据库 RDS。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心,利用清洁计算为万物互联的新世界提供源源不断的能源动力,目前开服的区域包括中国(华北、华东、华南、香港)、新加坡、美国(美东、美西)、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程,来了解阿里云能够为你的业务带来哪些帮助     相关的阿里云产品:云服务器ECS 云服务器 ECS(Elastic Compute Service)是一种弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
目录
相关文章
|
28天前
|
DataWorks 安全 关系型数据库
DataWorks产品使用合集之使用Flink CDC读取PostgreSQL数据时如何指定编码格式
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
|
22天前
|
Java 关系型数据库 MySQL
实时计算 Flink版产品使用问题之如何实现MySQL数据库之间的实时同步
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
23天前
|
Oracle 关系型数据库 Java
实时计算 Flink版产品使用问题之如何实现Oracle到其他系统的实时同步
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
23天前
|
消息中间件 SQL Kafka
实时计算 Flink版产品使用问题之如何实现OSS数据到Kafka的实时同步
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
23天前
|
SQL 监控 关系型数据库
实时计算 Flink版产品使用问题之使用mysql cdc配置StartupOptions.initial()全量之后就不增量了,是什么原因
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
21天前
|
SQL 关系型数据库 MySQL
实时计算 Flink版操作报错合集之CDC任务在异常后整个record sent从0初始化开始,是什么导致的
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
467 0
|
21天前
|
关系型数据库 数据库 流计算
实时计算 Flink版操作报错合集之在使用Flink CDC TiDB Connector时,无法获取到事件,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
348 0
|
21天前
|
关系型数据库 MySQL 数据库
实时计算 Flink版操作报错合集之下载了mysql的cdc的demo,在本地调试时,报错:找不到这个包,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
118 0
|
22天前
|
Java 数据库连接 数据处理
实时计算 Flink版产品使用问题之JDBC连接器实时同步的时候如何删除数据吗
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
26天前
|
消息中间件 关系型数据库 MySQL
Flink CDC 最佳实践(以 MySQL 为例)
Flink CDC 最佳实践(以 MySQL 为例)