Skip to content

Commit 97b36ce

Browse files
DaishuyuanleonardBang
Daishuyuan
authored and
leonardBang
committed
[FLINK-39252][pipeline-connector/sqlserver] Introduce SQLServer Pipeline Source connector
1 parent 7355d76 commit 97b36ce

44 files changed

Lines changed: 7165 additions & 23 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎docs/content.zh/docs/connectors/pipeline-connectors/overview.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ Flink CDC 提供了可用于 YAML 作业的 Pipeline Source 和 Sink 连接器
4444
| [Oracle]({{< ref "docs/connectors/pipeline-connectors/oracle" >}}) | Source | <li> [Oracle](https://www.oracle.com/database/) | [Oracle](https://mvnrepository.com/artifact/org.apache.flink/flink-cdc-pipeline-connector-oracle) |
4545
| [Paimon]({{< ref "docs/connectors/pipeline-connectors/paimon" >}}) | Sink | <li> [Paimon](https://paimon.apache.org/): 0.6, 0.7, 0.8, 0.9, 1.0, 1.1, 1.2, 1.3 | [Paimon](https://mvnrepository.com/artifact/org.apache.flink/flink-cdc-pipeline-connector-paimon) |
4646
| [Postgres]({{< ref "docs/connectors/pipeline-connectors/postgres" >}}) | Source | <li> [PostgreSQL](https://www.postgresql.org/) | [Postgres](https://mvnrepository.com/artifact/org.apache.flink/flink-cdc-pipeline-connector-postgres) |
47+
| [SQL Server]({{< ref "docs/connectors/pipeline-connectors/sqlserver" >}}) | Source | <li> [SQL Server](https://www.microsoft.com/sql-server/) | [SQL Server](https://mvnrepository.com/artifact/org.apache.flink/flink-cdc-pipeline-connector-sqlserver) |
4748
| [StarRocks]({{< ref "docs/connectors/pipeline-connectors/starrocks" >}}) | Sink | <li> [StarRocks](https://www.starrocks.io/): 2.x, 3.x | [StarRocks](https://mvnrepository.com/artifact/org.apache.flink/flink-cdc-pipeline-connector-starrocks) |
4849

4950
## Supported Flink Versions

‎docs/content.zh/docs/connectors/pipeline-connectors/sqlserver.md‎

Lines changed: 316 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,316 @@
1+
---
2+
title: "SQL Server"
3+
weight: 2
4+
type: docs
5+
aliases:
6+
- /connectors/pipeline-connectors/sqlserver
7+
---
8+
<!--
9+
Licensed to the Apache Software Foundation (ASF) under one
10+
or more contributor license agreements. See the NOTICE file
11+
distributed with this work for additional information
12+
regarding copyright ownership. The ASF licenses this file
13+
to you under the Apache License, Version 2.0 (the
14+
"License"); you may not use this file except in compliance
15+
with the License. You may obtain a copy of the License at
16+
17+
http://www.apache.org/licenses/LICENSE-2.0
18+
19+
Unless required by applicable law or agreed to in writing,
20+
software distributed under the License is distributed on an
21+
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
22+
KIND, either express or implied. See the License for the
23+
specific language governing permissions and limitations
24+
under the License.
25+
-->
26+
27+
# SQL Server Connector
28+
29+
SQL Server CDC Pipeline 连接器支持从 SQL Server 数据库读取快照数据和增量数据，并提供端到端的表数据同步能力。本文描述如何设置 SQL Server CDC Pipeline 连接器。
30+
31+
## 前置条件
32+
33+
- SQL Server Agent 正在运行。
34+
- 已为数据库和需要捕获的表开启 Change Data Capture。
35+
- 配置的 SQL Server 用户可以连接服务器并读取被捕获的表。
36+
- 当前连接器的 `tables` 配置只支持一个字面量数据库名；schema 和 table 部分可以使用正则表达式。
37+
38+
为数据库和表启用 CDC：
39+
40+
```sql
41+
USE MyDB;
42+
GO
43+
EXEC sys.sp_cdc_enable_db;
44+
GO
45+
46+
EXEC sys.sp_cdc_enable_table
47+
@source_schema = N'dbo',
48+
@source_name = N'MyTable',
49+
@role_name = NULL,
50+
@supports_net_changes = 0;
51+
GO
52+
```
53+
54+
检查表是否已启用 CDC：
55+
56+
```sql
57+
USE MyDB;
58+
GO
59+
EXEC sys.sp_cdc_help_change_data_capture;
60+
GO
61+
```
62+
63+
## 示例
64+
65+
从 SQL Server 读取数据同步到 Doris 的 Pipeline 可以定义如下：
66+
67+
```yaml
68+
source:
69+
type: sqlserver
70+
name: SQL Server Source
71+
hostname: 127.0.0.1
72+
port: 1433
73+
username: debezium
74+
password: password
75+
# 所有匹配的表必须使用同一个字面量数据库名。
76+
tables: inventory.dbo.\.*
77+
schema-change.enabled: true
78+
79+
sink:
80+
type: doris
81+
name: Doris Sink
82+
fenodes: 127.0.0.1:8030
83+
username: root
84+
password: password
85+
86+
pipeline:
87+
name: SQL Server to Doris Pipeline
88+
parallelism: 4
89+
```
90+
91+
## 连接器配置项
92+
93+
<div class="highlight">
94+
<table class="colwidths-auto docutils">
95+
<thead>
96+
<tr>
97+
<th class="text-left" style="width: 10%">Option</th>
98+
<th class="text-left" style="width: 8%">Required</th>
99+
<th class="text-left" style="width: 7%">Default</th>
100+
<th class="text-left" style="width: 10%">Type</th>
101+
<th class="text-left" style="width: 65%">Description</th>
102+
</tr>
103+
</thead>
104+
<tbody>
105+
<tr>
106+
<td>hostname</td>
107+
<td>required</td>
108+
<td style="word-wrap: break-word;">(none)</td>
109+
<td>String</td>
110+
<td>SQL Server 数据库服务器的 IP 地址或主机名。</td>
111+
</tr>
112+
<tr>
113+
<td>port</td>
114+
<td>optional</td>
115+
<td style="word-wrap: break-word;">1433</td>
116+
<td>Integer</td>
117+
<td>SQL Server 数据库服务器的整数端口号。</td>
118+
</tr>
119+
<tr>
120+
<td>username</td>
121+
<td>required</td>
122+
<td style="word-wrap: break-word;">(none)</td>
123+
<td>String</td>
124+
<td>连接 SQL Server 数据库服务器时使用的 SQL Server 用户名。</td>
125+
</tr>
126+
<tr>
127+
<td>password</td>
128+
<td>required</td>
129+
<td style="word-wrap: break-word;">(none)</td>
130+
<td>String</td>
131+
<td>连接 SQL Server 数据库服务器时使用的密码。</td>
132+
</tr>
133+
<tr>
134+
<td>tables</td>
135+
<td>required</td>
136+
<td style="word-wrap: break-word;">(none)</td>
137+
<td>String</td>
138+
<td>需要监控的 SQL Server 表名。schema 和 table 名支持正则表达式。所有表匹配规则必须使用同一个字面量数据库名；不支持数据库级别正则表达式或跨数据库匹配。点号（.）会被视为 database、schema 和 table 名的分隔符。如果需要在正则表达式中使用点号（.）匹配任意字符，必须使用反斜杠转义。<br>
139+
示例：inventory.dbo.\.*、inventory.dbo.user_table_[0-9]+、inventory.dbo.(app|web)_order_\.*</td>
140+
</tr>
141+
<tr>
142+
<td>tables.exclude</td>
143+
<td>optional</td>
144+
<td style="word-wrap: break-word;">(none)</td>
145+
<td>String</td>
146+
<td>在应用 `tables` 后需要排除的 SQL Server 表名。schema 和 table 名支持正则表达式。所有排除规则必须与 `tables` 选项使用同一个字面量数据库名。<br>
147+
示例：inventory.dbo.audit_\.*、inventory.dbo.tmp_[0-9]+</td>
148+
</tr>
149+
<tr>
150+
<td>schema-change.enabled</td>
151+
<td>optional</td>
152+
<td style="word-wrap: break-word;">true</td>
153+
<td>Boolean</td>
154+
<td>是否发送 schema change 事件，使下游 sink 可以同步表结构变更。</td>
155+
</tr>
156+
<tr>
157+
<td>server-time-zone</td>
158+
<td>optional</td>
159+
<td style="word-wrap: break-word;">(none)</td>
160+
<td>String</td>
161+
<td>数据库服务器中的会话时区。如果未设置，则使用 ZoneId.systemDefault() 确定服务器时区。</td>
162+
</tr>
163+
<tr>
164+
<td>scan.incremental.snapshot.chunk.key-column</td>
165+
<td>optional</td>
166+
<td style="word-wrap: break-word;">(none)</td>
167+
<td>String</td>
168+
<td>表快照的切分键。默认使用主键的第一列，该列必须是主键列。</td>
169+
</tr>
170+
<tr>
171+
<td>scan.incremental.snapshot.chunk.size</td>
172+
<td>optional</td>
173+
<td style="word-wrap: break-word;">8096</td>
174+
<td>Integer</td>
175+
<td>表快照的 chunk 大小，单位为行数。</td>
176+
</tr>
177+
<tr>
178+
<td>scan.snapshot.fetch.size</td>
179+
<td>optional</td>
180+
<td style="word-wrap: break-word;">1024</td>
181+
<td>Integer</td>
182+
<td>读取表快照时每次轮询的最大读取条数。</td>
183+
</tr>
184+
<tr>
185+
<td>scan.startup.mode</td>
186+
<td>optional</td>
187+
<td style="word-wrap: break-word;">initial</td>
188+
<td>String</td>
189+
<td>SQL Server CDC 消费者可选启动模式。合法值为 "initial"、"latest-offset"、"snapshot" 和 "timestamp"。</td>
190+
</tr>
191+
<tr>
192+
<td>scan.startup.timestamp-millis</td>
193+
<td>optional</td>
194+
<td style="word-wrap: break-word;">(none)</td>
195+
<td>Long</td>
196+
<td>当 `scan.startup.mode` 为 "timestamp" 时使用的启动时间戳。</td>
197+
</tr>
198+
<tr>
199+
<td>scan.incremental.snapshot.backfill.skip</td>
200+
<td>optional</td>
201+
<td style="word-wrap: break-word;">true</td>
202+
<td>Boolean</td>
203+
<td>是否在快照读取阶段跳过 backfill。跳过 backfill 可能导致部分 change log 事件以 at-least-once 语义被重放。</td>
204+
</tr>
205+
<tr>
206+
<td>connect.timeout</td>
207+
<td>optional</td>
208+
<td style="word-wrap: break-word;">30s</td>
209+
<td>Duration</td>
210+
<td>连接器尝试连接 SQL Server 数据库服务器后的最长等待时间。</td>
211+
</tr>
212+
<tr>
213+
<td>connect.max-retries</td>
214+
<td>optional</td>
215+
<td style="word-wrap: break-word;">3</td>
216+
<td>Integer</td>
217+
<td>连接器建立 SQL Server 数据库服务器连接的最大重试次数。</td>
218+
</tr>
219+
<tr>
220+
<td>connection.pool.size</td>
221+
<td>optional</td>
222+
<td style="word-wrap: break-word;">20</td>
223+
<td>Integer</td>
224+
<td>连接池大小。</td>
225+
</tr>
226+
<tr>
227+
<td>metadata.list</td>
228+
<td>optional</td>
229+
<td style="word-wrap: break-word;">(none)</td>
230+
<td>String</td>
231+
<td>从 SourceRecord 中读取并传递给下游的元数据列表，使用英文逗号分隔。可用元数据包括：database_name、schema_name、table_name、op_ts。</td>
232+
</tr>
233+
<tr>
234+
<td>scan.incremental.close-idle-reader.enabled</td>
235+
<td>optional</td>
236+
<td style="word-wrap: break-word;">false</td>
237+
<td>Boolean</td>
238+
<td>是否在快照阶段结束时关闭空闲 reader。此特性依赖 FLIP-147。</td>
239+
</tr>
240+
<tr>
241+
<td>scan.newly-added-table.enabled</td>
242+
<td>optional</td>
243+
<td style="word-wrap: break-word;">false</td>
244+
<td>Boolean</td>
245+
<td>是否扫描新增表。该选项仅在作业从 savepoint 或 checkpoint 启动时有用。</td>
246+
</tr>
247+
<tr>
248+
<td>debezium.*</td>
249+
<td>optional</td>
250+
<td style="word-wrap: break-word;">(none)</td>
251+
<td>String</td>
252+
<td>传递给 Debezium Embedded Engine 的 Debezium 属性。</td>
253+
</tr>
254+
<tr>
255+
<td>jdbc.properties.*</td>
256+
<td>optional</td>
257+
<td style="word-wrap: break-word;">(none)</td>
258+
<td>String</td>
259+
<td>传递自定义 JDBC URL 属性。例如：<code>jdbc.properties.encrypt=false</code>。</td>
260+
</tr>
261+
</tbody>
262+
</table>
263+
</div>
264+
265+
## 可用 Metadata
266+
267+
配置 `metadata.list` 后，以下 metadata 可以传递到下游。
268+
269+
<table class="colwidths-auto docutils">
270+
<thead>
271+
<tr>
272+
<th class="text-left" style="width: 15%">Key</th>
273+
<th class="text-left" style="width: 30%">DataType</th>
274+
<th class="text-left" style="width: 55%">Description</th>
275+
</tr>
276+
</thead>
277+
<tbody>
278+
<tr>
279+
<td>database_name</td>
280+
<td>STRING NOT NULL</td>
281+
<td>包含该行的数据库名称。</td>
282+
</tr>
283+
<tr>
284+
<td>schema_name</td>
285+
<td>STRING NOT NULL</td>
286+
<td>包含该行的 schema 名称。</td>
287+
</tr>
288+
<tr>
289+
<td>table_name</td>
290+
<td>STRING NOT NULL</td>
291+
<td>包含该行的表名。</td>
292+
</tr>
293+
<tr>
294+
<td>op_ts</td>
295+
<td>TIMESTAMP_LTZ(3) NOT NULL</td>
296+
<td>该变更在数据库中发生的时间。对于快照记录，该值始终为 0。</td>
297+
</tr>
298+
</tbody>
299+
</table>
300+
301+
## 启动读取位置
302+
303+
配置项 `scan.startup.mode` 指定 SQL Server CDC 消费者的启动模式。有效值包括：
304+
305+
- `initial`：先对被监控表执行初始快照，然后继续读取最新变更。
306+
- `latest-offset`：从最新 change log offset 开始读取。
307+
- `snapshot`：只读取快照。
308+
- `timestamp`：从 `scan.startup.timestamp-millis` 指定的时间戳开始读取。
309+
310+
## 限制
311+
312+
### 单数据库
313+
314+
`tables` 中的所有条目必须属于同一个字面量数据库。schema 和 table 名称支持正则表达式，但 database 部分不支持正则表达式。
315+
316+
{{< top >}}

‎docs/content/docs/connectors/pipeline-connectors/overview.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@ definition.
4747
| [Oracle]({{< ref "docs/connectors/pipeline-connectors/oracle" >}}) | Source | <li> [Oracle](https://www.oracle.com/database/) | [Oracle](https://mvnrepository.com/artifact/org.apache.flink/flink-cdc-pipeline-connector-oracle) |
4848
| [Paimon]({{< ref "docs/connectors/pipeline-connectors/paimon" >}}) | Sink | <li> [Paimon](https://paimon.apache.org/): 0.6, 0.7, 0.8, 0.9, 1.0, 1.1, 1.2, 1.3 | [Paimon](https://mvnrepository.com/artifact/org.apache.flink/flink-cdc-pipeline-connector-paimon) |
4949
| [Postgres]({{< ref "docs/connectors/pipeline-connectors/postgres" >}}) | Source | <li> [PostgreSQL](https://www.postgresql.org/) | [Postgres](https://mvnrepository.com/artifact/org.apache.flink/flink-cdc-pipeline-connector-postgres) |
50+
| [SQL Server]({{< ref "docs/connectors/pipeline-connectors/sqlserver" >}}) | Source | <li> [SQL Server](https://www.microsoft.com/sql-server/) | [SQL Server](https://mvnrepository.com/artifact/org.apache.flink/flink-cdc-pipeline-connector-sqlserver) |
5051
| [StarRocks]({{< ref "docs/connectors/pipeline-connectors/starrocks" >}}) | Sink | <li> [StarRocks](https://www.starrocks.io/): 2.x, 3.x | [StarRocks](https://mvnrepository.com/artifact/org.apache.flink/flink-cdc-pipeline-connector-starrocks) |
5152

5253

0 commit comments

Comments
 (0)