Flink 版本 | 说明 |
1.11 | 不支持 |
1.13 | 支持 |
1.14 | 不支持 |
1.16 | 不支持 |
USE MyDBGOEXEC sys.sp_cdc_enable_table@source_schema = N'dbo', -- 指定源表所属的 schema 名@source_name = N'MyTable', -- 指定需要读取的源表名@role_name = NULL, -- 指定一个角色 MyRole,您可以向该角色添加要对源表的捕获列授予 SELECT 权限的用户。sysadmin 或 db_owner 角色的用户也可以访问指定的源表。将 @role_name 的值设置为 NULL,以仅允许使用 sysadmin 或 db_owner 中的用户。@filegroup_name = NULL, -- 要用于为捕获实例创建的更改表的文件组,可以为 NULL。 如果指定,则必须为当前数据库定义 filegroup_name 。 如果为 NULL,则使用默认文件组。@supports_net_changes = 0 -- 是否为此捕获实例启用对查询净更改的支持。 如果表具有主键或表具有使用 @index_name 参数标识的唯一索引,则supports_net_changes 默认值为 1。 否则,该参数默认为 0。如果为 0,则只生成查询所有更改的支持函数。如果为 1,则还会生成查询净更改所需的函数。如果 supports_net_changes 设置为 1,则必须指定 index_name ,或者源表必须具有定义的主键。GO
USE MyDBGOEXEC sys.sp_cdc_help_change_data_captureGO
-- register a SqlServer table 'orders' in Flink SQLCREATE TABLE orders (id INT,order_date DATE,purchaser INT,quantity INT,product_id INT,PRIMARY KEY (id) NOT ENFORCED) WITH ('connector' = 'sqlserver-cdc','hostname' = 'localhost','port' = '1433','username' = 'sa','password' = 'Password!','database-name' = 'inventory','schema-name' = 'dbo','table-name' = 'orders');-- read snapshot and binlogs from orders tableSELECT * FROM orders;
参数 | 必填 | 默认值 | 类型 | 描述 |
connector | 是 | 无 | String | 固定值为 sqlserver-cd |
hostname | 是 | 无 | String | SQLSerer 数据库 IP 地址或主机名 |
username | 是 | 无 | String | SQLSerer 数据库用户名 |
password | 是 | 无 | String | SQLSerer 数据库密码 |
database-name | 是 | 无 | String | SQLSerer 源表所属数据库名 |
schema-name | 是 | 无 | String | SQLSerer 源表所属 schema 名。支持 Java 正则表达式,例如 (dbo.*) 可匹配 dbo、dbo1、dbo_test,对于正则表达式,强烈建议放置于括号内,以防止与 table-name 组合时出现错误 |
table-name | 是 | 无 | String | SQLServer 源表名。支持 Java 正则表达式,对于正则表达式,强烈建议放置于括号内,以防止与 schema-name 组合时出现错误 |
port | 否 | 1433 | Integer | SQLSerer 数据库端口 |
server-time-zone | 否 | UTC | String | SQLSerer 数据库会话时区设置,例如 'Asia/Shanghai' |
debezium.* | 否 | 无 | String | Debezium 属性参数,从更细粒度控制 Debezium 客户端的行为。例如 'debezium.snapshot.mode' = 'initial_only',详情参见 Debezium's SQLServer Connector properties |
元数据名 | 数据类型 | 描述 |
table_name | STRING NOT NULL | 包含该 Row 的表名称 |
schema_name | STRING NOT NULL | 包含该 Row 的 schema 名称 |
database_name | STRING NOT NULL | 包含该 Row 的数据库名称 |
op_ts | TIMESTAMP_LTZ(3) NOT NULL | Row 在数据库中进行更改的时间。全量阶段数据,该字段值为 0 |
CREATE TABLE products (table_name STRING METADATA FROM 'table_name' VIRTUAL,schema_name STRING METADATA FROM 'schema_name' VIRTUAL,db_name STRING METADATA FROM 'database_name' VIRTUAL,operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,id INT NOT NULL,name STRING,description STRING,weight DECIMAL(10,3)) WITH ('connector' = 'sqlserver-cdc','hostname' = 'localhost','port' = '1433','username' = 'sa','password' = 'Password!','database-name' = 'inventory','schema-name' = 'dbo','table-name' = 'products');
SQLServer 类型 | Flink SQL 类型 |
char(n) | CHAR(n) |
varchar(n) nvarchar(n) nchar(n) | VARCHAR(n) |
text ntext xml | STRING |
decimal(p, s) money smallmoney | DECIMAL(p, s) |
numeric | NUMERIC |
float real | DOUBLE |
bit | BOOLEAN |
int | INT |
tinyint | SMALLINT |
smallint | SMALLINT |
bigint | BIGINT |
date | DATE |
time(n) | TIME(n) |
datetime2 datetime smalldatetime | TIMESTAMP(n) |
datetimeoffset | TIMESTAMP_LTZ(3) |
execution.checkpointing.interval: 10minexecution.checkpointing.tolerable-failed-checkpoints: 100restart-strategy: fixed-delayrestart-strategy.fixed-delay.attempts: 2147483647
本页内容是否解决了您的问题?