You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by le...@apache.org on 2024/03/20 08:43:24 UTC
(flink-cdc) branch master updated: [FLINK-34740][cdc][doc-zh] Translate legacy cdc source docs to Chinese
This is an automated email from the ASF dual-hosted git repository.
leonard pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new bacb6a3b6 [FLINK-34740][cdc][doc-zh] Translate legacy cdc source docs to Chinese
bacb6a3b6 is described below
commit bacb6a3b60511856da5e396a9fcf7df217945b8e
Author: Kunni <lv...@alibaba-inc.com>
AuthorDate: Wed Mar 20 16:43:20 2024 +0800
[FLINK-34740][cdc][doc-zh] Translate legacy cdc source docs to Chinese
This closes #3174.
---
.../legacy-flink-cdc-sources/mongodb-cdc.md | 342 +++++-------
.../legacy-flink-cdc-sources/mysql-cdc.md | 608 +++++++++------------
.../legacy-flink-cdc-sources/oceanbase-cdc.md | 359 ++++++------
.../legacy-flink-cdc-sources/overview.md | 6 +-
.../build-real-time-data-lake-tutorial.md | 277 +++++-----
.../tutorials/build-streaming-etl-tutorial.md} | 129 +++--
.../tutorials/db2-tutorial.md | 2 +-
.../tutorials/mongodb-tutorial.md | 60 +-
.../tutorials/oceanbase-tutorial.md | 84 +--
.../tutorials/oracle-tutorial.md | 47 +-
.../tutorials/polardbx-tutorial.md | 261 ++++-----
.../tutorials/sqlserver-tutorial.md | 40 +-
.../tutorials/tidb-tutorial.md | 44 +-
.../legacy-flink-cdc-sources/overview.md | 6 +-
.../build-real-time-data-lake-tutorial.md | 4 +-
.../tutorials/build-streaming-etl-tutorial.md} | 6 +-
.../tutorials/db2-tutorial.md | 2 +-
17 files changed, 1047 insertions(+), 1230 deletions(-)
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/mongodb-cdc.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/mongodb-cdc.md
index 0f078644b..7974cff15 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/mongodb-cdc.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/mongodb-cdc.md
@@ -24,55 +24,61 @@ specific language governing permissions and limitations
under the License.
-->
-# MongoDB CDC Connector
+# MongoDB CDC 连接器
-The MongoDB CDC connector allows for reading snapshot data and incremental data from MongoDB. This document describes how to setup the MongoDB CDC connector to run SQL queries against MongoDB.
+MongoDB CDC 连接器允许从 MongoDB 读取快照数据和增量数据。 本文档描述了如何设置 MongoDB CDC 连接器以针对 MongoDB 运行 SQL 查询。
-Dependencies
+依赖
------------
-In order to setup the MongoDB CDC connector, the following table provides dependency information for both projects using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles.
+为了设置 MongoDB CDC 连接器, 下表提供了使用构建自动化工具（如 Maven 或 SBT ）和带有 SQLJar 捆绑包的 SQLClient 的两个项目的依赖关系信息。
### Maven dependency
-
-{{< artifact flink-connector-mongodb-cdc >}}
+```
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-mongodb-cdc</artifactId>
+ <!-- 请使用已发布的版本依赖，snapshot 版本的依赖需要本地自行编译。 -->
+ <version>3.1-SNAPSHOT</version>
+</dependency>
+```
### SQL Client JAR
-```Download link is available only for stable releases.```
+```下载链接仅适用于稳定版本。```
-Download [flink-sql-connector-mongodb-cdc-3.0-SNAPSHOT.jar](https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-mongodb-cdc/3.0-SNAPSHOT/flink-sql-connector-mongodb-cdc-3.0-SNAPSHOT.jar) and put it under `<FLINK_HOME>/lib/`.
+下载 [flink-sql-connector-mongodb-cdc-3.0-SNAPSHOT.jar](https://repo1.maven.org/maven2/com/ververica/flink-connector-mongodb-cdc/3.0.0/flink-connector-mongodb-cdc-3.0.0.jar) 把它放在 `<FLINK_HOME>/lib/`.
-**Note:** flink-sql-connector-mongodb-cdc-XXX-SNAPSHOT version is the code corresponding to the development branch. Users need to download the source code and compile the corresponding jar. Users should use the released version, such as [flink-sql-connector-mongodb-cdc-3.0.0.jar](https://mvnrepository.com/artifact/com.ververica/flink-sql-connector-mongodb-cdc), the released version will be available in the Maven central warehouse.
+**注意:** flink-sql-connector-mongodb-cdc-XXX-SNAPSHOT 版本是与开发分支相对应的代码。 用户需要下载源代码并编译相应的jar。 用户应使用已发布的版本，例如 [flink-sql-connector-mongodb-cdc-3.0.0.jar](https://mvnrepository.com/artifact/com.ververica/flink-sql-connector-mongodb-cdc), 发布的版本将在 Maven 中央仓库中提供。
-Setup MongoDB
+设置 MongoDB
----------------
-### Availability
-- MongoDB version
+### 可用性
+- MongoDB 版本
- MongoDB version >= 3.6 <br>
-We use [change streams](https://docs.mongodb.com/manual/changeStreams/) feature (new in version 3.6) to capture change data.
+ MongoDB 版本 >= 3.6 <br>
+ 我们使用 [更改流](https://docs.mongodb.com/manual/changeStreams/) 功能（3.6 版中新增），以捕获更改数据。
-- Cluster Deployment
+- 集群部署
- [replica sets](https://docs.mongodb.com/manual/replication/) or [sharded clusters](https://docs.mongodb.com/manual/sharding/) is required.
+ [副本集](https://docs.mongodb.com/manual/replication/) 或者 [分片集群](https://docs.mongodb.com/manual/sharding/) 是必需的。
-- Storage Engine
+- 存储引擎
- [WiredTiger](https://docs.mongodb.com/manual/core/wiredtiger/#std-label-storage-wiredtiger) storage engine is required.
+ [WiredTiger](https://docs.mongodb.com/manual/core/wiredtiger/#std-label-storage-wiredtiger) 存储引擎是必需的。
-- [Replica set protocol version](https://docs.mongodb.com/manual/reference/replica-configuration/#mongodb-rsconf-rsconf.protocolVersion)
+- [副本集协议版本](https://docs.mongodb.com/manual/reference/replica-configuration/#mongodb-rsconf-rsconf.protocolVersion)
- Replica set protocol version 1 [(pv1)](https://docs.mongodb.com/manual/reference/replica-configuration/#mongodb-rsconf-rsconf.protocolVersion) is required. <br>
-Starting in version 4.0, MongoDB only supports pv1. pv1 is the default for all new replica sets created with MongoDB 3.2 or later.
+ 副本集协议版本 1 [(pv1)](https://docs.mongodb.com/manual/reference/replica-configuration/#mongodb-rsconf-rsconf.protocolVersion) 是必需的。 <br>
+ 从 4.0 版本开始，MongoDB 只支持pv1。 pv1 是使用 MongoDB 3.2 或更高版本创建的所有新副本集的默认值。
-- Privileges
+- 权限
- `changeStream` and `read` privileges are required by MongoDB Kafka Connector.
+ `changeStream` and `read` 是 MongoDB Kafka Connector 必需权限。
- You can use the following example for simple authorization.<br>
- For more detailed authorization, please refer to [MongoDB Database User Roles](https://docs.mongodb.com/manual/reference/built-in-roles/#database-user-roles).
+ 你可以使用以下示例进行简单的授权。<br>
+ 有关更详细的授权, 请参照 [MongoDB 数据库用户角色](https://docs.mongodb.com/manual/reference/built-in-roles/#database-user-roles).
```javascript
use admin;
@@ -80,7 +86,7 @@ Starting in version 4.0, MongoDB only supports pv1. pv1 is the default for all n
{
role: "flinkrole",
privileges: [{
- // Grant privileges on all non-system collections in all databases
+ // 所有数据库中所有非系统集合的 grant 权限
resource: { db: "", collection: "" },
actions: [
"splitVector",
@@ -91,8 +97,8 @@ Starting in version 4.0, MongoDB only supports pv1. pv1 is the default for all n
"changeStream" ]
}],
roles: [
- // Read config.collections and config.chunks
- // for sharded cluster snapshot splitting.
+ // 阅读 config.collections 和 config.chunks
+ // 用于分片集群快照拆分。
{ role: 'read', db: 'config' }
]
}
@@ -110,20 +116,20 @@ Starting in version 4.0, MongoDB only supports pv1. pv1 is the default for all n
```
-How to create a MongoDB CDC table
+如何创建 MongoDB CDC 表
----------------
-The MongoDB CDC table can be defined as following:
+MongoDB CDC 表可以定义如下：
```sql
--- register a MongoDB table 'products' in Flink SQL
+-- 在 Flink SQL 中注册 MongoDB 表 `products`
CREATE TABLE products (
- _id STRING, // must be declared
+ _id STRING, // 必须声明
name STRING,
weight DECIMAL(10,3),
tags ARRAY<STRING>, -- array
- price ROW<amount DECIMAL(10,2), currency STRING>, -- embedded document
- suppliers ARRAY<ROW<name STRING, address STRING>>, -- embedded documents
+ price ROW<amount DECIMAL(10,2), currency STRING>, -- 嵌入式文档
+ suppliers ARRAY<ROW<name STRING, address STRING>>, -- 嵌入式文档
PRIMARY KEY(_id) NOT ENFORCED
) WITH (
'connector' = 'mongodb-cdc',
@@ -134,17 +140,17 @@ CREATE TABLE products (
'collection' = 'products'
);
--- read snapshot and change events from products collection
+-- 从 `products` 集合中读取快照和更改事件
SELECT * FROM products;
```
-**Note that**
+**请注意**
-MongoDB's change event record doesn't have updated before message. So, we can only convert it to Flink's UPSERT changelog stream.
-An upsert stream requires a unique key, so we must declare `_id` as primary key.
-We can't declare other column as primary key, because delete operation does not contain the key and value besides `_id` and `sharding key`.
+MongoDB 的更改事件记录在消息之前没有更新。因此，我们只能将其转换为 Flink 的 UPSERT 更改日志流。
+upstart 流需要一个唯一的密钥，所以我们必须声明 `_id` 作为主键。
+我们不能将其他列声明为主键, 因为删除操作不包含除 `_id` 和 `sharding key` 之外的键和值。
-Connector Options
+连接器选项
----------------
<div class="highlight">
@@ -164,21 +170,21 @@ Connector Options
<td>required</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Specify what connector to use, here should be <code>mongodb-cdc</code>.</td>
+ <td>指定要使用的连接器，此处应为 <code>mongodb-cdc</code>.</td>
</tr>
<tr>
<td>scheme</td>
<td>optional</td>
<td style="word-wrap: break-word;">mongodb</td>
<td>String</td>
- <td>The protocol connected to MongoDB. eg. <code>mongodb or mongodb+srv.</code></td>
+ <td>指定 MongoDB 连接协议。 eg. <code>mongodb or mongodb+srv.</code></td>
</tr>
<tr>
<td>hosts</td>
<td>required</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>The comma-separated list of hostname and port pairs of the MongoDB servers.<br>
+ <td>MongoDB 服务器的主机名和端口对的逗号分隔列表。<br>
eg. <code>localhost:27017,localhost:27018</code>
</td>
</tr>
@@ -187,8 +193,8 @@ Connector Options
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Name of the database user to be used when connecting to MongoDB.<br>
- This is required only when MongoDB is configured to use authentication.
+ <td>连接到 MongoDB 时要使用的数据库用户的名称。<br>
+ 只有当 MongoDB 配置为使用身份验证时，才需要这样做。
</td>
</tr>
<tr>
@@ -196,8 +202,8 @@ Connector Options
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Password to be used when connecting to MongoDB.<br>
- This is required only when MongoDB is configured to use authentication.
+ <td>连接到 MongoDB 时要使用的密码。<br>
+ 只有当 MongoDB 配置为使用身份验证时，才需要这样做。
</td>
</tr>
<tr>
@@ -205,23 +211,23 @@ Connector Options
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Name of the database to watch for changes. If not set then all databases will be captured. <br>
- The database also supports regular expressions to monitor multiple databases matching the regular expression.</td>
+ <td>要监视更改的数据库的名称。 如果未设置，则将捕获所有数据库。 <br>
+ 该数据库还支持正则表达式来监视与正则表达式匹配的多个数据库。</td>
</tr>
<tr>
<td>collection</td>
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Name of the collection in the database to watch for changes. If not set then all collections will be captured.<br>
- The collection also supports regular expressions to monitor multiple collections matching fully-qualified collection identifiers.</td>
+ <td>数据库中要监视更改的集合的名称。 如果未设置，则将捕获所有集合。<br>
+ 该集合还支持正则表达式来监视与完全限定的集合标识符匹配的多个集合。</td>
</tr>
<tr>
<td>connection.options</td>
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>The ampersand-separated <a href="https://docs.mongodb.com/manual/reference/connection-string/#std-label-connections-connection-options">connection options</a> of MongoDB. eg. <br>
+ <td><a href="https://docs.mongodb.com/manual/reference/connection-string/#std-label-connections-connection-options">MongoDB连接选项</a>。 例如: <br>
<code>replicaSet=test&connectTimeoutMS=300000</code>
</td>
</tr>
@@ -230,108 +236,77 @@ Connector Options
<td>optional</td>
<td style="word-wrap: break-word;">initial</td>
<td>String</td>
- <td>Optional startup mode for MongoDB CDC consumer, valid enumerations are "initial", "latest-offset" and "timestamp".
- Please see <a href="#startup-reading-position">Startup Reading Position</a> section for more detailed information.</td>
+ <td> MongoDB CDC 消费者可选的启动模式，
+ 合法的模式为 "initial"，"latest-offset" 和 "timestamp"。
+ 请查阅 <a href="#a-name-id-002-a">启动模式</a> 章节了解更多详细信息。</td>
</tr>
<tr>
<td>scan.startup.timestamp-millis</td>
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>Long</td>
- <td>Timestamp in millis of the start point, only used for <code>'timestamp'</code> startup mode.</td>
- </tr>
- <tr>
- <td>copy.existing.queue.size</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">10240</td>
- <td>Integer</td>
- <td>The max size of the queue to use when copying data.</td>
+ <td>起始毫秒数, 仅适用于 <code>'timestamp'</code> 启动模式.</td>
</tr>
<tr>
<td>batch.size</td>
<td>optional</td>
<td style="word-wrap: break-word;">1024</td>
<td>Integer</td>
- <td>The cursor batch size.</td>
+ <td>Cursor 批次大小。</td>
</tr>
<tr>
<td>poll.max.batch.size</td>
<td>optional</td>
<td style="word-wrap: break-word;">1024</td>
<td>Integer</td>
- <td>Maximum number of change stream documents to include in a single batch when polling for new data.</td>
+ <td>轮询新数据时，单个批处理中要包含的更改流文档的最大数量。</td>
</tr>
<tr>
<td>poll.await.time.ms</td>
<td>optional</td>
<td style="word-wrap: break-word;">1000</td>
<td>Integer</td>
- <td>The amount of time to wait before checking for new results on the change stream.</td>
+ <td>在更改流上检查新结果之前等待的时间。</td>
</tr>
<tr>
<td>heartbeat.interval.ms</td>
<td>optional</td>
<td style="word-wrap: break-word;">0</td>
<td>Integer</td>
- <td>The length of time in milliseconds between sending heartbeat messages. Use 0 to disable.</td>
- </tr>
- <tr>
- <td>scan.full-changelog</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">false</td>
- <td>Boolean</td>
- <td>Whether try to generate full-mode changelog based on pre- and post-images in MongoDB. Refer to <a href="#a-name-id-003-a">Full Changelog</a> for more details. Supports MongoDB 6.0 and above only.</td>
+ <td>心跳间隔（毫秒）。使用 0 禁用。</td>
</tr>
<tr>
<td>scan.incremental.snapshot.enabled</td>
<td>optional</td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
- <td>Whether enable incremental snapshot. The incremental snapshot feature only supports after MongoDB 4.0.</td>
+ <td>是否启用增量快照。增量快照功能仅支持 MongoDB 4.0 之后的版本。</td>
</tr>
<tr>
<td>scan.incremental.snapshot.chunk.size.mb</td>
<td>optional</td>
<td style="word-wrap: break-word;">64</td>
<td>Integer</td>
- <td>The chunk size mb of incremental snapshot.</td>
- </tr>
- <tr>
- <td>scan.incremental.snapshot.chunk.samples</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">20</td>
- <td>Integer</td>
- <td>The samples count per chunk when using sample partition strategy during incremental snapshot.</td>
+ <td>增量快照的区块大小 mb。</td>
</tr>
<tr>
<td>scan.incremental.close-idle-reader.enabled</td>
<td>optional</td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
- <td>Whether to close idle readers at the end of the snapshot phase. <br>
- The flink version is required to be greater than or equal to 1.14 when 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' is set to true.<br>
- If the flink version is greater than or equal to 1.15, the default value of 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' has been changed to true,
- so it does not need to be explicitly configured 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = 'true'
- </td>
- </tr>
- <tr>
- <td>scan.cursor.no-timeout</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">true</td>
- <td>Boolean</td>
- <td>MongoDB server normally times out idle cursors after an inactivity period (10 minutes) to prevent excess memory use. Set this option to true to prevent that. Only available when parallelism snapshot is enabled.</td>
+ <td>是否在快照结束后关闭空闲的 Reader。 此特性需要 flink 版本大于等于 1.14 并且 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' 需要设置为 true。</td>
</tr>
</tbody>
</table>
</div>
-Note: `heartbeat.interval.ms` is highly recommended setting a proper value larger than 0 **if the collection changes slowly**.
-The heartbeat event can push the `resumeToken` forward to avoid `resumeToken` being expired when we recover the Flink job from a checkpoint or savepoint.
+注意: `heartbeat.interval.ms` 强烈建议设置一个大于 0 的适当值 **如果集合更改缓慢**.
+当我们从检查点或保存点恢复 Flink 作业时，心跳事件可以向前推送 `resumeToken`，以避免 `resumeToken` 过期。
-Available Metadata
+可用元数据
----------------
-The following format metadata can be exposed as read-only (VIRTUAL) columns in a table definition.
+以下格式元数据可以在表定义中公开为只读（VIRTUAL）列。
<table class="colwidths-auto docutils">
<thead>
@@ -345,33 +320,33 @@ The following format metadata can be exposed as read-only (VIRTUAL) columns in a
<tr>
<td>database_name</td>
<td>STRING NOT NULL</td>
- <td>Name of the database that contain the row.</td>
+ <td>包含该行的数据库的名称。</td>
</tr>
<tr>
<td>collection_name</td>
<td>STRING NOT NULL</td>
- <td>Name of the collection that contain the row.</td>
+ <td>包含该行的集合的名称。</td>
</tr>
<tr>
<td>op_ts</td>
<td>TIMESTAMP_LTZ(3) NOT NULL</td>
- <td>It indicates the time that the change was made in the database. <br>If the record is read from snapshot of the table instead of the change stream, the value is always 0.</td>
+ <td>它指示在数据库中进行更改的时间。 <br>如果记录是从表的快照而不是改变流中读取的，该值将始终为0。</td>
</tr>
</tbody>
</table>
-The extended CREATE TABLE example demonstrates the syntax for exposing these metadata fields:
+扩展的 CREATE TABLE 示例演示了用于公开这些元数据字段的语法：
```sql
CREATE TABLE products (
db_name STRING METADATA FROM 'database_name' VIRTUAL,
collection_name STRING METADATA FROM 'collection_name' VIRTUAL,
operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
- _id STRING, // must be declared
+ _id STRING, // 必须声明
name STRING,
weight DECIMAL(10,3),
tags ARRAY<STRING>, -- array
- price ROW<amount DECIMAL(10,2), currency STRING>, -- embedded document
- suppliers ARRAY<ROW<name STRING, address STRING>>, -- embedded documents
+ price ROW<amount DECIMAL(10,2), currency STRING>, -- 嵌入式文档
+ suppliers ARRAY<ROW<name STRING, address STRING>>, -- 嵌入式文档
PRIMARY KEY(_id) NOT ENFORCED
) WITH (
'connector' = 'mongodb-cdc',
@@ -383,23 +358,22 @@ CREATE TABLE products (
);
```
-Features
+特性
--------
-### Exactly-Once Processing
+### 精确一次处理
-The MongoDB CDC connector is a Flink Source connector which will read database snapshot first and then continues to read change stream events with **exactly-once processing** even failures happen.
+MongoDB CDC 连接器是一个 Flink Source 连接器，它将首先读取数据库快照，然后在处理**甚至失败时继续读取带有**的更改流事件。
-### Startup Reading Position
+### 启动模式<a name="启动模式" id="002" ></a>
-The config option `scan.startup.mode` specifies the startup mode for MongoDB CDC consumer. The valid enumerations are:
+配置选项```scan.startup.mode```指定 MySQL CDC 使用者的启动模式。有效枚举包括：
-- `initial` (default): Performs an initial snapshot on the monitored database tables upon first startup, and continue to read the latest oplog.
-- `latest-offset`: Never to perform snapshot on the monitored database tables upon first startup, just read from
- the end of the oplog which means only have the changes since the connector was started.
-- `timestamp`: Skip snapshot phase and start reading oplog events from a specific timestamp.
+- `initial` （默认）：在第一次启动时对受监视的数据库表执行初始快照，并继续读取最新的 oplog。
+- `latest-offset`：首次启动时，从不对受监视的数据库表执行快照， 连接器仅从 oplog 的结尾处开始读取，这意味着连接器只能读取在连接器启动之后的数据更改。
+- `timestamp`：跳过快照阶段，从指定的时间戳开始读取 oplog 事件。
-For example in DataStream API:
+例如使用 DataStream API:
```java
MongoDBSource.builder()
.startupOptions(StartupOptions.latest()) // Start from latest offset
@@ -412,45 +386,49 @@ and with SQL:
```SQL
CREATE TABLE mongodb_source (...) WITH (
'connector' = 'mongodb-cdc',
- 'scan.startup.mode' = 'latest-offset', -- Start from latest offset
+ 'scan.startup.mode' = 'latest-offset', -- 从最晚位点启动
...
- 'scan.startup.mode' = 'timestamp', -- Start from timestamp
- 'scan.startup.timestamp-millis' = '1667232000000' -- Timestamp under timestamp startup mode
+ 'scan.incremental.snapshot.enabled' = 'true', -- 指定时间戳启动，需要开启增量快照读
+ 'scan.startup.mode' = 'timestamp', -- 指定时间戳启动模式
+ 'scan.startup.timestamp-millis' = '1667232000000' -- 启动毫秒时间
...
)
```
-### Change Streams
+**Notes:**
+- 'timestamp' 指定时间戳启动模式，需要开启增量快照读。
+
+### 更改流
-We integrate the [MongoDB's official Kafka Connector](https://docs.mongodb.com/kafka-connector/current/kafka-source/) to read snapshot or change events from MongoDB and drive it by Debezium's `EmbeddedEngine`.
+我们将 [MongoDB's official Kafka Connector](https://docs.mongodb.com/kafka-connector/current/kafka-source/) 从 MongoDB 中读取快照或更改事件，并通过 Debezium 的 `EmbeddedEngine` 进行驱动。
-Debezium's `EmbeddedEngine` provides a mechanism for running a single Kafka Connect `SourceConnector` within an application's process, and it can drive any standard Kafka Connect `SourceConnector` properly even which is not provided by Debezium.
+Debezium 的 `EmbeddedEngine` 提供了一种在应用程序进程中运行单个 Kafka Connect `SourceConnector` 的机制，并且它可以正确地驱动任何标准的 Kafka Connect `SourceConnector`，即使它不是由 Debezium 提供的。
-We choose **MongoDB's official Kafka Connector** instead of the **Debezium's MongoDB Connector** because they use a different change data capture mechanism.
+我们选择 **MongoDB 的官方 Kafka连接器**，而不是 **Debezium 的MongoDB 连接器**，因为它们使用了不同的更改数据捕获机制。
-- For Debezium's MongoDB Connector, it reads the `oplog.rs` collection of each replica-set's master node.
-- For MongoDB's Kafka Connector, it subscribes `Change Stream` of MongoDB.
+- 对于 Debezium 的 MongoDB 连接器，它读取每个复制集主节点的 `oplog.rs` 集合。
+- 对于 MongoDB 的 Kafka 连接器，它订阅了 MongoDB 的 `更改流`。
-MongoDB's `oplog.rs` collection doesn't keep the changed record's update before state, so it's hard to extract the full document state by a single `oplog.rs` record and convert it to change log stream accepted by Flink (Insert Only, Upsert, All).
-Additionally, MongoDB 5 (released in July 2021) has changed the oplog format, so the current Debezium connector cannot be used with it.
+MongoDB 的`oplog.rs` 集合没有在状态之前保持更改记录的更新， 因此，很难通过单个 `oplog.rs` 记录提取完整的文档状态，并将其转换为 Flink 接受的更改日志流（Insert Only，Upsert，All）。
+此外，MongoDB 5（2021 7月发布）改变了 oplog 格式，因此当前的 Debezium 连接器不能与其一起使用。
-**Change Stream** is a new feature provided by MongoDB 3.6 for replica sets and sharded clusters that allows applications to access real-time data changes without the complexity and risk of tailing the oplog.<br>
-Applications can use change streams to subscribe to all data changes on a single collection, a database, or an entire deployment, and immediately react to them.
+**Change Stream**是 MongoDB 3.6 为副本集和分片集群提供的一项新功能，它允许应用程序访问实时数据更改，而不会带来跟踪操作日志的复杂性和风险。<br>
+应用程序可以使用更改流来订阅单个集合上的所有数据更改， 数据库或整个部署，并立即对其做出反应。
-**Lookup Full Document for Update Operations** is a feature provided by **Change Stream** which can configure the change stream to return the most current majority-committed version of the updated document. Because of this feature, we can easily collect the latest full document and convert the change log to Flink's **Upsert Changelog Stream**.
+**查找更新操作的完整文档**是**变更流**提供的一项功能，它可以配置变更流以返回更新文档的最新多数提交版本。由于该功能，我们可以轻松收集最新的完整文档，并将更改日志转换为 Flink 的**Upsert Changelog Stream**。
-By the way, Debezium's MongoDB change streams exploration mentioned by [DBZ-435](https://issues.redhat.com/browse/DBZ-435) is on roadmap.<br>
-If it's done, we can consider integrating two kinds of source connector for users to choose.
+顺便说一句，[DBZ-435](https://issues.redhat.com/browse/DBZ-435)提到的Debezium的MongoDB变更流探索,正在制定路线图。<br>
+如果完成了，我们可以考虑集成两种源连接器供用户选择。
### DataStream Source
-The MongoDB CDC connector can also be a DataStream source. You can create a SourceFunction as the following shows:
+MongoDB CDC 连接器也可以是一个数据流源。 你可以创建 SourceFunction，如下所示：
```java
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
-import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
-import org.apache.flink.cdc.connectors.mongodb.MongoDBSource;
+import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
+import com.ververica.cdc.connectors.mongodb.MongoDBSource;
public class MongoDBSourceExample {
public static void main(String[] args) throws Exception {
@@ -458,44 +436,44 @@ public class MongoDBSourceExample {
.hosts("localhost:27017")
.username("flink")
.password("flinkpw")
- .databaseList("inventory") // set captured database, support regex
- .collectionList("inventory.products", "inventory.orders") //set captured collections, support regex
+ .databaseList("inventory") // 设置捕获的数据库，支持正则表达式
+ .collectionList("inventory.products", "inventory.orders") //设置捕获的集合，支持正则表达式
.deserializer(new JsonDebeziumDeserializationSchema())
.build();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.addSource(sourceFunction)
- .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering
+ .print().setParallelism(1); // 对 sink 使用并行度 1 以保持消息顺序
env.execute();
}
}
```
-The MongoDB CDC incremental connector (after 2.3.0) can be used as the following shows:
+MongoDB CDC 增量连接器（2.3.0 之后）可以使用，如下所示：
```java
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.cdc.connectors.mongodb.source.MongoDBSource;
-import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
+import com.ververica.cdc.connectors.mongodb.source.MongoDBSource;
+import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
public class MongoDBIncrementalSourceExample {
public static void main(String[] args) throws Exception {
MongoDBSource<String> mongoSource =
MongoDBSource.<String>builder()
.hosts("localhost:27017")
- .databaseList("inventory") // set captured database, support regex
- .collectionList("inventory.products", "inventory.orders") //set captured collections, support regex
+ .databaseList("inventory") // 设置捕获的数据库，支持正则表达式
+ .collectionList("inventory.products", "inventory.orders") //设置捕获的集合，支持正则表达式
.username("flink")
.password("flinkpw")
.deserializer(new JsonDebeziumDeserializationSchema())
.build();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- // enable checkpoint
+ // 启用检查点
env.enableCheckpointing(3000);
- // set the source parallelism to 2
+ // 将 source 并行度设置为 2
env.fromSource(mongoSource, WatermarkStrategy.noWatermarks(), "MongoDBIncrementalSource")
.setParallelism(2)
.print()
@@ -506,72 +484,19 @@ public class MongoDBIncrementalSourceExample {
}
```
-**Note:**
-- If database regex is used, `readAnyDatabase` role is required.
-- The incremental snapshot feature only supports after MongoDB 4.0.
-
-### Full Changelog<a name="Full Changelog" id="003" ></a>
-
-MongoDB 6.0 and above supports emitting change stream events containing document before and after the change was made (aka. pre- and post-images).
-
-- The pre-image is the document before it was replaced, updated, or deleted. There is no pre-image for an inserted document.
+**注意:**
+- 如果使用数据库正则表达式，则需要 `readAnyDatabase` 角色。
+- 增量快照功能仅支持 MongoDB 4.0 之后的版本。
-- The post-image is the document after it was inserted, replaced, or updated. There is no post-image for a deleted document.
-
-MongoDB CDC could make uses of pre-image and post-images to generate full-mode changelog stream including Insert, Update Before, Update After, and Delete data rows, thereby avoiding additional `ChangelogNormalize` downstream node.
-
-To enable this feature, here's some prerequisites:
-
-- MongoDB version must be 6.0 or above;
-- Enable `preAndPostImages` feature at the database level:
-```javascript
-db.runCommand({
- setClusterParameter: {
- changeStreamOptions: {
- preAndPostImages: {
- expireAfterSeconds: 'off' // replace with custom image expiration time
- }
- }
- }
-})
-```
-- Enable `changeStreamPreAndPostImages` feature for collections to be monitored:
-```javascript
-db.runCommand({
- collMod: "<< collection name >>",
- changeStreamPreAndPostImages: {
- enabled: true
- }
-})
-```
-- Enable MongoDB CDC's `scan.full-changelog` feature:
-
-```java
-MongoDBSource.builder()
- .scanFullChangelog(true)
- ...
- .build()
-```
-
-or with Flink SQL:
-
-```SQL
-CREATE TABLE mongodb_source (...) WITH (
- 'connector' = 'mongodb-cdc',
- 'scan.full-changelog' = 'true',
- ...
-)
-```
-
-Data Type Mapping
+数据类型映射
----------------
-[BSON](https://docs.mongodb.com/manual/reference/bson-types/) short for **Binary JSON** is a binary-encoded serialization of JSON-like format used to store documents and make remote procedure calls in MongoDB.
+[BSON](https://docs.mongodb.com/manual/reference/bson-types/) **二进制 JSON**的缩写是一种类似 JSON 格式的二进制编码序列，用于在 MongoDB 中存储文档和进行远程过程调用。
-[Flink SQL Data Type](https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/types/) is similar to the SQL standard’s data type terminology which describes the logical type of a value in the table ecosystem. It can be used to declare input and/or output types of operations.
+[Flink SQL Data Type](https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/types/) 类似于 SQL 标准的数据类型术语，该术语描述了表生态系统中值的逻辑类型。它可以用于声明操作的输入和/或输出类型。
-In order to enable Flink SQL to process data from heterogeneous data sources, the data types of heterogeneous data sources need to be uniformly converted to Flink SQL data types.
+为了使 Flink SQL 能够处理来自异构数据源的数据，异构数据源的数据类型需要统一转换为 Flink SQL 数据类型。
-The following is the mapping of BSON type and Flink SQL type.
+以下是 BSON 类型和 Flink SQL 类型的映射。
<div class="wy-table-responsive">
@@ -674,10 +599,8 @@ The following is the mapping of BSON type and Flink SQL type.
</table>
</div>
-
-Reference
+参考
--------
-
- [MongoDB Kafka Connector](https://docs.mongodb.com/kafka-connector/current/kafka-source/)
- [Change Streams](https://docs.mongodb.com/manual/changeStreams/)
- [Replication](https://docs.mongodb.com/manual/replication/)
@@ -686,8 +609,7 @@ Reference
- [WiredTiger](https://docs.mongodb.com/manual/core/wiredtiger/#std-label-storage-wiredtiger)
- [Replica set protocol](https://docs.mongodb.com/manual/reference/replica-configuration/#mongodb-rsconf-rsconf.protocolVersion)
- [Connection String Options](https://docs.mongodb.com/manual/reference/connection-string/#std-label-connections-connection-options)
-- [Document Pre- and Post-Images](https://www.mongodb.com/docs/v6.0/changeStreams/#change-streams-with-document-pre--and-post-images)
- [BSON Types](https://docs.mongodb.com/manual/reference/bson-types/)
-- [Flink DataTypes](https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/types/)
+- [Flink DataTypes](https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/table/types/)
{{< top >}}
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/mysql-cdc.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/mysql-cdc.md
index 50d148ca7..2fd2a5843 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/mysql-cdc.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/mysql-cdc.md
@@ -24,88 +24,94 @@ specific language governing permissions and limitations
under the License.
-->
-# MySQL CDC Connector
+# MySQL CDC 连接器
-The MySQL CDC connector allows for reading snapshot data and incremental data from MySQL database. This document describes how to setup the MySQL CDC connector to run SQL queries against MySQL databases.
+MySQL CDC 连接器允许从 MySQL 数据库读取快照数据和增量数据。本文描述了如何设置 MySQL CDC 连接器来对 MySQL 数据库运行 SQL 查询。
+## 支持的数据库
-## Supported Databases
+| Connector | Database | Driver |
+|----------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------------------|
+| mysql-cdc | <li> [MySQL](https://dev.mysql.com/doc): 5.6, 5.7, 8.0.x <li> [RDS MySQL](https://www.aliyun.com/product/rds/mysql): 5.6, 5.7, 8.0.x <li> [PolarDB MySQL](https://www.aliyun.com/product/polardb): 5.6, 5.7, 8.0.x <li> [Aurora MySQL](https://aws.amazon.com/cn/rds/aurora): 5.6, 5.7, 8.0.x <li> [MariaDB](https://mariadb.org): 10.x <li> [PolarDB X](https://github.com/ApsaraDB/galaxysql): 2.0.1 | JDBC Driver: 8.0.27 |
-| Connector | Database | Driver |
-|---------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------------------|
-| [mysql-cdc](../mysql-cdc) | <li> [MySQL](https://dev.mysql.com/doc): 5.6, 5.7, 8.0.x <li> [RDS MySQL](https://www.aliyun.com/product/rds/mysql): 5.6, 5.7, 8.0.x <li> [PolarDB MySQL](https://www.aliyun.com/product/polardb): 5.6, 5.7, 8.0.x <li> [Aurora MySQL](https://aws.amazon.com/cn/rds/aurora): 5.6, 5.7, 8.0.x <li> [MariaDB](https://mariadb.org): 10.x <li> [PolarDB X](https://github.com/ApsaraDB/galaxysql): 2.0.1 | JDBC Driver: 8.0.27 |
-
-Dependencies
+依赖
------------
-In order to setup the MySQL CDC connector, the following table provides dependency information for both projects using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles.
+为了设置 MySQL CDC 连接器，下表提供了使用构建自动化工具（如 Maven 或 SBT ）和带有 SQL JAR 包的 SQL 客户端的两个项目的依赖关系信息。
### Maven dependency
-{{< artifact flink-connector-mysql-cdc >}}
+```
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-mysql-cdc</artifactId>
+ <!-- 请使用已发布的版本依赖，snapshot 版本的依赖需要本地自行编译。 -->
+ <version>3.1-SNAPSHOT</version>
+</dependency>
+```
### SQL Client JAR
-```Download link is available only for stable releases.```
+```下载链接仅在已发布版本可用，请在文档网站左下角选择浏览已发布的版本。```
-Download flink-sql-connector-mysql-cdc-3.0-SNAPSHOT.jar and put it under `<FLINK_HOME>/lib/`.
+下载 [flink-sql-connector-mysql-cdc-3.0.0.jar](https://repo1.maven.org/maven2/com/ververica/flink-connector-mysql-cdc/3.0.0/flink-connector-mysql-cdc-3.0.0.jar) 到 `<FLINK_HOME>/lib/` 目录下。
-**Note:** flink-sql-connector-mysql-cdc-XXX-SNAPSHOT version is the code corresponding to the development branch. Users should use the released version, such as [flink-sql-connector-mysql-cdc-3.0.0.jar](https://mvnrepository.com/artifact/com.ververica/flink-connector-mysql-cdc), the released version will be available in the Maven central warehouse.
+**注意:** flink-sql-connector-mysql-cdc-XXX-SNAPSHOT 版本是开发分支`release-XXX`对应的快照版本，快照版本用户需要下载源代码并编译相应的 jar。用户应使用已经发布的版本，例如 [flink-sql-connector-mysql-cdc-3.0.0.jar](https://mvnrepository.com/artifact/com.ververica/flink-connector-mysql-cdc) 当前已发布的所有版本都可以在 Maven 中央仓库获取。
-Setup MySQL server
+配置 MySQL 服务器
----------------
-You have to define a MySQL user with appropriate permissions on all databases that the Debezium MySQL connector monitors.
+你必须定义一个 MySQL 用户，该用户对 MySQL CDC 连接器监视的所有数据库都应该具有所需的权限。
-1. Create the MySQL user:
+1. 创建 MySQL 用户：
```sql
mysql> CREATE USER 'user'@'localhost' IDENTIFIED BY 'password';
```
-2. Grant the required permissions to the user:
+2. 向用户授予所需的权限：
```sql
mysql> GRANT SELECT, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'user' IDENTIFIED BY 'password';
```
-**Note:** The RELOAD permissions is not required any more when `scan.incremental.snapshot.enabled` is enabled (enabled by default).
+**注意:** 在 `scan.incremental.snapshot.enabled` 参数已启用时（默认情况下已启用）时，不再需要授予 reload 权限。
-3. Finalize the user’s permissions:
+3. 刷新用户权限：
```sql
mysql> FLUSH PRIVILEGES;
```
-See more about the [permission explanation](https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-creating-user).
+查看更多用户权限问题请参考 [权限说明](https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-creating-user).
-Notes
+注意事项
----------------
-### Set a different SERVER ID for each reader
+### 为每个 Reader 设置不同的 Server id
-Every MySQL database client for reading binlog should have a unique id, called server id. MySQL server will use this id to maintain network connection and the binlog position. Therefore, if different jobs share a same server id, it may result to read from wrong binlog position.
-Thus, it is recommended to set different server id for each reader via the [SQL Hints](https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/hints.html),
-e.g. assuming the source parallelism is 4, then we can use `SELECT * FROM source_table /*+ OPTIONS('server-id'='5401-5404') */ ;` to assign unique server id for each of the 4 source readers.
+每个用于读取 binlog 的 MySQL 数据库客户端都应该有一个唯一的 id，称为 Server id。 MySQL 服务器将使用此 id 来维护网络连接和 binlog 位置。 因此，如果不同的作业共享相同的 Server id， 则可能导致从错误的 binlog 位置读取数据。
+因此，建议通过为每个 Reader 设置不同的 Server id [SQL Hints](https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/sql/hints.html),
+假设 Source 并行度为 4, 我们可以使用 `SELECT * FROM source_table /*+ OPTIONS('server-id'='5401-5404') */ ;` 来为 4 个 Source readers 中的每一个分配唯一的 Server id。
-### Setting up MySQL session timeouts
+### 设置 MySQL 会话超时时间
-When an initial consistent snapshot is made for large databases, your established connection could timeout while the tables are being read. You can prevent this behavior by configuring interactive_timeout and wait_timeout in your MySQL configuration file.
-- `interactive_timeout`: The number of seconds the server waits for activity on an interactive connection before closing it. See [MySQL documentations](https://dev.mysql.com/doc/refman/8.0/en/server-system-variables.html#sysvar_interactive_timeout).
-- `wait_timeout`: The number of seconds the server waits for activity on a noninteractive connection before closing it. See [MySQL documentations](https://dev.mysql.com/doc/refman/8.0/en/server-system-variables.html#sysvar_wait_timeout).
+当为大型数据库创建初始一致快照时，你建立的连接可能会在读取表时碰到超时问题。你可以通过在 MySQL 侧配置 interactive_timeout 和 wait_timeout 来缓解此类问题。
+- `interactive_timeout`: 服务器在关闭交互连接之前等待活动的秒数。 更多信息请参考 [MySQL documentations](https://dev.mysql.com/doc/refman/8.0/en/server-system-variables.html#sysvar_interactive_timeout).
+- `wait_timeout`: 服务器在关闭非交互连接之前等待活动的秒数。 更多信息请参考 [MySQL documentations](https://dev.mysql.com/doc/refman/8.0/en/server-system-variables.html#sysvar_wait_timeout).
-How to create a MySQL CDC table
+如何创建 MySQL CDC 表
----------------
-The MySQL CDC table can be defined as following:
+MySQL CDC 表可以定义如下：
```sql
--- checkpoint every 3000 milliseconds
+-- 每 3 秒做一次 checkpoint，用于测试，生产配置建议5到10分钟
Flink SQL> SET 'execution.checkpointing.interval' = '3s';
--- register a MySQL table 'orders' in Flink SQL
+-- 在 Flink SQL中注册 MySQL 表 'orders'
Flink SQL> CREATE TABLE orders (
order_id INT,
order_date TIMESTAMP(0),
@@ -123,11 +129,11 @@ Flink SQL> CREATE TABLE orders (
'database-name' = 'mydb',
'table-name' = 'orders');
--- read snapshot and binlogs from orders table
+-- 从订单表读取全量数据(快照)和增量数据(binlog)
Flink SQL> SELECT * FROM orders;
```
-Connector Options
+连接器选项
----------------
<div class="highlight">
@@ -147,62 +153,57 @@ Connector Options
<td>required</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Specify what connector to use, here should be <code>'mysql-cdc'</code>.</td>
+ <td>指定要使用的连接器, 这里应该是 <code>'mysql-cdc'</code>.</td>
</tr>
<tr>
<td>hostname</td>
<td>required</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>IP address or hostname of the MySQL database server.</td>
+ <td> MySQL 数据库服务器的 IP 地址或主机名。</td>
</tr>
<tr>
<td>username</td>
<td>required</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Name of the MySQL database to use when connecting to the MySQL database server.</td>
+ <td>连接到 MySQL 数据库服务器时要使用的 MySQL 用户的名称。</td>
</tr>
<tr>
<td>password</td>
<td>required</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Password to use when connecting to the MySQL database server.</td>
+ <td>连接 MySQL 数据库服务器时使用的密码。</td>
</tr>
<tr>
<td>database-name</td>
<td>required</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Database name of the MySQL server to monitor. The database-name also supports regular expressions to monitor multiple tables matches the regular expression.</td>
+ <td>要监视的 MySQL 服务器的数据库名称。数据库名称还支持正则表达式，以监视多个与正则表达式匹配的表。</td>
</tr>
<tr>
<td>table-name</td>
<td>required</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>
- Table name of the MySQL database to monitor. The table-name also supports regular expressions to monitor multiple tables that satisfy the regular expressions. Note: When the MySQL CDC connector regularly matches the table name, it will concat the database-name and table-name filled in by the user through the string `\\.` to form a full-path regular expression, and then use the regular expression to match the fully qualified name of the table in the MySQL database.
- </td>
+ <td>需要监视的 MySQL 数据库的表名。表名支持正则表达式，以监视满足正则表达式的多个表。注意：MySQL CDC 连接器在正则匹配表名时，会把用户填写的 database-name， table-name 通过字符串 `\\.` 连接成一个全路径的正则表达式，然后使用该正则表达式和 MySQL 数据库中表的全限定名进行正则匹配。</td>
</tr>
<tr>
<td>port</td>
<td>optional</td>
<td style="word-wrap: break-word;">3306</td>
<td>Integer</td>
- <td>Integer port number of the MySQL database server.</td>
+ <td> MySQL 数据库服务器的整数端口号。</td>
</tr>
<tr>
<td>server-id</td>
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>A numeric ID or a numeric ID range of this database client, The numeric ID syntax is like '5400',
- the numeric ID range syntax is like '5400-5408', The numeric ID range syntax is recommended when 'scan.incremental.snapshot.enabled' enabled.
- Every ID must be unique across all currently-running database processes in the MySQL cluster. This connector joins the MySQL cluster
- as another server (with this unique ID) so it can read the binlog. By default, a random number is generated between 5400 and 6400,
- though we recommend setting an explicit value.
+ <td>读取数据使用的 server id，server id 可以是个整数或者一个整数范围，比如 '5400' 或 '5400-5408',
+ 建议在 'scan.incremental.snapshot.enabled' 参数为启用时，配置成整数范围。因为在当前 MySQL 集群中运行的所有 slave 节点，标记每个 salve 节点的 id 都必须是唯一的。 所以当连接器加入 MySQL 集群作为另一个 slave 节点（并且具有唯一 id 的情况下），它就可以读取 binlog。 默认情况下，连接器会在 5400 和 6400 之间生成一个随机数，但是我们建议用户明确指定 Server id。
</td>
</tr>
<tr>
@@ -210,14 +211,14 @@ Connector Options
<td>optional</td>
<td style="word-wrap: break-word;">true</td>
<td>Boolean</td>
- <td>Incremental snapshot is a new mechanism to read snapshot of a table. Compared to the old snapshot mechanism,
- the incremental snapshot has many advantages, including:
- (1) source can be parallel during snapshot reading,
- (2) source can perform checkpoints in the chunk granularity during snapshot reading,
- (3) source doesn't need to acquire global read lock (FLUSH TABLES WITH READ LOCK) before snapshot reading.
- If you would like the source run in parallel, each parallel reader should have an unique server id, so
- the 'server-id' must be a range like '5400-6400', and the range must be larger than the parallelism.
- Please see <a href="#incremental-snapshot-reading ">Incremental Snapshot Reading</a>section for more detailed information.
+ <td>增量快照是一种读取表快照的新机制，与旧的快照机制相比，
+ 增量快照有许多优点，包括：
+ （1）在快照读取期间，Source 支持并发读取，
+ （2）在快照读取期间，Source 支持进行 chunk 粒度的 checkpoint，
+ （3）在快照读取之前，Source 不需要数据库锁权限。
+ 如果希望 Source 并行运行，则每个并行 Readers 都应该具有唯一的 Server id，所以
+ Server id 必须是类似 `5400-6400` 的范围，并且该范围必须大于并行度。
+ 请查阅 <a href="#a-name-id-001-a">增量快照读取</a> 章节了解更多详细信息。
</td>
</tr>
<tr>
@@ -225,67 +226,68 @@ Connector Options
<td>optional</td>
<td style="word-wrap: break-word;">8096</td>
<td>Integer</td>
- <td>The chunk size (number of rows) of table snapshot, captured tables are split into multiple chunks when read the snapshot of table.</td>
+ <td>表快照的块大小（行数），读取表的快照时，捕获的表被拆分为多个块。</td>
</tr>
<tr>
<td>scan.snapshot.fetch.size</td>
<td>optional</td>
<td style="word-wrap: break-word;">1024</td>
<td>Integer</td>
- <td>The maximum fetch size for per poll when read table snapshot.</td>
+ <td>读取表快照时每次读取数据的最大条数。</td>
</tr>
<tr>
<td>scan.startup.mode</td>
<td>optional</td>
<td style="word-wrap: break-word;">initial</td>
<td>String</td>
- <td>Optional startup mode for MySQL CDC consumer, valid enumerations are "initial", "earliest-offset", "latest-offset", "specific-offset" and "timestamp".
- Please see <a href="#startup-reading-position">Startup Reading Position</a> section for more detailed information.</td>
+ <td> MySQL CDC 消费者可选的启动模式，
+ 合法的模式为 "initial"，"earliest-offset"，"latest-offset"，"specific-offset" 和 "timestamp"。
+ 请查阅 <a href="#a-name-id-002-a">启动模式</a> 章节了解更多详细信息。</td>
</tr>
<tr>
<td>scan.startup.specific-offset.file</td>
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Optional binlog file name used in case of "specific-offset" startup mode</td>
+ <td>在 "specific-offset" 启动模式下，启动位点的 binlog 文件名。</td>
</tr>
<tr>
<td>scan.startup.specific-offset.pos</td>
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>Long</td>
- <td>Optional binlog file position used in case of "specific-offset" startup mode</td>
+ <td>在 "specific-offset" 启动模式下，启动位点的 binlog 文件位置。</td>
</tr>
<tr>
<td>scan.startup.specific-offset.gtid-set</td>
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Optional GTID set used in case of "specific-offset" startup mode</td>
+ <td>在 "specific-offset" 启动模式下，启动位点的 GTID 集合。</td>
</tr>
<tr>
<td>scan.startup.specific-offset.skip-events</td>
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>Long</td>
- <td>Optional number of events to skip after the specific starting offset</td>
+ <td>在指定的启动位点后需要跳过的事件数量。</td>
</tr>
<tr>
<td>scan.startup.specific-offset.skip-rows</td>
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>Long</td>
- <td>Optional number of rows to skip after the specific starting offset</td>
+ <td>在指定的启动位点后需要跳过的数据行数量。</td>
</tr>
<tr>
<td>server-time-zone</td>
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>The session time zone in database server, e.g. "Asia/Shanghai".
- It controls how the TIMESTAMP type in MYSQL converted to STRING.
- See more <a href="https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-temporal-types">here</a>.
- If not set, then ZoneId.systemDefault() is used to determine the server time zone.
+ <td>数据库服务器中的会话时区， 例如： "Asia/Shanghai".
+ 它控制 MYSQL 中的时间戳类型如何转换为字符串。
+ 更多请参考 <a href="https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-temporal-types"> 这里</a>.
+ 如果没有设置，则使用ZoneId.systemDefault()来确定服务器时区。
</td>
</tr>
<tr>
@@ -295,82 +297,67 @@ Connector Options
<td style="word-wrap: break-word;">1000</td>
<td>Integer</td>
<td>
-During a snapshot operation, the connector will query each included table to produce a read event for all rows in that table. This parameter determines whether the MySQL connection will pull all results for a table into memory (which is fast but requires large amounts of memory), or whether the results will instead be streamed (can be slower, but will work for very large tables). The value specifies the minimum number of rows a table must contain before the connector will stream results, [...]
+ 在快照操作期间，连接器将查询每个包含的表，以生成该表中所有行的读取事件。 此参数确定 MySQL 连接是否将表的所有结果拉入内存（速度很快，但需要大量内存）， 或者结果是否需要流式传输（传输速度可能较慢，但适用于非常大的表）。 该值指定了在连接器对结果进行流式处理之前，表必须包含的最小行数，默认值为1000。将此参数设置为`0`以跳过所有表大小检查，并始终在快照期间对所有结果进行流式处理。</td>
</tr>
<tr>
<td>connect.timeout</td>
<td>optional</td>
<td style="word-wrap: break-word;">30s</td>
<td>Duration</td>
- <td>The maximum time that the connector should wait after trying to connect to the MySQL database server before timing out.</td>
+ <td>连接器在尝试连接到 MySQL 数据库服务器后超时前应等待的最长时间。</td>
</tr>
<tr>
<td>connect.max-retries</td>
<td>optional</td>
<td style="word-wrap: break-word;">3</td>
<td>Integer</td>
- <td>The max retry times that the connector should retry to build MySQL database server connection.</td>
+ <td>连接器应重试以建立 MySQL 数据库服务器连接的最大重试次数。</td>
</tr>
<tr>
<td>connection.pool.size</td>
<td>optional</td>
<td style="word-wrap: break-word;">20</td>
<td>Integer</td>
- <td>The connection pool size.</td>
+ <td>连接池大小。</td>
</tr>
<tr>
<td>jdbc.properties.*</td>
<td>optional</td>
<td style="word-wrap: break-word;">20</td>
<td>String</td>
- <td>Option to pass custom JDBC URL properties. User can pass custom properties like 'jdbc.properties.useSSL' = 'false'.</td>
+ <td>传递自定义 JDBC URL 属性的选项。用户可以传递自定义属性，如 'jdbc.properties.useSSL' = 'false'.</td>
</tr>
<tr>
<td>heartbeat.interval</td>
<td>optional</td>
<td style="word-wrap: break-word;">30s</td>
<td>Duration</td>
- <td>The interval of sending heartbeat event for tracing the latest available binlog offsets.</td>
+ <td>用于跟踪最新可用 binlog 偏移的发送心跳事件的间隔。</td>
</tr>
<tr>
<td>debezium.*</td>
<td>optional</td>
<td style="word-wrap: break-word;">(none)</td>
<td>String</td>
- <td>Pass-through Debezium's properties to Debezium Embedded Engine which is used to capture data changes from MySQL server.
+ <td>将 Debezium 的属性传递给 Debezium 嵌入式引擎，该引擎用于从 MySQL 服务器捕获数据更改。
For example: <code>'debezium.snapshot.mode' = 'never'</code>.
- See more about the <a href="https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-connector-properties">Debezium's MySQL Connector properties</a></td>
+ 查看更多关于 <a href="https://debezium.io/documentation/reference/1.9/connectors/mysql.html#mysql-connector-properties"> Debezium 的 MySQL 连接器属性</a></td>
</tr>
<tr>
<td>scan.incremental.close-idle-reader.enabled</td>
<td>optional</td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
- <td>Whether to close idle readers at the end of the snapshot phase. <br>
- The flink version is required to be greater than or equal to 1.14 when 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' is set to true.<br>
- If the flink version is greater than or equal to 1.15, the default value of 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' has been changed to true,
- so it does not need to be explicitly configured 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' = 'true'
- </td>
- </tr>
- <tr>
- <td>debezium.binary.handling.mode</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">(none)</td>
- <td>String</td>
- <td>debezium.binary.handling.mode can be set to one of the following values:
- none: No processing is performed, and the binary data type is transmitted as a byte array (byte array).
- base64: The binary data type is converted to a Base64-encoded string and transmitted.
- hex: The binary data type is converted to a hexadecimal string and transmitted.
- The default value is none. Depending on your requirements and data types, you can choose the appropriate processing mode. If your database contains a large number of binary data types, it is recommended to use base64 or hex mode to make it easier to handle during transmission.</td>
+ <td>是否在快照结束后关闭空闲的 Reader。 此特性需要 flink 版本大于等于 1.14 并且 'execution.checkpointing.checkpoints-after-tasks-finish.enabled' 需要设置为 true。</td>
</tr>
</tbody>
</table>
</div>
-Available Metadata
+支持的元数据
----------------
-The following format metadata can be exposed as read-only (VIRTUAL) columns in a table definition.
+下表中的元数据可以在 DDL 中作为只读（虚拟）meta 列声明。
<table class="colwidths-auto docutils">
<thead>
@@ -384,154 +371,138 @@ The following format metadata can be exposed as read-only (VIRTUAL) columns in a
<tr>
<td>table_name</td>
<td>STRING NOT NULL</td>
- <td>Name of the table that contain the row.</td>
+ <td>当前记录所属的表名称。</td>
</tr>
<tr>
<td>database_name</td>
<td>STRING NOT NULL</td>
- <td>Name of the database that contain the row.</td>
+ <td>当前记录所属的库名称。</td>
</tr>
<tr>
<td>op_ts</td>
<td>TIMESTAMP_LTZ(3) NOT NULL</td>
- <td>It indicates the time that the change was made in the database. <br>If the record is read from snapshot of the table instead of the binlog, the value is always 0.</td>
- </tr>
- <tr>
- <td>row_kind</td>
- <td>STRING NOT NULL</td>
- <td>It indicates the row kind of the changelog,Note: The downstream SQL operator may fail to compare due to this new added column when processing the row retraction if
-the source operator chooses to output the 'row_kind' column for each record. It is recommended to use this metadata column only in simple synchronization jobs.
-<br>'+I' means INSERT message, '-D' means DELETE message, '-U' means UPDATE_BEFORE message and '+U' means UPDATE_AFTER message.</td>
+ <td>当前记录表在数据库中更新的时间。 <br>如果从表的快照而不是 binlog 读取记录，该值将始终为0。</td>
</tr>
</tbody>
</table>
-The extended CREATE TABLE example demonstrates the syntax for exposing these metadata fields:
-
+下述创建表示例展示元数据列的用法：
```sql
-CREATE TABLE products
-(
- db_name STRING METADATA FROM 'database_name' VIRTUAL,
- table_name STRING METADATA FROM 'table_name' VIRTUAL,
- operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
- operation STRING METADATA FROM 'row_kind' VIRTUAL,
- order_id INT,
- order_date TIMESTAMP(0),
+CREATE TABLE products (
+ db_name STRING METADATA FROM 'database_name' VIRTUAL,
+ table_name STRING METADATA FROM 'table_name' VIRTUAL,
+ operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
+ order_id INT,
+ order_date TIMESTAMP(0),
customer_name STRING,
- price DECIMAL(10, 5),
- product_id INT,
- order_status BOOLEAN,
- PRIMARY KEY (order_id) NOT ENFORCED
+ price DECIMAL(10, 5),
+ product_id INT,
+ order_status BOOLEAN,
+ PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
- 'connector' = 'mysql-cdc',
- 'hostname' = 'localhost',
- 'port' = '3306',
- 'username' = 'root',
- 'password' = '123456',
- 'database-name' = 'mydb',
- 'table-name' = 'orders'
- );
+ 'connector' = 'mysql-cdc',
+ 'hostname' = 'localhost',
+ 'port' = '3306',
+ 'username' = 'root',
+ 'password' = '123456',
+ 'database-name' = 'mydb',
+ 'table-name' = 'orders'
+);
```
-The extended CREATE TABLE example demonstrates the usage of regex to match multi-tables:
-
+下述创建表示例展示使用正则表达式匹配多张库表的用法：
```sql
-CREATE TABLE products
-(
- db_name STRING METADATA FROM 'database_name' VIRTUAL,
- table_name STRING METADATA FROM 'table_name' VIRTUAL,
- operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
- operation STRING METADATA FROM 'row_kind' VIRTUAL,
- order_id INT,
- order_date TIMESTAMP(0),
+CREATE TABLE products (
+ db_name STRING METADATA FROM 'database_name' VIRTUAL,
+ table_name STRING METADATA FROM 'table_name' VIRTUAL,
+ operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
+ order_id INT,
+ order_date TIMESTAMP(0),
customer_name STRING,
- price DECIMAL(10, 5),
- product_id INT,
- order_status BOOLEAN,
- PRIMARY KEY (order_id) NOT ENFORCED
+ price DECIMAL(10, 5),
+ product_id INT,
+ order_status BOOLEAN,
+ PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
- 'connector' = 'mysql-cdc',
- 'hostname' = 'localhost',
- 'port' = '3306',
- 'username' = 'root',
- 'password' = '123456',
- 'database-name' = '(^(test).*|^(tpc).*|txc|.*[p$]|t{2})',
- 'table-name' = '(t[5-8]|tt)'
- );
+ 'connector' = 'mysql-cdc',
+ 'hostname' = 'localhost',
+ 'port' = '3306',
+ 'username' = 'root',
+ 'password' = '123456',
+ 'database-name' = '(^(test).*|^(tpc).*|txc|.*[p$]|t{2})',
+ 'table-name' = '(t[5-8]|tt)'
+);
```
<table class="colwidths-auto docutils">
<thead>
<tr>
- <th class="text-left" style="width: 15%">example</th>
- <th class="text-left" style="width: 30%">expression</th>
- <th class="text-left" style="width: 55%">description</th>
+ <th class="text-left" style="width: 15%">匹配示例</th>
+ <th class="text-left" style="width: 30%">表达式</th>
+ <th class="text-left" style="width: 55%">描述</th>
</tr>
</thead>
<tbody>
<tr>
- <td>prefix match</td>
+ <td>前缀匹配</td>
<td>^(test).*</td>
- <td>This matches the database name or table name starts with prefix of test, e.g test1、test2. </td>
+ <td>匹配前缀为test的数据库名或表名，例如test1、test2等。</td>
</tr>
<tr>
- <td>suffix match</td>
+ <td>后缀匹配</td>
<td>.*[p$]</td>
- <td>This matches the database name or table name ends with suffix of p, e.g cdcp、edcp. </td>
+ <td>匹配后缀为p的数据库名或表名，例如cdcp、edcp等。</td>
</tr>
<tr>
- <td>specific match</td>
+ <td>特定匹配</td>
<td>txc</td>
- <td>This matches the database name or table name according to a specific name, e.g txc. </td>
+ <td>匹配具体的数据库名或表名。</td>
</tr>
</tbody>
</table>
-It will use `database-name\\.table-name` as a pattern to match tables, as above examples using pattern `(^(test).*|^(tpc).*|txc|.*[p$]|t{2})\\.(t[5-8]|tt)` matches txc.tt、test2.test5.
-
+进行库表匹配时，会使用正则表达式 `database-name\\.table-name` 来与MySQL表的全限定名做匹配，所以该例子使用 `(^(test).*|^(tpc).*|txc|.*[p$]|t{2})\\.(t[5-8]|tt)`，可以匹配到表 txc.tt、test2.test5。
-Features
+支持的特性
--------
-### Incremental Snapshot Reading
+### 增量快照读取<a name="增量快照读取" id="001" ></a>
-Incremental snapshot reading is a new mechanism to read snapshot of a table. Compared to the old snapshot mechanism, the incremental snapshot has many advantages, including:
-* (1) MySQL CDC Source can be parallel during snapshot reading
-* (2) MySQL CDC Source can perform checkpoints in the chunk granularity during snapshot reading
-* (3) MySQL CDC Source doesn't need to acquire global read lock (FLUSH TABLES WITH READ LOCK) before snapshot reading
+增量快照读取是一种读取表快照的新机制。与旧的快照机制相比，增量快照具有许多优点，包括：
+* （1）在快照读取期间，Source 支持并发读取，
+* （2）在快照读取期间，Source 支持进行 chunk 粒度的 checkpoint，
+* （3）在快照读取之前，Source 不需要数据库锁权限。
-If you would like the source run in parallel, each parallel reader should have an unique server id, so the 'server-id' must be a range like '5400-6400',
-and the range must be larger than the parallelism.
+如果希望 source 并行运行，则每个并行 reader 都应该具有唯一的 server id，因此`server id`的范围必须类似于 `5400-6400`，
+且范围必须大于并行度。在增量快照读取过程中，MySQL CDC Source 首先通过表的主键将表划分成多个块（chunk），
+然后 MySQL CDC Source 将多个块分配给多个 reader 以并行读取表的数据。
-During the incremental snapshot reading, the MySQL CDC Source firstly splits snapshot chunks (splits) by primary key of table,
-and then MySQL CDC Source assigns the chunks to multiple readers to read the data of snapshot chunk.
+#### 并发读取
-#### Controlling Parallelism
-
-Incremental snapshot reading provides the ability to read snapshot data parallelly.
-You can control the source parallelism by setting the job parallelism `parallelism.default`. For example, in SQL CLI:
+增量快照读取提供了并行读取快照数据的能力。
+你可以通过设置作业并行度的方式来控制 Source 的并行度 `parallelism.default`. For example, in SQL CLI:
```sql
Flink SQL> SET 'parallelism.default' = 8;
```
-#### Checkpoint
+#### 全量阶段支持 checkpoint
-Incremental snapshot reading provides the ability to perform checkpoint in chunk level. It resolves the checkpoint timeout problem in previous version with old snapshot reading mechanism.
+增量快照读取提供了在区块级别执行检查点的能力。它使用新的快照读取机制解决了以前版本中的检查点超时问题。
-#### Lock-free
+#### 无锁算法
-The MySQL CDC source use **incremental snapshot algorithm**, which avoid acquiring global read lock (FLUSH TABLES WITH READ LOCK) and thus doesn't need `RELOAD` permission.
+MySQL CDC source 使用 增量快照算法, 避免了数据库锁的使用，因此不需要 “RELOAD” 权限。
-#### MySQL High Availability Support
+#### MySQL高可用性支持
-The ```mysql-cdc``` connector offers high availability of MySQL high available cluster by using the [GTID](https://dev.mysql.com/doc/refman/5.7/en/replication-gtids-concepts.html) information. To obtain the high availability, the MySQL cluster need enable the GTID mode, the GTID mode in your mysql config file should contain following settings:
+```mysql cdc``` 连接器通过使用 [GTID](https://dev.mysql.com/doc/refman/5.7/en/replication-gtids-concepts.html) 提供 MySQL 高可用集群的高可用性信息。为了获得高可用性， MySQL集群需要启用 GTID 模式，MySQL 配置文件中的 GTID 模式应该包含以下设置：
```yaml
gtid_mode = on
enforce_gtid_consistency = on
```
-If the monitored MySQL server address contains slave instance, you need set following settings to the MySQL conf file. The setting ```log-slave-updates = 1``` enables the slave instance to also write the data that synchronized from master to its binlog, this makes sure that the ```mysql-cdc``` connector can consume entire data from the slave instance.
+如果监控的MySQL服务器地址包含从实例，则需要对MySQL配置文件设置以下设置。设置 ```log slave updates=1``` 允许从实例也将从主实例同步的数据写入其binlog， 这确保了```mysql cdc```连接器可以使用从实例中的全部数据。
```yaml
gtid_mode = on
@@ -539,40 +510,39 @@ enforce_gtid_consistency = on
log-slave-updates = 1
```
-After the server you monitored fails in MySQL cluster, you only need to change the monitored server address to other available server and then restart the job from the latest checkpoint/savepoint, the job will restore from the checkpoint/savepoint and won't miss any records.
+MySQL 集群中你监控的服务器出现故障后, 你只需将受监视的服务器地址更改为其他可用服务器，然后从最新的检查点/保存点重新启动作业, 作业将从 checkpoint/savepoint 恢复，不会丢失任何记录。
-It's recommended to configure a DNS(Domain Name Service) or VIP(Virtual IP Address) for your MySQL cluster, using the DNS or VIP address for ```mysql-cdc``` connector, the DNS or VIP would automatically route the network request to the active MySQL server. In this way, you don't need to modify the address and restart your pipeline anymore.
+建议为 MySQL 集群配置 DNS（域名服务）或 VIP（虚拟 IP 地址）， 使用```mysql cdc```连接器的 DNS 或 VIP 地址， DNS或VIP将自动将网络请求路由到活动MySQL服务器。 这样，你就不再需要修改地址和重新启动管道。
-#### MySQL Heartbeat Event Support
+#### MySQL心跳事件支持
-If the table updates infrequently, the binlog file or GTID set may have been cleaned in its last committed binlog position.
-The CDC job may restart fails in this case. So the heartbeat event will help update binlog position. By default heartbeat event is enabled in MySQL CDC source and the interval is set to 30 seconds. You can specify the interval by using table option ```heartbeat.interval```, or set the option to `0s` to disable heartbeat events.
+如果表不经常更新，则 binlog 文件或 GTID 集可能已在其最后提交的 binlog 位置被清理。
+在这种情况下，CDC 作业可能会重新启动失败。因此心跳事件将帮助更新 binlog 位置。 默认情况下，MySQL CDC Source 启用心跳事件，间隔设置为30秒。 可以使用表选项```heartbeat```指定间隔。或将选项设置为```0s```以禁用心跳事件。
-#### How Incremental Snapshot Reading works
+#### 增量快照读取的工作原理
-When the MySQL CDC source is started, it reads snapshot of table parallelly and then reads binlog of table with single parallelism.
+当 MySQL CDC Source 启动时，它并行读取表的快照，然后以单并行度的方式读取表的 binlog。
-In snapshot phase, the snapshot is cut into multiple snapshot chunks according to primary key of table and the size of table rows.
-Snapshot chunks is assigned to multiple snapshot readers. Each snapshot reader reads its received chunks with [chunk reading algorithm](#snapshot-chunk-reading) and send the read data to downstream.
-The source manages the process status (finished or not) of chunks, thus the source of snapshot phase can support checkpoint in chunk level.
-If a failure happens, the source can be restored and continue to read chunks from last finished chunks.
+在快照阶段，根据表的主键和表行的大小将快照切割成多个快照块。
+快照块被分配给多个快照读取器。每个快照读取器使用 [区块读取算法](#snapshot-chunk-reading) 并将读取的数据发送到下游。
+Source 会管理块的进程状态（完成或未完成），因此快照阶段的 Source 可以支持块级别的 checkpoint。
+如果发生故障，可以恢复 Source 并继续从最后完成的块中读取块。
-After all snapshot chunks finished, the source will continue to read binlog in a single task.
-In order to guarantee the global data order of snapshot records and binlog records, binlog reader will start to read data
-until there is a complete checkpoint after snapshot chunks finished to make sure all snapshot data has been consumed by downstream.
-The binlog reader tracks the consumed binlog position in state, thus source of binlog phase can support checkpoint in row level.
+所有快照块完成后，Source 将继续在单个任务中读取 binlog。
+为了保证快照记录和 binlog 记录的全局数据顺序，binlog reader 将开始读取数据直到快照块完成后并有一个完整的 checkpoint，以确保所有快照数据已被下游消费。
+binlog reader 在状态中跟踪所使用的 binlog 位置，因此 binlog 阶段的 Source 可以支持行级别的 checkpoint。
-Flink performs checkpoints for the source periodically, in case of failover, the job will restart and restore from the last successful checkpoint state and guarantees the exactly once semantic.
+Flink 定期为 Source 执行 checkpoint，在故障转移的情况下，作业将重新启动并从最后一个成功的 checkpoint 状态恢复，并保证只执行一次语义。
-##### Snapshot Chunk Splitting
+##### 全量阶段分片算法
-When performing incremental snapshot reading, MySQL CDC source need a criterion which used to split the table.
-MySQL CDC Source use a splitting column to split the table to multiple splits (chunks). By default, MySQL CDC source will identify the primary key column of the table and use the first column in primary key as the splitting column.
-If there is no primary key in the table, incremental snapshot reading will fail and you can disable `scan.incremental.snapshot.enabled` to fallback to old snapshot reading mechanism.
+在执行增量快照读取时，MySQL CDC source 需要一个用于分片的的算法。
+MySQL CDC Source 使用主键列将表划分为多个分片（chunk）。 默认情况下，MySQL CDC source 会识别表的主键列，并使用主键中的第一列作为用作分片列。
+如果表中没有主键， 增量快照读取将失败，你可以禁用 `scan.incremental.snapshot.enabled` 来回退到旧的快照读取机制。
-For numeric and auto incremental splitting column, MySQL CDC Source efficiently splits chunks by fixed step length.
-For example, if you had a table with a primary key column of `id` which is auto-incremental BIGINT type, the minimum value was `0` and maximum value was `100`,
-and the table option `scan.incremental.snapshot.chunk.size` value is `25`, the table would be split into following chunks:
+对于数值和自动增量拆分列，MySQL CDC Source 按固定步长高效地拆分块。
+例如，如果你有一个主键列为`id`的表，它是自动增量 BIGINT 类型，最小值为`0`，最大值为`100`，
+和表选项 `scan.incremental.snapshot.chunk.size` 大小 `value`为`25`，表将被拆分为以下块：
```
(-∞, 25),
@@ -582,8 +552,8 @@ and the table option `scan.incremental.snapshot.chunk.size` value is `25`, the t
[100, +∞)
```
-For other primary key column type, MySQL CDC Source executes the statement in the form of `SELECT MAX(STR_ID) AS chunk_high FROM (SELECT * FROM TestTable WHERE STR_ID > 'uuid-001' limit 25)` to get the low and high value for each chunk,
-the splitting chunks set would be like:
+对于其他主键列类型， MySQL CDC Source 将以下形式执行语句： `SELECT MAX(STR_ID) AS chunk_high FROM (SELECT * FROM TestTable WHERE STR_ID > 'uuid-001' limit 25)` 来获得每个区块的低值和高值，
+分割块集如下所示：
```
(-∞, 'uuid-001'),
@@ -593,135 +563,133 @@ the splitting chunks set would be like:
[uuid-def, +∞).
```
-##### Chunk Reading Algorithm
+##### Chunk 读取算法
+
+对于上面的示例`MyTable`，如果 MySQL CDC Source 并行度设置为 4，MySQL CDC Source 将在每一个 executes 运行 4 个 Readers **通过偏移信号算法**
+获取快照区块的最终一致输出。 **偏移信号算法**简单描述如下：
-For above example `MyTable`, if the MySQL CDC Source parallelism was set to 4, MySQL CDC Source would run 4 readers which each executes **Offset Signal Algorithm** to
-get a final consistent output of the snapshot chunk. The **Offset Signal Algorithm** simply describes as following:
+* (1) 将当前 binlog 位置记录为`LOW`偏移量
+* (2) 通过执行语句读取并缓冲快照区块记录 `SELECT * FROM MyTable WHERE id > chunk_low AND id <= chunk_high`
+* (3) 将当前 binlog 位置记录为`HIGH`偏移量
+* (4) 从`LOW`偏移量到`HIGH`偏移量读取属于快照区块的 binlog 记录
+* (5) 将读取的 binlog 记录向上插入缓冲区块记录，并发出缓冲区中的所有记录作为快照区块的最终输出（全部作为插入记录）
+* (6) 继续读取并发出属于 *单个 binlog reader* 中`HIGH`偏移量之后的区块的 binlog 记录。
- * (1) Record current binlog position as `LOW` offset
- * (2) Read and buffer the snapshot chunk records by executing statement `SELECT * FROM MyTable WHERE id > chunk_low AND id <= chunk_high`
- * (3) Record current binlog position as `HIGH` offset
- * (4) Read the binlog records that belong to the snapshot chunk from `LOW` offset to `HIGH` offset
- * (5) Upsert the read binlog records into the buffered chunk records, and emit all records in the buffer as final output (all as INSERT records) of the snapshot chunk
- * (6) Continue to read and emit binlog records belong to the chunk after the `HIGH` offset in *single binlog reader*.
+该算法的是基于 [DBLog Paper](https://arxiv.org/pdf/2010.12597v1.pdf) 并结合 Flink 的一个变种, 请参考它了解更多详细信息。
-The algorithm is inspired by [DBLog Paper](https://arxiv.org/pdf/2010.12597v1.pdf), please refer it for more detail.
-
-**Note:** If the actual values for the primary key are not uniformly distributed across its range, this may lead to unbalanced tasks when incremental snapshot read.
+**注意:** 如果主键的实际值在其范围内分布不均匀，则在增量快照读取时可能会导致任务不平衡。
-### Exactly-Once Processing
+### Exactly-Once 处理
-The MySQL CDC connector is a Flink Source connector which will read table snapshot chunks first and then continues to read binlog,
-both snapshot phase and binlog phase, MySQL CDC connector read with **exactly-once processing** even failures happen.
+MySQL CDC 连接器是一个 Flink Source 连接器，它将首先读取表快照块，然后继续读取 binlog，
+无论是在快照阶段还是读取 binlog 阶段，MySQL CDC 连接器都会在处理时**准确读取数据**，即使任务出现了故障。
-### Startup Reading Position
+### 启动模式<a name="启动模式" id="002" ></a>
-The config option `scan.startup.mode` specifies the startup mode for MySQL CDC consumer. The valid enumerations are:
+配置选项```scan.startup.mode```指定 MySQL CDC 使用者的启动模式。有效枚举包括：
-- `initial` (default): Performs an initial snapshot on the monitored database tables upon first startup, and continue to read the latest binlog.
-- `earliest-offset`: Skip snapshot phase and start reading binlog events from the earliest accessible binlog offset.
-- `latest-offset`: Never to perform snapshot on the monitored database tables upon first startup, just read from
-the end of the binlog which means only have the changes since the connector was started.
-- `specific-offset`: Skip snapshot phase and start reading binlog events from a specific offset. The offset could be
-specified with binlog filename and position, or a GTID set if GTID is enabled on server.
-- `timestamp`: Skip snapshot phase and start reading binlog events from a specific timestamp.
+- `initial` （默认）：在第一次启动时对受监视的数据库表执行初始快照，并继续读取最新的 binlog。
+- `earliest-offset`：跳过快照阶段，从可读取的最早 binlog 位点开始读取
+- `latest-offset`：首次启动时，从不对受监视的数据库表执行快照， 连接器仅从 binlog 的结尾处开始读取，这意味着连接器只能读取在连接器启动之后的数据更改。
+- `specific-offset`：跳过快照阶段，从指定的 binlog 位点开始读取。位点可通过 binlog 文件名和位置指定，或者在 GTID 在集群上启用时通过 GTID 集合指定。
+- `timestamp`：跳过快照阶段，从指定的时间戳开始读取 binlog 事件。
-For example in DataStream API:
+例如使用 DataStream API:
```java
MySQLSource.builder()
- .startupOptions(StartupOptions.earliest()) // Start from earliest offset
- .startupOptions(StartupOptions.latest()) // Start from latest offset
- .startupOptions(StartupOptions.specificOffset("mysql-bin.000003", 4L) // Start from binlog file and offset
- .startupOptions(StartupOptions.specificOffset("24DA167-0C0C-11E8-8442-00059A3C7B00:1-19")) // Start from GTID set
- .startupOptions(StartupOptions.timestamp(1667232000000L) // Start from timestamp
+ .startupOptions(StartupOptions.earliest()) // 从最早位点启动
+ .startupOptions(StartupOptions.latest()) // 从最晚位点启动
+ .startupOptions(StartupOptions.specificOffset("mysql-bin.000003", 4L) // 从指定 binlog 文件名和位置启动
+ .startupOptions(StartupOptions.specificOffset("24DA167-0C0C-11E8-8442-00059A3C7B00:1-19")) // 从 GTID 集合启动
+ .startupOptions(StartupOptions.timestamp(1667232000000L) // 从时间戳启动
...
.build()
```
-and with SQL:
+使用 SQL:
```SQL
CREATE TABLE mysql_source (...) WITH (
'connector' = 'mysql-cdc',
- 'scan.startup.mode' = 'earliest-offset', -- Start from earliest offset
- 'scan.startup.mode' = 'latest-offset', -- Start from latest offset
- 'scan.startup.mode' = 'specific-offset', -- Start from specific offset
- 'scan.startup.mode' = 'timestamp', -- Start from timestamp
- 'scan.startup.specific-offset.file' = 'mysql-bin.000003', -- Binlog filename under specific offset startup mode
- 'scan.startup.specific-offset.pos' = '4', -- Binlog position under specific offset mode
- 'scan.startup.specific-offset.gtid-set' = '24DA167-0C0C-11E8-8442-00059A3C7B00:1-19', -- GTID set under specific offset startup mode
- 'scan.startup.timestamp-millis' = '1667232000000' -- Timestamp under timestamp startup mode
+ 'scan.startup.mode' = 'earliest-offset', -- 从最早位点启动
+ 'scan.startup.mode' = 'latest-offset', -- 从最晚位点启动
+ 'scan.startup.mode' = 'specific-offset', -- 从特定位点启动
+ 'scan.startup.mode' = 'timestamp', -- 从特定位点启动
+ 'scan.startup.specific-offset.file' = 'mysql-bin.000003', -- 在特定位点启动模式下指定 binlog 文件名
+ 'scan.startup.specific-offset.pos' = '4', -- 在特定位点启动模式下指定 binlog 位置
+ 'scan.startup.specific-offset.gtid-set' = '24DA167-0C0C-11E8-8442-00059A3C7B00:1-19', -- 在特定位点启动模式下指定 GTID 集合
+ 'scan.startup.timestamp-millis' = '1667232000000' -- 在时间戳启动模式下指定启动时间戳
...
)
```
-**Notes:**
-1. MySQL source will print the current binlog position into logs with INFO level on checkpoint, with the prefix
-"Binlog offset on checkpoint {checkpoint-id}". It could be useful if you want to restart the job from a specific checkpointed position.
-2. If schema of capturing tables was changed previously, starting with earliest offset, specific offset or timestamp
-could fail as the Debezium reader keeps the current latest table schema internally and earlier records with unmatched schema cannot be correctly parsed.
+**注意**：
+1. MySQL source 会在 checkpoint 时将当前位点以 INFO 级别打印到日志中，日志前缀为 "Binlog offset on checkpoint {checkpoint-id}"。
+ 该日志可以帮助将作业从某个 checkpoint 的位点开始启动的场景。
+2. 如果捕获变更的表曾经发生过表结构变化，从最早位点、特定位点或时间戳启动可能会发生错误，因为 Debezium 读取器会在内部保存当前的最新表结构，结构不匹配的早期数据无法被正确解析。
+
### DataStream Source
```java
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.cdc.debezium.JsonDebeziumDeserializationSchema;
-import org.apache.flink.cdc.connectors.mysql.source.MySqlSource;
+import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
+import com.ververica.cdc.connectors.mysql.source.MySqlSource;
public class MySqlSourceExample {
public static void main(String[] args) throws Exception {
MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
.hostname("yourHostname")
.port(yourPort)
- .databaseList("yourDatabaseName") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
- .tableList("yourDatabaseName.yourTableName") // set captured table
+ .databaseList("yourDatabaseName") // 设置捕获的数据库， 如果需要同步整个数据库，请将 tableList 设置为 ".*".
+ .tableList("yourDatabaseName.yourTableName") // 设置捕获的表
.username("yourUsername")
.password("yourPassword")
- .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
+ .deserializer(new JsonDebeziumDeserializationSchema()) // 将 SourceRecord 转换为 JSON 字符串
.build();
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- // enable checkpoint
+ // 设置 3s 的 checkpoint 间隔
env.enableCheckpointing(3000);
env
.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
- // set 4 parallel source tasks
+ // 设置 source 节点的并行度为 4
.setParallelism(4)
- .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering
+ .print().setParallelism(1); // 设置 sink 节点并行度为 1
env.execute("Print MySQL Snapshot + Binlog");
}
}
```
-### Scan Newly Added Tables
+### 动态加表
-Scan Newly Added Tables feature enables you add new tables to monitor for existing running pipeline, the newly added tables will read theirs snapshot data firstly and then read their changelog automatically.
-
-Imaging this scenario: At the beginning, a Flink job monitor tables `[product, user, address]`, but after some days we would like the job can also monitor tables `[order, custom]` which contains history data, and we need the job can still reuse existing state of the job, this feature can resolve this case gracefully.
+扫描新添加的表功能使你可以添加新表到正在运行的作业中，新添加的表将首先读取其快照数据，然后自动读取其变更日志。
-The following operations show how to enable this feature to resolve above scenario. An existing Flink job which uses CDC Source like:
+想象一下这个场景：一开始， Flink 作业监控表 `[product, user, address]`, 但几天后，我们希望这个作业还可以监控表 `[order, custom]`，这些表包含历史数据，我们需要作业仍然可以复用作业的已有状态，动态加表功能可以优雅地解决此问题。
+
+以下操作显示了如何启用此功能来解决上述场景。 使用现有的 Flink CDC Source 作业，如下：
```java
MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
.hostname("yourHostname")
.port(yourPort)
- .scanNewlyAddedTableEnabled(true) // enable scan the newly added tables feature
- .databaseList("db") // set captured database
- .tableList("db.product, db.user, db.address") // set captured tables [product, user, address]
+ .scanNewlyAddedTableEnabled(true) // 启用扫描新添加的表功能
+ .databaseList("db") // 设置捕获的数据库
+ .tableList("db.product, db.user, db.address") // 设置捕获的表 [product, user, address]
.username("yourUsername")
.password("yourPassword")
- .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
+ .deserializer(new JsonDebeziumDeserializationSchema()) // 将 SourceRecord 转换为 JSON 字符串
.build();
- // your business code
+ // 你的业务代码
```
-If we would like to add new tables `[order, custom]` to an existing Flink job，just need to update the `tableList()` value of the job to include `[order, custom]` and restore the job from previous savepoint.
+如果我们想添加新表 `[order, custom]` 对于现有的 Flink 作业，只需更新 `tableList()` 将新增表 `[order, custom]` 加入并从已有的 savepoint 恢复作业。
-_Step 1_: Stop the existing Flink job with savepoint.
+_Step 1_: 使用 savepoint 停止现有的 Flink 作业。
```shell
$ ./bin/flink stop $Existing_Flink_JOB_ID
```
@@ -729,73 +697,43 @@ $ ./bin/flink stop $Existing_Flink_JOB_ID
Suspending job "cca7bc1061d61cf15238e92312c2fc20" with a savepoint.
Savepoint completed. Path: file:/tmp/flink-savepoints/savepoint-cca7bc-bb1e257f0dab
```
-_Step 2_: Update the table list option for the existing Flink job .
-1. update `tableList()` value.
-2. build the jar of updated job.
+_Step 2_: 更新现有 Flink 作业的表列表选项。
+1. 更新 `tableList()` 参数.
+2. 编译更新后的作业，示例如下：
```java
MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
.hostname("yourHostname")
.port(yourPort)
.scanNewlyAddedTableEnabled(true)
.databaseList("db")
- .tableList("db.product, db.user, db.address, db.order, db.custom") // set captured tables [product, user, address ,order, custom]
+ .tableList("db.product, db.user, db.address, db.order, db.custom") // 设置捕获的表 [product, user, address ,order, custom]
.username("yourUsername")
.password("yourPassword")
- .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
+ .deserializer(new JsonDebeziumDeserializationSchema()) // 将 SourceRecord 转换为 JSON 字符串
.build();
- // your business code
+ // 你的业务代码
```
-_Step 3_: Restore the updated Flink job from savepoint.
+_Step 3_: 从 savepoint 还原更新后的 Flink 作业。
```shell
$ ./bin/flink run \
--detached \
--fromSavepoint /tmp/flink-savepoints/savepoint-cca7bc-bb1e257f0dab \
./FlinkCDCExample.jar
```
-**Note:** Please refer the doc [Restore the job from previous savepoint](https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/deployment/cli/#command-line-interface) for more details.
-
-### Tables Without primary keys
+**注意:** 请参考文档 [Restore the job from previous savepoint](https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/deployment/cli/#command-line-interface) 了解更多详细信息。
-Starting from version 2.4.0, MySQL CDC support tables that do not have a primary key. To use a table without primary keys, you must configure the `scan.incremental.snapshot.chunk.key-column` option and specify one non-null field.
+### 关于无主键表
-There are two places that need to be taken care of.
+从2.4.0 版本开始支持无主键表，使用无主键表必须设置 `scan.incremental.snapshot.chunk.key-column`，且只能选择非空类型的一个字段。
-1. If there is an index in the table, try to use a column which is contained in the index in `scan.incremental.snapshot.chunk.key-column`. This will increase the speed of select statement.
-2. The processing semantics of a MySQL CDC table without primary keys is determined based on the behavior of the column that are specified by the `scan.incremental.snapshot.chunk.key-column`.
- * If no update operation is performed on the specified column, the exactly-once semantics is ensured.
- * If the update operation is performed on the specified column, only the at-least-once semantics is ensured. However, you can specify primary keys at downstream and perform the idempotence operation to ensure data correctness.
-
-### About converting binary type data to base64 encoded data
-
-```sql
-CREATE TABLE products (
- db_name STRING METADATA FROM 'database_name' VIRTUAL,
- table_name STRING METADATA FROM 'table_name' VIRTUAL,
- operation_ts TIMESTAMP_LTZ(3) METADATA FROM 'op_ts' VIRTUAL,
- order_id INT,
- order_date TIMESTAMP(0),
- customer_name STRING,
- price DECIMAL(10, 5),
- product_id INT,
- order_status BOOLEAN,
- binary_data STRING,
- PRIMARY KEY(order_id) NOT ENFORCED
-) WITH (
- 'connector' = 'mysql-cdc',
- 'hostname' = 'localhost',
- 'port' = '3306',
- 'username' = 'root',
- 'password' = '123456',
- 'database-name' = 'test_db',
- 'table-name' = 'test_tb',
- 'debezium.binary.handling.mode' = 'base64'
-);
-```
+在使用无主键表时，需要注意以下两种情况。
-`binary_data` field in the database is of type VARBINARY(N). In some scenarios, we need to convert binary data to base64 encoded string data. This feature can be enabled by adding the parameter 'debezium.binary.handling.mode'='base64',
-By adding this parameter, we can map the binary field type to 'STRING' in Flink SQL, thereby obtaining base64 encoded string data.
+1. 配置 `scan.incremental.snapshot.chunk.key-column` 时，如果表中存在索引，请尽量使用索引中的列来加快 select 速度。
+2. 无主键表的处理语义由 `scan.incremental.snapshot.chunk.key-column` 指定的列的行为决定：
+* 如果指定的列不存在更新操作，此时可以保证 Exactly once 语义。
+* 如果指定的列存在更新操作，此时只能保证 At least once 语义。但可以结合下游，通过指定下游主键，结合幂等性操作来保证数据的正确性。
-Data Type Mapping
+数据类型映射
----------------
<div class="wy-table-responsive">
@@ -906,8 +844,7 @@ Data Type Mapping
where 38 < p <= 65<br>
</td>
<td>STRING</td>
- <td>The precision for DECIMAL data type is up to 65 in MySQL, but the precision for DECIMAL is limited to 38 in Flink.
- So if you define a decimal column whose precision is greater than 38, you should map it to STRING to avoid precision loss.</td>
+ <td>在 MySQL 中，十进制数据类型的精度高达 65，但在 Flink 中，十进制数据类型的精度仅限于 38。所以，如果定义精度大于 38 的十进制列，则应将其映射到字符串以避免精度损失。在 MySQL 中，十进制数据类型的精度高达65，但在Flink中，十进制数据类型的精度仅限于38。所以，如果定义精度大于 38 的十进制列，则应将其映射到字符串以避免精度损失。</td>
</tr>
<tr>
<td>
@@ -954,7 +891,7 @@ Data Type Mapping
<td>
BIT(n)
</td>
- <td>BINARY(⌈(n + 7) / 8⌉)</td>
+ <td>BINARY(⌈n/8⌉)</td>
<td></td>
</tr>
<tr>
@@ -989,7 +926,7 @@ Data Type Mapping
LONGBLOB<br>
</td>
<td>BYTES</td>
- <td>Currently, for BLOB data type in MySQL, only the blob whose length isn't greater than 2,147,483,647(2 ** 31 - 1) is supported. </td>
+ <td>目前，对于 MySQL 中的 BLOB 数据类型，仅支持长度不大于 2147483647（2**31-1）的 blob。 </td>
</tr>
<tr>
<td>
@@ -1010,15 +947,15 @@ Data Type Mapping
JSON
</td>
<td>STRING</td>
- <td>The JSON data type will be converted into STRING with JSON format in Flink.</td>
+ <td> JSON 数据类型将在 Flink 中转换为 JSON 格式的字符串。</td>
</tr>
<tr>
<td>
SET
</td>
<td>ARRAY<STRING></td>
- <td>As the SET data type in MySQL is a string object that can have zero or more values,
- it should always be mapped to an array of string
+ <td>因为 MySQL 中的 SET 数据类型是一个字符串对象，可以有零个或多个值
+ 它应该始终映射到字符串数组。
</td>
</tr>
<tr>
@@ -1036,34 +973,35 @@ Data Type Mapping
STRING
</td>
<td>
- The spatial data types in MySQL will be converted into STRING with a fixed Json format.
- Please see <a href="#mysql-spatial-data-types-mapping ">MySQL Spatial Data Types Mapping</a> section for more detailed information.
+ MySQL 中的空间数据类型将转换为具有固定 Json 格式的字符串。
+ 请参考 MySQL <a href="#a-name-id-003-a">空间数据类型映射</a> 章节了解更多详细信息。
</td>
</tr>
</tbody>
</table>
</div>
-### MySQL Spatial Data Types Mapping
-The spatial data types except for `GEOMETRYCOLLECTION` in MySQL will be converted into Json String with a fixed format like:<br>
+### 空间数据类型映射<a name="空间数据类型映射" id="003"></a>
+
+MySQL中除`GEOMETRYCOLLECTION`之外的空间数据类型都会转换为 Json 字符串，格式固定，如：<br>
```json
{"srid": 0 , "type": "xxx", "coordinates": [0, 0]}
```
-The field `srid` identifies the SRS in which the geometry is defined, SRID 0 is the default for new geometry values if no SRID is specified.
-As only MySQL 8+ support to specific SRID when define spatial data type, the field `srid` will always be 0 in MySQL with a lower version.
+字段`srid`标识定义几何体的 SRS，如果未指定 SRID，则 SRID 0 是新几何体值的默认值。
+由于 MySQL 8+ 在定义空间数据类型时只支持特定的 SRID，因此在版本较低的MySQL中，字段`srid`将始终为 0。
-The field `type` identifies the spatial data type, such as `POINT`/`LINESTRING`/`POLYGON`.
+字段`type`标识空间数据类型，例如`POINT`/`LINESTRING`/`POLYGON`。
-The field `coordinates` represents the `coordinates` of the spatial data.
+字段`coordinates`表示空间数据的`坐标`。
-For `GEOMETRYCOLLECTION`, it will be converted into Json String with a fixed format like:<br>
+对于`GEOMETRYCOLLECTION`，它将转换为 Json 字符串，格式固定，如：<br>
```json
{"srid": 0 , "type": "GeometryCollection", "geometries": [{"type":"Point","coordinates":[10,10]}]}
```
-The field `geometries` is an array contains all spatial data.
+`Geometrics`字段是一个包含所有空间数据的数组。
-The example for different spatial data types mapping is as follows:
+不同空间数据类型映射的示例如下：
<div class="wy-table-responsive">
<table class="colwidths-auto docutils">
<thead>
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/oceanbase-cdc.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/oceanbase-cdc.md
index c051dd265..7ad77a29f 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/oceanbase-cdc.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/oceanbase-cdc.md
@@ -24,18 +24,26 @@ specific language governing permissions and limitations
under the License.
-->
-# OceanBase CDC Connector
+# OceanBase CDC 连接器
-The OceanBase CDC connector allows for reading snapshot data and incremental data from OceanBase. This document describes how to set up the OceanBase CDC connector to run SQL queries against OceanBase.
+OceanBase CDC 连接器允许从 OceanBase 读取快照数据和增量数据。本文介绍了如何设置 OceanBase CDC 连接器以对 OceanBase 进行 SQL 查询。
-Dependencies
-------------
+## 依赖
-In order to set up the OceanBase CDC connector, the following table provides dependency information for both projects using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles.
+为了使用 OceanBase CDC 连接器，您必须提供相关的依赖信息。以下依赖信息适用于使用自动构建工具（如 Maven 或 SBT）构建的项目和带有 SQL JAR 包的 SQL 客户端。
+为了使用 OceanBase CDC 连接器，您必须提供相关的依赖信息。以下依赖信息适用于使用自动构建工具（如 Maven 或 SBT）构建的项目和带有 SQL JAR 包的 SQL 客户端。
-{{< artifact flink-connector-oceanbase-cdc >}}
+```xml
+<dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-oceanbase-cdc</artifactId>
+ <!-- 请使用已发布的版本依赖，snapshot 版本的依赖需要本地自行编译。 -->
+ <version>3.1-SNAPSHOT</version>
+</dependency>
+
+```
-If you want to use OceanBase JDBC driver to connect to the enterprise edition database, you should also include the following dependency in your class path.
+如果您是要连接企业版的 OceanBase，您可能需要使用 OceanBase 官方的 JDBC 驱动，这时需要引入如下依赖。
```xml
<dependency>
@@ -45,57 +53,51 @@ If you want to use OceanBase JDBC driver to connect to the enterprise edition da
</dependency>
```
-### SQL Client JAR
+## 下载 SQL 客户端 JAR 包
-```Download link is available only for stable releases.```
+```下载链接仅在已发布版本可用，请在文档网站左下角选择浏览已发布的版本。```
-Download [flink-sql-connector-oceanbase-cdc-3.0-SNAPSHOT.jar](https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-oceanbase-cdc/3.0-SNAPSHOT/flink-sql-connector-oceanbase-cdc-3.0-SNAPSHOT.jar) and put it under `<FLINK_HOME>/lib/`.
+下载[flink-sql-connector-oceanbase-cdc-3.0-SNAPSHOT.jar](https://repo1.maven.org/maven2/com/ververica/flink-connector-oceanbase-cdc/3.0.0/flink-connector-oceanbase-cdc-3.0.0.jar) 到 `<FLINK_HOME>/lib/` 目录下。
-**Note:** flink-sql-connector-oceanbase-cdc-XXX-SNAPSHOT version is the code corresponding to the development branch. Users need to download the source code and compile the corresponding jar. Users should use the released version, such as [flink-sql-connector-oceanbase-cdc-3.0.0.jar](https://mvnrepository.com/artifact/com.ververica/flink-sql-connector-oceanbase-cdc), the released version will be available in the Maven central warehouse.
+**注意:** flink-sql-connector-oceanbase-cdc-XXX-SNAPSHOT 版本是开发分支`release-XXX`对应的快照版本，快照版本用户需要下载源代码并编译相应的 jar。用户应使用已经发布的版本，例如 [flink-sql-connector-oceanbase-cdc-3.0.0.jar](https://mvnrepository.com/artifact/com.ververica/flink-sql-connector-oceanbase-cdc) 当前已发布的所有版本都可以在 Maven 中央仓库获取。
-For JDBC driver, the cdc jar above already contains MySQL JDBC driver 5.1.47, which is our recommended version. Due to the license issue, we can not include the OceanBase JDBC driver in the cdc jar. If you need to use it, you can download it from [here](https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.2/oceanbase-client-2.4.2.jar) and put it under `<FLINK_HOME>/lib/`, you also need to set the start option `jdbc.driver` to `com.oceanbase.jdbc.Driver`.
+对于 JDBC 驱动，上述的 cdc jar 文件中已经包含了我们推荐的 MySQL 驱动版本 5.1.47。由于开源许可证的原因，我们不能在上述 cdc jar 文件中包含 OceanBase 的官方 JDBC 驱动，如果您需要使用它，可以从[这里](https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.2/oceanbase-client-2.4.2.jar)下载，然后放到 `<FLINK_HOME>/lib/` 目录下，同时需要将配置项 `jdbc.driver` 设为 `com.oceanbase.jdbc.Driver`。
-Setup OceanBase and LogProxy Server
-----------------------
+### 配置 OceanBase 数据库和 oblogproxy 服务
-1. Set up the OceanBase cluster following the [doc](https://github.com/oceanbase/oceanbase#quick-start).
+1. 按照 [文档](https://github.com/oceanbase/oceanbase#quick-start) 配置 OceanBase 集群。
+2. 在 sys 租户中，为 oblogproxy 创建一个带密码的用户。
-2. Create a user with password in `sys` tenant, this user is used in OceanBase LogProxy.
-
- ```shell
- mysql -h${host} -P${port} -uroot
-
- mysql> SHOW TENANT;
- mysql> CREATE USER ${sys_username} IDENTIFIED BY '${sys_password}';
- mysql> GRANT ALL PRIVILEGES ON *.* TO ${sys_username} WITH GRANT OPTION;
- ```
-
-3. Create a user in the tenant you want to monitor, this is used to read data for snapshot and change event.
+ ```bash
+ mysql -h${host} -P${port} -uroot
+ mysql> SHOW TENANT;
+ mysql> CREATE USER ${sys_username} IDENTIFIED BY '${sys_password}';
+ mysql> GRANT ALL PRIVILEGES ON *.* TO ${sys_username} WITH GRANT OPTION;
+ ```
-4. For users of OceanBase Community Edition, you need to get the `rootserver-list`. You can use the following command to get the value:
+3. 为你想要监控的租户创建一个用户，这个用户用来读取快照数据和变化事件数据。
+4. OceanBase 社区版用户需要获取`rootserver-list`，可以使用以下命令获取：
- ```shell
- mysql> show parameters like 'rootservice_list';
+ ```bash
+ mysql> SHOW PARAMETERS LIKE 'rootservice_list';
```
-
- For users of OceanBase Enterprise Edition, you need to get the `config-url`. You can use the following command to get the value:
+ OceanBase 企业版用户需要获取 `config-url`，可以使用以下命令获取：
```shell
mysql> show parameters like 'obconfig_url';
```
-5. Setup OceanBase LogProxy. For users of OceanBase Community Edition, you can follow the [quick start](https://github.com/oceanbase/oblogproxy#getting-started).
+5. 按照 [文档](https://github.com/oceanbase/oblogproxy#getting-started) 配置 oblogproxy。
-How to create a OceanBase CDC table
-----------------
+## 创建 OceanBase CDC 表
-The OceanBase CDC table can be defined as following:
+使用以下命令，创建 OceanBase CDC 表：
```sql
--- checkpoint every 3000 milliseconds
+-- 每 3 秒做一次 checkpoint，用于测试，生产配置建议 5 到 10 分钟
Flink SQL> SET 'execution.checkpointing.interval' = '3s';
--- register a OceanBase table 'orders' in Flink SQL
+-- 在 Flink SQL 中创建 OceanBase 表 `orders`
Flink SQL> CREATE TABLE orders (
order_id INT,
order_date TIMESTAMP(0),
@@ -120,11 +122,11 @@ Flink SQL> CREATE TABLE orders (
'working-mode' = 'memory'
);
--- read snapshot and binlogs from orders table
+-- 从表 orders 中读取快照数据和 binlog 数据
Flink SQL> SELECT * FROM orders;
```
-If you want to use OceanBase Oracle mode, you need to add the OceanBase jdbc jar file to Flink and set up the enterprise edition of oblogproxy, then you can create a table in Flink as following:
+如果您使用的是企业版的 OceanBase Oracle 模式，您需要先添加 OceanBase 的官方 JDBC 驱动 jar 包到 Flink 环境，并且部署企业版的 oblogproxy 服务，然后通过以下命令创建 OceanBase CDC 表：
```sql
Flink SQL> CREATE TABLE orders (
@@ -154,235 +156,230 @@ Flink SQL> CREATE TABLE orders (
);
```
-You can also try the quickstart tutorial that sync data from OceanBase to Elasticsearch, please refer [Flink CDC Tutorial]({{< ref "docs/connectors/legacy-flink-cdc-sources/tutorials/oceanbase-tutorial" >}}) for more information.
+您也可以访问 Flink CDC 官网文档，快速体验将数据从 OceanBase 导入到 Elasticsearch。更多信息，参考 [Flink CDC 官网文档](https://nightlies.apache.org/flink/flink-cdc-docs-release-3.0/docs/connectors/legacy-flink-cdc-sources/tutorials/oceanbase-tutorial/)。
-Connector Options
-----------------
+## OceanBase CDC 连接器选项
-The OceanBase CDC Connector contains some options for both sql and stream api as the following sheet.
+OceanBase CDC 连接器包括用于 SQL 和 DataStream API 的选项，如下表所示。
-*Note*: The connector supports two ways to specify the table list to listen to, and will get the union of the results when both way are used at the same time.
-1. Use `database-name` and `table-name` to match database and table names in regex. As the `obcdc` (former `liboblog`) only supports `fnmatch` now, we can't use regex directly to filter change events, so these two options can only be used in `initial` startup mode.
-2. Use `table-list` to match the exact value of database and table names.
+*注意*：连接器支持两种方式来指定需要监听的表，两种方式同时使用时会监听两种方式匹配的所有表。
+1. 使用 `database-name` 和 `table-name` 匹配正则表达式中的数据库和表名。 由于`obcdc`（以前的`liboblog`）现在只支持`fnmatch`匹配，我们不能直接使用正则过滤 changelog 事件，所以通过两个选项去匹配去指定监听表只能在`initial`启动模式下使用。
+2. 使用 `table-list` 去匹配数据库名和表名的准确列表。
<div class="highlight">
<table class="colwidths-auto docutils">
<thead>
<tr>
- <th class="text-left" style="width: 10%">Option</th>
- <th class="text-left" style="width: 8%">Required</th>
- <th class="text-left" style="width: 7%">Default</th>
- <th class="text-left" style="width: 10%">Type</th>
- <th class="text-left" style="width: 65%">Description</th>
+ <th class="text-left" style="width: 10%">配置项</th>
+ <th class="text-left" style="width: 8%">是否必选</th>
+ <th class="text-left" style="width: 7%">默认值</th>
+ <th class="text-left" style="width: 10%">类型</th>
+ <th class="text-left" style="width: 65%">描述</th>
</tr>
</thead>
<tbody>
<tr>
<td>connector</td>
- <td>required</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">无</td>
<td>String</td>
- <td>Specify what connector to use, here should be <code>'oceanbase-cdc'</code>.</td>
+ <td>指定要使用的连接器，此处为 <code>'oceanbase-cdc'</code>。</td>
</tr>
<tr>
<td>scan.startup.mode</td>
- <td>required</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">无</td>
<td>String</td>
- <td>Specify the startup mode for OceanBase CDC consumer, valid enumerations are
- <code>'initial'</code>,<code>'latest-offset'</code> or <code>'timestamp'</code>.
- </td>
+ <td>指定 OceanBase CDC 消费者的启动模式。可取值为<code>'initial'</code>,<code>'latest-offset'</code> or
+ <code>'timestamp'</code>。</td>
</tr>
<tr>
<td>scan.startup.timestamp</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">无</td>
<td>Long</td>
- <td>Timestamp in seconds of the start point, only used for <code>'timestamp'</code> startup mode.</td>
+ <td>起始点的时间戳，单位为秒。仅在启动模式为 <code>'timestamp'</code> 时可用。</td>
</tr>
<tr>
<td>username</td>
- <td>required</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">无</td>
<td>String</td>
- <td>Username to be used when connecting to OceanBase.</td>
+ <td>连接 OceanBase 数据库的用户的名称。</td>
</tr>
<tr>
<td>password</td>
- <td>required</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">无</td>
<td>String</td>
- <td>Password to be used when connecting to OceanBase.</td>
+ <td>连接 OceanBase 数据库时使用的密码。</td>
</tr>
<tr>
<td>tenant-name</td>
- <td>required</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">无</td>
<td>String</td>
- <td>Tenant name of OceanBase to monitor, should be exact value.</td>
+ <td>待监控 OceanBase 数据库的租户名，应该填入精确值。</td>
</tr>
<tr>
<td>database-name</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">无</td>
<td>String</td>
- <td>Database name of OceanBase to monitor, should be regular expression. Only can be used with 'initial' mode.</td>
+ <td>待监控 OceanBase 数据库的数据库名，应该是正则表达式，该选项只支持和 'initial' 模式一起使用。</td>
</tr>
<tr>
<td>table-name</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">无</td>
<td>String</td>
- <td>Table name of OceanBase to monitor, should be regular expression. Only can be used with 'initial' mode.</td>
+ <td>待监控 OceanBase 数据库的表名，应该是正则表达式，该选项只支持和 'initial' 模式一起使用。</td>
</tr>
<tr>
<td>table-list</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">无</td>
<td>String</td>
- <td>List of full names of tables, separated by commas, e.g. "db1.table1, db2.table2".</td>
+ <td>待监控 OceanBase 数据库的全路径的表名列表，逗号分隔，如："db1.table1, db2.table2"。</td>
</tr>
<tr>
<td>hostname</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">无</td>
<td>String</td>
- <td>IP address or hostname of the OceanBase database server or OceanBase Proxy server.</td>
+ <td>OceanBase 数据库或 OceanBbase 代理 ODP 的 IP 地址或主机名。</td>
</tr>
<tr>
<td>port</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">无</td>
<td>Integer</td>
- <td>Integer port number to connect to OceanBase. It can be the SQL port of OceanBase server, which is 2881 by default, or the port of OceanBase proxy service, which is 2883 by default.</td>
+ <td>
+ OceanBase 数据库服务器的整数端口号。可以是 OceanBase 服务器的 SQL 端口号（默认值为 2881）<br>
+ 或 OceanBase代理服务的端口号（默认值为 2883）</td>
</tr>
<tr>
<td>connect.timeout</td>
- <td>optional</td>
+ <td>否</td>
<td style="word-wrap: break-word;">30s</td>
<td>Duration</td>
- <td>The maximum time that the connector should wait after trying to connect to the OceanBase database server before timing out.</td>
+ <td>连接器在尝试连接到 OceanBase 数据库服务器超时前的最长时间。</td>
</tr>
<tr>
<td>server-time-zone</td>
- <td>optional</td>
+ <td>否</td>
<td style="word-wrap: break-word;">+00:00</td>
<td>String</td>
- <td>The session timezone which controls how temporal types are converted to STRING in OceanBase. Can be UTC offset in format "±hh:mm", or named time zones if the time zone information tables in the mysql database have been created and populated.</td>
+ <td>
+ 数据库服务器中的会话时区，用户控制 OceanBase 的时间类型如何转换为 STRING。<br>
+ 合法的值可以是格式为"±hh:mm"的 UTC 时区偏移量，<br>
+ 如果 mysql 数据库中的时区信息表已创建，合法的值则可以是创建的时区。
+ </td>
</tr>
<tr>
<td>logproxy.host</td>
- <td>required</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">无</td>
<td>String</td>
- <td>Hostname or IP address of OceanBase log proxy service.</td>
+ <td>OceanBase 日志代理服务 的 IP 地址或主机名。</td>
</tr>
<tr>
<td>logproxy.port</td>
- <td>required</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>是</td>
+ <td style="word-wrap: break-word;">无</td>
<td>Integer</td>
- <td>Port number of OceanBase log proxy service.</td>
+ <td>OceanBase 日志代理服务 的端口号。</td>
</tr>
<tr>
<td>logproxy.client.id</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">By rule.</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">规则生成</td>
<td>String</td>
- <td>Id of a log proxy client connection, will be in format {flink_ip}_{process_id}_{timestamp}_{thread_id}_{tenant} by default.</td>
+ <td>OceanBase日志代理服务的客户端连接 ID，默认值的生成规则是 {flink_ip}_{process_id}_{timestamp}_{thread_id}_{tenant}。</td>
</tr>
<tr>
<td>rootserver-list</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">无</td>
<td>String</td>
- <td>The semicolon-separated list of OceanBase root servers in format `ip:rpc_port:sql_port`, required for OceanBase CE.</td>
+ <td>OceanBase root 服务器列表，服务器格式为 `ip:rpc_port:sql_port`，<br>多个服务器地址使用英文分号 `;` 隔开，OceanBase 社区版本必填。</td>
</tr>
<tr>
<td>config-url</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">无</td>
<td>String</td>
- <td>The url to get the server info from the config server, required for OceanBase EE.</td>
+ <td>从配置服务器获取服务器信息的 url, OceanBase 企业版本必填。</td>
</tr>
<tr>
<td>working-mode</td>
- <td>optional</td>
+ <td>否</td>
<td style="word-wrap: break-word;">storage</td>
<td>String</td>
- <td>Working mode of `obcdc` in LogProxy, can be `storage` or `memory`.</td>
+ <td>日志代理中 `libobcdc` 的工作模式 , 可以是 `storage` 或 `memory`。</td>
</tr>
<tr>
<td>compatible-mode</td>
- <td>optional</td>
+ <td>否</td>
<td style="word-wrap: break-word;">mysql</td>
<td>String</td>
- <td>Compatible mode of OceanBase, can be `mysql` or `oracle`.</td>
+ <td>OceanBase 的兼容模式，可以是 `mysql` 或 `oracle`。</td>
</tr>
<tr>
<td>jdbc.driver</td>
- <td>optional</td>
+ <td>否</td>
<td style="word-wrap: break-word;">com.mysql.jdbc.Driver</td>
<td>String</td>
- <td>JDBC driver class for snapshot reading.</td>
+ <td>全量读取时使用的 jdbc 驱动类名。</td>
</tr>
<tr>
<td>jdbc.properties.*</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">(none)</td>
- <td>String</td>
- <td>Option to pass custom JDBC URL properties. User can pass custom properties like 'jdbc.properties.useSSL' = 'false'.</td>
- </tr>
- <tr>
- <td>obcdc.properties.*</td>
- <td>optional</td>
- <td style="word-wrap: break-word;">(none)</td>
+ <td>否</td>
+ <td style="word-wrap: break-word;">无</td>
<td>String</td>
- <td>Option to pass custom configurations to the <code>libobcdc</code>, eg: 'obcdc.properties.sort_trans_participants' = '1'. Please refer to <a href="https://en.oceanbase.com/docs/common-oceanbase-database-10000000000872541">obcdc parameters</a> for more details.</td>
+ <td>传递自定义 JDBC URL 属性的选项。用户可以传递自定义属性，如 'jdbc.properties.useSSL' = 'false'。</td>
</tr>
</tbody>
</table>
</div>
-Available Metadata
-----------------
+## 支持的元数据
-The following format metadata can be exposed as read-only (VIRTUAL) columns in a table definition.
+在创建表时，您可以使用以下格式的元数据作为只读列（VIRTUAL）。
<table class="colwidths-auto docutils">
<thead>
<tr>
- <th class="text-left" style="width: 15%">Key</th>
- <th class="text-left" style="width: 30%">DataType</th>
- <th class="text-left" style="width: 55%">Description</th>
+ <th class="text-left" style="width: 15%">列名</th>
+ <th class="text-left" style="width: 30%">数据类型</th>
+ <th class="text-left" style="width: 55%">描述</th>
</tr>
</thead>
<tbody>
<tr>
<td>tenant_name</td>
<td>STRING NOT NULL</td>
- <td>Name of the tenant that contains the row.</td>
+ <td>当前记录所属的租户名称。</td>
</tr>
<tr>
<td>database_name</td>
<td>STRING NOT NULL</td>
- <td>Name of the database that contains the row.</td>
+ <td>当前记录所属的库名。</td>
</tr>
<tr>
<td>table_name</td>
<td>STRING NOT NULL</td>
- <td>Name of the table that contains the row.</td>
+ <td>当前记录所属的表名称。</td>
</tr>
<tr>
<td>op_ts</td>
<td>TIMESTAMP_LTZ(3) NOT NULL</td>
- <td>It indicates the time that the change was made in the database. <br>
- If the record is read from snapshot of the table instead of the change stream, the value is always 0.</td>
+ <td>该值表示此修改在数据库中发生的时间。如果这条记录是该表在快照阶段读取的记录，则该值返回 0。</td>
</tr>
</tbody>
</table>
-The extended CREATE TABLE example demonstrates the syntax for exposing these metadata fields:
+如下 SQL 展示了如何在表中使用这些元数据列：
```sql
CREATE TABLE products (
@@ -409,35 +406,32 @@ CREATE TABLE products (
'port' = '2881',
'rootserver-list' = '127.0.0.1:2882:2881',
'logproxy.host' = '127.0.0.1',
- 'logproxy.port' = '2983',
- 'working-mode' = 'memory'
-);
+ 'logproxy.port' = '2983');
```
-Features
---------
+## 特性
-### At-Least-Once Processing
+### At-Least-Once 处理
-The OceanBase CDC connector is a Flink Source connector which will read database snapshot first and then continues to read change events with **at-least-once processing**.
+OceanBase CDC 连接器是一个 Flink Source 连接器。它将首先读取数据库快照，然后再读取变化事件，并进行 **At-Least-Once 处理**。
-OceanBase is a kind of distributed database whose log files are distributed on different servers. As there is no position information like MySQL binlog offset, we can only use timestamp as the position mark. In order to ensure the completeness of reading data, `liboblog` (a C++ library to read OceanBase log record) might read some log data before the given timestamp. So in this way we may read duplicate data whose timestamp is around the start point, and only 'at-least-once' can be guaranteed.
+OceanBase 数据库是一个分布式数据库，它的日志也分散在不同的服务器上。由于没有类似 MySQL binlog 偏移量的位置信息，OceanBase 数据库用时间戳作为位置标记。为确保读取完整的数据，liboblog（读取 OceanBase 日志记录的 C++ 库）可能会在给定的时间戳之前读取一些日志数据。因此，OceanBase 数据库可能会读到起始点附近时间戳的重复数据，可保证 **At-Least-Once 处理**。
-### Startup Reading Position
+### 启动模式
-The config option `scan.startup.mode` specifies the startup mode for OceanBase CDC consumer. The valid enumerations are:
+配置选项 `scan.startup.mode` 指定 OceanBase CDC 连接器的启动模式。可用取值包括：
-- `initial`: Performs an initial snapshot on the monitored table upon first startup, and continue to read the latest commit log.
-- `latest-offset`: Never to perform snapshot on the monitored table upon first startup and just read the latest commit log since the connector is started.
-- `timestamp`: Never to perform snapshot on the monitored table upon first startup and just read the commit log from the given `scan.startup.timestamp`.
+- `initial`（默认）：在首次启动时对受监视的数据库表执行初始快照，并继续读取最新的提交日志。
+- `latest-offset`：首次启动时，不对受监视的数据库表执行快照，仅从连接器启动时读取提交日志。
+- `timestamp`：在首次启动时不对受监视的数据库表执行初始快照，仅从指定的 `scan.startup.timestamp` 读取最新的提交日志。
-### Consume Commit Log
+### 消费提交日志
-The OceanBase CDC Connector using [oblogclient](https://github.com/oceanbase/oblogclient) to consume commit log from OceanBase LogProxy.
+OceanBase CDC 连接器使用 [oblogclient](https://github.com/oceanbase/oblogclient) 消费 OceanBase日志代理服务 中的事务日志。
### DataStream Source
-The OceanBase CDC connector can also be a DataStream source. You can create a SourceFunction as the following shows:
+OceanBase CDC 连接器也可以作为 DataStream Source 使用。您可以按照如下创建一个 SourceFunction：
```java
import org.apache.flink.api.common.typeinfo.TypeInformation;
@@ -451,10 +445,10 @@ import org.apache.flink.table.data.RowData;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.table.types.logical.RowType;
-import org.apache.flink.cdc.connectors.oceanbase.OceanBaseSource;
-import org.apache.flink.cdc.connectors.oceanbase.source.RowDataOceanBaseDeserializationSchema;
-import org.apache.flink.cdc.connectors.oceanbase.table.OceanBaseDeserializationSchema;
-import org.apache.flink.cdc.connectors.oceanbase.table.StartupMode;
+import com.ververica.cdc.connectors.oceanbase.OceanBaseSource;
+import com.ververica.cdc.connectors.oceanbase.source.RowDataOceanBaseDeserializationSchema;
+import com.ververica.cdc.connectors.oceanbase.table.OceanBaseDeserializationSchema;
+import com.ververica.cdc.connectors.oceanbase.table.StartupMode;
import java.time.ZoneId;
import java.util.Arrays;
@@ -511,25 +505,27 @@ public class OceanBaseSourceExample {
}
}
```
-Data Type Mapping
-----------------
-### Mysql Mode
+## 数据类型映射
+
+### Mysql 模式
<div class="wy-table-responsive">
<table class="colwidths-auto docutils">
<thead>
<tr>
- <th class="text-left">OceanBase type</th>
- <th class="text-left">Flink SQL type</th>
- <th class="text-left">NOTE</th>
+ <th class="text-left">OceanBase 数据类型</th>
+ <th class="text-left">Flink SQL 类型</th>
+ <th class="text-left">描述</th>
</tr>
</thead>
<tbody>
<tr>
- <td>BOOLEAN<br>
+ <td>
+ BOOLEAN<br>
TINYINT(1)<br>
- BIT(1)</td>
+ BIT(1)
+ </td>
<td>BOOLEAN</td>
<td></td>
</tr>
@@ -541,7 +537,8 @@ Data Type Mapping
<tr>
<td>
SMALLINT<br>
- TINYINT UNSIGNED</td>
+ TINYINT UNSIGNED
+ </td>
<td>SMALLINT</td>
<td></td>
</tr>
@@ -549,14 +546,16 @@ Data Type Mapping
<td>
INT<br>
MEDIUMINT<br>
- SMALLINT UNSIGNED</td>
+ SMALLINT UNSIGNED
+ </td>
<td>INT</td>
<td></td>
</tr>
<tr>
<td>
BIGINT<br>
- INT UNSIGNED</td>
+ INT UNSIGNED
+ </td>
<td>BIGINT</td>
<td></td>
</tr>
@@ -568,7 +567,7 @@ Data Type Mapping
<tr>
<td>
REAL<br>
- FLOAT<br>
+ FLOAT
</td>
<td>FLOAT</td>
<td></td>
@@ -584,8 +583,7 @@ Data Type Mapping
<td>
NUMERIC(p, s)<br>
DECIMAL(p, s)<br>
- where p <= 38<br>
- </td>
+ where p <= 38 </td>
<td>DECIMAL(p, s)</td>
<td></td>
</tr>
@@ -593,13 +591,13 @@ Data Type Mapping
<td>
NUMERIC(p, s)<br>
DECIMAL(p, s)<br>
- where 38 < p <=65<br>
- </td>
+ where 38 < p <=65 </td>
<td>STRING</td>
- <td>DECIMAL is equivalent to NUMERIC. The precision for DECIMAL data type is up to 65 in OceanBase, but
- the precision for DECIMAL is limited to 38 in Flink.
- So if you define a decimal column whose precision is greater than 38, you should map it to STRING to
- avoid precision loss.</td>
+ <td>
+ DECIMAL 等同于 NUMERIC。在 OceanBase 数据库中，DECIMAL 数据类型的精度最高为 65。<br>
+ 但在 Flink 中，DECIMAL 的最高精度为 38。因此，<br>
+ 如果你定义了一个精度大于 38 的 DECIMAL 列，你应当将其映射为 STRING，以避免精度损失。
+ </td>
</tr>
<tr>
<td>DATE</td>
@@ -633,7 +631,7 @@ Data Type Mapping
</tr>
<tr>
<td>BIT(n)</td>
- <td>BINARY(⌈(n + 7) / 8⌉)</td>
+ <td>BINARY(⌈n/8⌉)</td>
<td></td>
</tr>
<tr>
@@ -651,7 +649,7 @@ Data Type Mapping
TINYTEXT<br>
TEXT<br>
MEDIUMTEXT<br>
- LONGTEXT<br>
+ LONGTEXT
</td>
<td>STRING</td>
<td></td>
@@ -661,7 +659,7 @@ Data Type Mapping
TINYBLOB<br>
BLOB<br>
MEDIUMBLOB<br>
- LONGBLOB<br>
+ LONGBLOB
</td>
<td>BYTES</td>
<td></td>
@@ -679,18 +677,21 @@ Data Type Mapping
<tr>
<td>SET</td>
<td>ARRAY<STRING></td>
- <td>As the SET data type in OceanBase is a string object that can have zero or more values, it should always be mapped to an array of string</td>
+ <td>
+ 因为 OceanBase 的 SET 类型是用包含一个或多个值的字符串对象表示，<br>
+ 所以映射到 Flink 时是一个字符串数组
+ </td>
</tr>
<tr>
<td>JSON</td>
<td>STRING</td>
- <td>The JSON data type will be converted into STRING with JSON format in Flink.</td>
+ <td>JSON 类型的数据在 Flink 中会转化为 JSON 格式的字符串</td>
</tr>
</tbody>
</table>
</div>
-### Oracle Mode
+### Oracle 模式
<div class="wy-table-responsive">
<table class="colwidths-auto docutils">
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/overview.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/overview.md
index 3d22e69a3..5a485e946 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/overview.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/overview.md
@@ -24,12 +24,12 @@ specific language governing permissions and limitations
under the License.
-->
-# Legacy CDC sources for Apache Flink
+# Legacy CDC sources
-Flink CDC sources is a set of source connectors for <a href="https://flink.apache.org/">Apache Flink<sup>®</sup></a>, ingesting changes from different databases using change data capture (CDC).
+Legacy CDC sources is a set of source connectors for <a href="https://flink.apache.org/">Apache Flink<sup>®</sup></a>, ingesting changes from different databases using change data capture (CDC).
Some CDC sources integrate Debezium as the engine to capture data changes. So it can fully leverage the ability of Debezium. See more about what is [Debezium](https://github.com/debezium/debezium).
-You can also read [tutorials]({{< ref "docs/connectors/legacy-flink-cdc-sources/tutorials/mysql-postgres-tutorial" >}}) about how to use these sources.
+You can also read [tutorials]({{< ref "docs/connectors/legacy-flink-cdc-sources/tutorials/build-streaming-etl-tutorial" >}}) about how to use these sources.
{{< img src="/fig/cdc-flow.png" width="600px" alt="Flink CDC" >}}
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/build-real-time-data-lake-tutorial.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/build-real-time-data-lake-tutorial.md
index ff45befbf..0f2fe0a9b 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/build-real-time-data-lake-tutorial.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/build-real-time-data-lake-tutorial.md
@@ -1,6 +1,6 @@
---
title: "使用 Flink CDC 构建实时数据湖"
-weight: 999
+weight: 998
type: docs
aliases:
- /connectors/legacy-flink-cdc-sources/tutorials/build-real-time-data-lake-tutorial.html
@@ -26,130 +26,117 @@ specific language governing permissions and limitations
under the License.
-->
-# Using Flink CDC to synchronize data from MySQL sharding tables and build real-time data lake
+# 使用 Flink CDC 构建实时数据湖
-For OLTP databases, to deal with a huge number of data in a single table, we usually do database and table sharding to get better throughput.
-But sometimes, for convenient analysis, we need to merge them into one table when loading them to data warehouse or data lake.
+在 OLTP 系统中，为了解决单表数据量大的问题，通常采用分库分表的方式将单个大表进行拆分以提高系统的吞吐量。
+但是为了方便数据分析，通常需要将分库分表拆分出的表在同步到数据仓库、数据湖时，再合并成一个大表。
-This tutorial will show how to use Flink CDC to build a real-time data lake for such a scenario.
-You can walk through the tutorial easily in the docker environment. The entire process uses standard SQL syntax without a single line of Java/Scala code or IDE installation.
+这篇教程将展示如何使用 Flink CDC 构建实时数据湖来应对这种场景，本教程的演示基于 Docker，只涉及 SQL，无需一行 Java/Scala 代码，也无需安装 IDE，你可以很方便地在自己的电脑上完成本教程的全部内容。
-The following sections will take the pipeline from MySQL to [Iceberg](https://iceberg.apache.org/) as an example. The overview of the architecture is as follows:
+接下来将以数据从 MySQL 同步到 [Iceberg](https://iceberg.apache.org/) 为例展示整个流程，架构图如下所示：
{{< img src="/fig/real-time-data-lake-tutorial/real-time-data-lake-tutorial.png" alt="Real-time data lake with Flink CDC" >}}
-You can also use other data sources like Oracle/Postgres and sinks like Hudi to build your own pipeline.
+你也可以使用不同的 source 比如 Oracle/Postgres 和 sink 比如 Hudi 来构建自己的 ETL 流程。
-## Preparation
-Prepare a Linux or MacOS computer with Docker installed.
-
-## Preparing JAR package required
-**Download links are available only for stable releases, SNAPSHOT dependencies need to be built based on master or release-branches by yourself.**
-- flink-sql-connector-mysql-cdc-3.0-SNAPSHOT.jar
-- [flink-shaded-hadoop-2-uber-2.7.5-10.0.jar](https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.7.5-10.0/flink-shaded-hadoop-2-uber-2.7.5-10.0.jar)
-- [iceberg-flink-runtime-1.16-1.3.1.jar](https://repo.maven.apache.org/maven2/org/apache/iceberg/iceberg-flink-runtime-1.16/1.3.1/iceberg-flink-runtime-1.16-1.3.1.jar)
+## 准备阶段
+准备一台已经安装了 Docker 的 Linux 或者 MacOS 电脑。
+### 准备教程所需要的组件
+接下来的教程将以 `docker-compose` 的方式准备所需要的组件。
+使用下面的内容创建一个 `docker-compose.yml` 文件：
+```
+version: '2.1'
+services:
+ sql-client:
+ user: flink:flink
+ image: yuxialuo/flink-sql-client:1.13.2.v1
+ depends_on:
+ - jobmanager
+ - mysql
+ environment:
+ FLINK_JOBMANAGER_HOST: jobmanager
+ MYSQL_HOST: mysql
+ volumes:
+ - shared-tmpfs:/tmp/iceberg
+ jobmanager:
+ user: flink:flink
+ image: flink:1.13.2-scala_2.11
+ ports:
+ - "8081:8081"
+ command: jobmanager
+ environment:
+ - |
+ FLINK_PROPERTIES=
+ jobmanager.rpc.address: jobmanager
+ volumes:
+ - shared-tmpfs:/tmp/iceberg
+ taskmanager:
+ user: flink:flink
+ image: flink:1.13.2-scala_2.11
+ depends_on:
+ - jobmanager
+ command: taskmanager
+ environment:
+ - |
+ FLINK_PROPERTIES=
+ jobmanager.rpc.address: jobmanager
+ taskmanager.numberOfTaskSlots: 2
+ volumes:
+ - shared-tmpfs:/tmp/iceberg
+ mysql:
+ image: debezium/example-mysql:1.1
+ ports:
+ - "3306:3306"
+ environment:
+ - MYSQL_ROOT_PASSWORD=123456
+ - MYSQL_USER=mysqluser
+ - MYSQL_PASSWORD=mysqlpw
+
+volumes:
+ shared-tmpfs:
+ driver: local
+ driver_opts:
+ type: "tmpfs"
+ device: "tmpfs"
+```
-### Starting components required
-The components required in this tutorial are all managed in containers, so we will use `docker-compose` to start them.
+该 Docker Compose 中包含的容器有：
+- SQL-Client: Flink SQL Client, 用来提交 SQL 查询和查看 SQL 的执行结果
+- Flink Cluster：包含 Flink JobManager 和 Flink TaskManager，用来执行 Flink SQL
+- MySQL：作为分库分表的数据源，存储本教程的 `user` 表
-1. Create `Dockerfile` file using following contents:
- ```dockerfile
- FROM flink:1.16.0-scala_2.12
- # Place the downloaded jar packages in the lib directory at the same level.
- COPY ./lib /opt/flink/lib
- RUN apt-get update && apt-get install tree
- ```
+***注意：***
+1. 为了简化整个教程，本教程需要的 jar 包都已经被打包进 SQL-Client 容器中了，镜像的构建脚本可以在 [GitHub](https://github.com/luoyuxia/flink-cdc-tutorial/tree/main/flink-cdc-iceberg-demo/sql-client) 上找到。
+ 如果你想要在自己的 Flink 环境运行本教程，需要下载下面列出的包并且把它们放在 Flink 所在目录的 lib 目录下，即 `FLINK_HOME/lib/`。
-2. Create `docker-compose.yml` file using following contents:
- ```yml
- version: '2.1'
- services:
- sql-client:
- user: flink:flink
- build: .
- command: bin/sql-client.sh
- depends_on:
- - jobmanager
- - mysql
- environment:
- - MYSQL_HOST=mysql
- - |
- FLINK_PROPERTIES=
- jobmanager.rpc.address: jobmanager
- rest.address: jobmanager
- volumes:
- - shared-tmpfs:/tmp/iceberg
- jobmanager:
- user: flink:flink
- build: .
- ports:
- - "8081:8081"
- command: jobmanager
- environment:
- - |
- FLINK_PROPERTIES=
- jobmanager.rpc.address: jobmanager
- volumes:
- - shared-tmpfs:/tmp/iceberg
- taskmanager:
- user: flink:flink
- build: .
- depends_on:
- - jobmanager
- command: taskmanager
- environment:
- - |
- FLINK_PROPERTIES=
- jobmanager.rpc.address: jobmanager
- taskmanager.numberOfTaskSlots: 2
- volumes:
- - shared-tmpfs:/tmp/iceberg
- mysql:
- image: debezium/example-mysql:1.1
- ports:
- - "3306:3306"
- environment:
- - MYSQL_ROOT_PASSWORD=123456
- - MYSQL_USER=mysqluser
- - MYSQL_PASSWORD=mysqlpw
-
- volumes:
- shared-tmpfs:
- driver: local
- driver_opts:
- type: "tmpfs"
- device: "tmpfs"
- ```
+ **下载链接只对已发布的版本有效, SNAPSHOT 版本需要本地编译**
- The Docker Compose environment consists of the following containers:
- - SQL-Client: Flink SQL Client, used to submit queries and visualize their results.
- - Flink Cluster: a Flink JobManager and a Flink TaskManager container to execute queries.
- - MySQL: mainly used as a data source to store the sharding table.
+ - [flink-sql-connector-mysql-cdc-2.4.0.jar](https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.4.0/flink-sql-connector-mysql-cdc-2.4.0.jar)
+ - [flink-shaded-hadoop-2-uber-2.7.5-10.0.jar](https://repo.maven.apache.org/maven2/org/apache/flink/flink-shaded-hadoop-2-uber/2.7.5-10.0/flink-shaded-hadoop-2-uber-2.7.5-10.0.jar)
+ - [iceberg-flink-1.13-runtime-0.13.0-SNAPSHOT.jar](https://raw.githubusercontent.com/luoyuxia/flink-cdc-tutorial/main/flink-cdc-iceberg-demo/sql-client/lib/iceberg-flink-1.13-runtime-0.13.0-SNAPSHOT.jar)
-3. To start all containers, run the following command in the directory that contains the `docker-compose.yml` file:
- ```shell
- docker-compose up -d
- ```
- This command automatically starts all the containers defined in the Docker Compose configuration in a detached mode. Run `docker ps` to check whether these containers are running properly.
- We can also visit [http://localhost:8081/](http://localhost:8081/) to see if Flink is running normally.
+ 目前支持 Flink 1.13 的 `iceberg-flink-runtime` jar 包还没有发布，所以我们在这里提供了一个支持 Flink 1.13 的 `iceberg-flink-runtime` jar 包，这个 jar 包是基于 Iceberg 的 master 分支打包的。
+ 当 Iceberg 0.13.0 版本发布后，你也可以在 [apache official repository](https://repo.maven.apache.org/maven2/org/apache/iceberg/iceberg-flink-runtime/) 下载到支持 Flink 1.13 的 `iceberg-flink-runtime` jar 包。
+2. 本教程接下来用到的容器相关的命令都需要在 `docker-compose.yml` 所在目录下执行
-
-***Note:***
-* If you want to run with your own Flink environment, remember to download the jar packages and then put them to `FLINK_HOME/lib/`.
-* All the following commands involving `docker-compose` should be executed in the directory of the `docker-compose.yml` file.
+在 `docker-compose.yml` 所在目录下执行下面的命令来启动本教程需要的组件：
+```shell
+docker-compose up -d
+```
+该命令将以 detached 模式自动启动 Docker Compose 配置中定义的所有容器。你可以通过 `docker ps` 来观察上述的容器是否正常启动了，也可以通过访问 [http://localhost:8081/](http://localhost:8081//) 来查看 Flink 是否运行正常。
{{< img src="/fig/real-time-data-lake-tutorial/flink-ui.png" alt="Flink UI" >}}
-### Preparing data in databases
-1. Enter mysql's container:
+### 准备数据
+1. 进入 MySQL 容器中
```shell
docker-compose exec mysql mysql -uroot -p123456
```
-2. Create databases/tables and populate data:
+2. 创建数据和表，并填充数据
- Create a logical sharding table `user` sharded in different databases and tables physically.
+ 创建两个不同的数据库，并在每个数据库中创建两个表，作为 `user` 表分库分表下拆分出的表。
```sql
CREATE DATABASE db_1;
USE db_1;
@@ -193,30 +180,28 @@ The components required in this tutorial are all managed in containers, so we wi
INSERT INTO user_2 VALUES (220,"user_220","Shanghai","123567891234","user_220@foo.com");
```
-## Creating tables using Flink DDL in Flink SQL CLI
-First, use the following command to enter the Flink SQL CLI Container:
+## 在 Flink SQL CLI 中使用 Flink DDL 创建表
+首先，使用如下的命令进入 Flink SQL CLI 容器中：
```shell
-docker-compose run sql-client
+docker-compose exec sql-client ./sql-client
```
-
-We should see the welcome screen of the CLI client:
+我们可以看到如下界面：
{{< img src="/fig/real-time-data-lake-tutorial/flink-sql-client.png" alt="Flink SQL Client" >}}
-Then do the following steps in Flink SQL CLI:
+然后，进行如下步骤：
+1. 开启 checkpoint，每隔3秒做一次 checkpoint
-1. Enable checkpoints every 3 seconds
-
- Checkpoint is disabled by default, we need to enable it to commit Iceberg transactions.
- Besides, the beginning of mysql-cdc binlog phase also requires waiting a complete checkpoint to avoid disorder of binlog records.
+ Checkpoint 默认是不开启的，我们需要开启 Checkpoint 来让 Iceberg 可以提交事务。
+ 并且，mysql-cdc 在 binlog 读取阶段开始前，需要等待一个完整的 checkpoint 来避免 binlog 记录乱序的情况。
```sql
-- Flink SQL
Flink SQL> SET execution.checkpointing.interval = 3s;
```
-2. Create MySQL sharding source table
+2. 创建 MySQL 分库分表 source 表
- Create a source table that captures the data from the logical sharding table `user`. Here, we use regex to match all the physical tables.
- Besides, the table defines metadata column to identify which database/table the record comes from.
+ 创建 source 表 `user_source` 来捕获MySQL中所有 `user` 表的数据，在表的配置项 `database-name` , `table-name` 使用正则表达式来匹配这些表。
+ 并且，`user_source` 表也定义了 metadata 列来区分数据是来自哪个数据库和表。
```sql
-- Flink SQL
Flink SQL> CREATE TABLE user_source (
@@ -238,10 +223,10 @@ Then do the following steps in Flink SQL CLI:
'table-name' = 'user_[0-9]+'
);
```
-3. Create Iceberg sink table
+3. 创建 Iceberg sink 表
- Create a sink table `all_users_sink` used to load data to Iceberg.
- We define `database_name`, `table_name` and `id` as a combined primary key, because `id` maybe not unique across different databases and tables.
+ 创建 sink 表 `all_users_sink`，用来将数据加载至 Iceberg 中。
+ 在这个 sink 表，考虑到不同的 MySQL 数据库表的 `id` 字段的值可能相同，我们定义了复合主键 (`database_name`, `table_name`, `id`)。
```sql
-- Flink SQL
Flink SQL> CREATE TABLE all_users_sink (
@@ -262,68 +247,66 @@ Then do the following steps in Flink SQL CLI:
);
```
-## Streaming to Iceberg
-1. Streaming write data from MySQL to Iceberg using the following Flink SQL:
+## 流式写入 Iceberg
+
+1. 使用下面的 Flink SQL 语句将数据从 MySQL 写入 Iceberg 中
```sql
-- Flink SQL
Flink SQL> INSERT INTO all_users_sink select * from user_source;
```
- It will start a streaming job which will synchronize historical and incremental data from MySQL to Iceberg continuously.
- The running job can be found in [Flink UI](http://localhost:8081/#/job/running), and it looks like:
-
-
+ 上述命令将会启动一个流式作业，源源不断将 MySQL 数据库中的全量和增量数据同步到 Iceberg 中。
+ 在 [Flink UI](http://localhost:8081/#/job/running) 上可以看到这个运行的作业：
+
{{< img src="/fig/real-time-data-lake-tutorial/flink-cdc-iceberg-running-job.png" alt="CDC to Iceberg Running Job" >}}
-
- Then, we can use the following command to see the files written to Iceberg:
+
+ 然后我们就可以使用如下的命令看到 Iceberg 中的写入的文件：
```shell
docker-compose exec sql-client tree /tmp/iceberg/warehouse/default_database/
```
- It should look like:
+ 如下所示：
{{< img src="/fig/real-time-data-lake-tutorial/files-in-iceberg.png" alt="Files in Iceberg" >}}
- The actual files may differ in your environment, but the structure of the directory should be similar.
+ 在你的运行环境中，实际的文件可能与上面的截图不相同，但是整体的目录结构应该相似。
-2. Use the following Flink SQL to query the data written to `all_users_sink`:
+2. 使用下面的 Flink SQL 语句查询表 `all_users_sink` 中的数据
```sql
-- Flink SQL
Flink SQL> SELECT * FROM all_users_sink;
```
- We can see the data queried in the Flink SQL CLI:
-
+ 在 Flink SQL CLI 中我们可以看到如下查询结果：
+
{{< img src="/fig/real-time-data-lake-tutorial/data_in_iceberg.png" alt="Data in Iceberg" >}}
-
-3. Make some changes in the MySQL databases, and then the data in Iceberg table `all_users_sink` will also change in real time.
-
- (3.1) Insert a new user in table `db_1.user_1`
+
+3. 修改 MySQL 中表的数据，Iceberg 中的表 `all_users_sink` 中的数据也将实时更新：
+
+ (3.1) 在 `db_1.user_1` 表中插入新的一行
```sql
--- db_1
INSERT INTO db_1.user_1 VALUES (111,"user_111","Shanghai","123567891234","user_111@foo.com");
```
- (3.2) Update a user in table `db_1.user_2`
+ (3.2) 更新 `db_1.user_2` 表的数据
```sql
--- db_1
UPDATE db_1.user_2 SET address='Beijing' WHERE id=120;
```
- (3.3) Delete a user in table `db_2.user_2`
+ (3.3) 在 `db_2.user_2` 表中删除一行
```sql
--- db_2
DELETE FROM db_2.user_2 WHERE id=220;
```
+ 每执行一步，我们就可以在 Flink Client CLI 中使用 `SELECT * FROM all_users_sink` 查询表 `all_users_sink` 来看到数据的变化。
+
+ 最后的查询结果如下所示：
- After executing each step, we can query the table `all_users_sink` using `SELECT * FROM all_users_sink` in Flink SQL CLI to see the changes.
-
- The final query result is as follows:
-
{{< img src="/fig/real-time-data-lake-tutorial/final-data-in-iceberg.png" alt="Final Data in Iceberg" >}}
- From the latest result in Iceberg, we can see that there is a new record of `(db_1, user_1, 111)`, and the address of `(db_1, user_2, 120)` has been updated to `Beijing`.
- Besides, the record of `(db_2, user_2, 220)` has been deleted. The result is exactly the same with the changes we did in MySQL.
-
-## Clean up
-After finishing the tutorial, run the following command in the directory of `docker-compose.yml` to stop all containers:
+ 从 Iceberg 的最新结果中可以看到新增了`(db_1, user_1, 111)`的记录，`(db_1, user_2, 120)`的地址更新成了 `Beijing`，且`(db_2, user_2, 220)`的记录被删除了，与我们在 MySQL 做的数据更新完全一致。
+
+## 环境清理
+本教程结束后，在 `docker-compose.yml` 文件所在的目录下执行如下命令停止所有容器：
```shell
docker-compose down
```
diff --git a/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/mysql-postgres-tutorial.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/build-streaming-etl-tutorial.md
similarity index 59%
rename from docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/mysql-postgres-tutorial.md
rename to docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/build-streaming-etl-tutorial.md
index 41a218918..f6e2b0a58 100644
--- a/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/mysql-postgres-tutorial.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/build-streaming-etl-tutorial.md
@@ -1,6 +1,6 @@
---
-title: "MySQL & PostgreSQL Tutorial"
-weight: 2
+title: "使用 Flink CDC 构建 Streaming ETL"
+weight: 999
type: docs
aliases:
- /connectors/legacy-flink-cdc-sources/tutorials/mysql-postgres-tutorial.html
@@ -24,26 +24,23 @@ specific language governing permissions and limitations
under the License.
-->
-# Streaming ETL for MySQL and PostgreSQL with Flink CDC
+# 使用 Flink CDC 构建 Streaming ETL
-This tutorial is to show how to quickly build streaming ETL for MySQL and Postgres with Flink CDC.
+这篇教程将展示如何基于 Flink CDC 快速构建 MySQL 和 Postgres 的流式 ETL。本教程的演示都将在 Flink SQL CLI 中进行，只涉及 SQL，无需一行 Java/Scala 代码，也无需安装 IDE。
-Assuming we are running an e-commerce business. The product and order data stored in MySQL, the shipment data related to the order is stored in Postgres.
-We want to enrich the orders using the product and shipment table, and then load the enriched orders to ElasticSearch in real time.
+假设我们正在经营电子商务业务，商品和订单的数据存储在 MySQL 中，订单对应的物流信息存储在 Postgres 中。
+对于订单表，为了方便进行分析，我们希望让它关联上其对应的商品和物流信息，构成一张宽表，并且实时把它写到 ElasticSearch 中。
-In the following sections, we will describe how to use Flink Mysql/Postgres CDC to implement it.
-All exercises in this tutorial are performed in the Flink SQL CLI, and the entire process uses standard SQL syntax, without a single line of Java/Scala code or IDE installation.
-
-The overview of the architecture is as follows:
+接下来的内容将介绍如何使用 Flink Mysql/Postgres CDC 来实现这个需求，系统的整体架构如下图所示：
{{< img src="/fig/mysql-postgres-tutorial/flink-cdc-streaming-etl.png" width="700px" alt="Flink CDC Streaming ETL" >}}
-## Preparation
-Prepare a Linux or MacOS computer with Docker installed.
+## 准备阶段
+准备一台已经安装了 Docker 的 Linux 或者 MacOS 电脑。
-### Starting components required
-The components required in this demo are all managed in containers, so we will use `docker-compose` to start them.
+### 准备教程所需要的组件
+接下来的教程将以 `docker-compose` 的方式准备所需要的组件。
-Create `docker-compose.yml` file using following contents:
+使用下面的内容创建一个 `docker-compose.yml` 文件：
```
version: '2.1'
services:
@@ -85,35 +82,34 @@ services:
ports:
- "5601:5601"
```
-The Docker Compose environment consists of the following containers:
-- MySQL: the `products`,`orders` tables will be store in the database. They will be joined with data in Postgres to enrich the orders.
-- Postgres: the `shipments` table will be store in the database.
-- Elasticsearch: mainly used as a data sink to store enriched orders.
-- Kibana: used to visualize the data in Elasticsearch.
+该 Docker Compose 中包含的容器有：
+- MySQL: 商品表 `products` 和 订单表 `orders` 将存储在该数据库中， 这两张表将和 Postgres 数据库中的物流表 `shipments`进行关联，得到一张包含更多信息的订单表 `enriched_orders`
+- Postgres: 物流表 `shipments` 将存储在该数据库中
+- Elasticsearch: 最终的订单表 `enriched_orders` 将写到 Elasticsearch
+- Kibana: 用来可视化 ElasticSearch 的数据
-To start all containers, run the following command in the directory that contains the `docker-compose.yml` file.
+在 `docker-compose.yml` 所在目录下执行下面的命令来启动本教程需要的组件：
```shell
docker-compose up -d
```
-This command automatically starts all the containers defined in the Docker Compose configuration in a detached mode. Run docker ps to check whether these containers are running properly.
-We can also visit [http://localhost:5601/](http://localhost:5601/) to see if Kibana is running normally.
+该命令将以 detached 模式自动启动 Docker Compose 配置中定义的所有容器。你可以通过 docker ps 来观察上述的容器是否正常启动了，也可以通过访问 [http://localhost:5601/](http://localhost:5601/) 来查看 Kibana 是否运行正常。
-### Preparing Flink and JAR package required
-1. Download [Flink 1.18.0](https://archive.apache.org/dist/flink/flink-1.18.0/flink-1.18.0-bin-scala_2.12.tgz) and unzip it to the directory `flink-1.18.0`
-2. Download following JAR package required and put them under `flink-1.18.0/lib/`:
+### 下载 Flink 和所需要的依赖包
+1. 下载 [Flink 1.17.0](https://archive.apache.org/dist/flink/flink-1.17.0/flink-1.17.0-bin-scala_2.12.tgz) 并将其解压至目录 `flink-1.17.0`
+2. 下载下面列出的依赖包，并将它们放到目录 `flink-1.17.0/lib/` 下：
- **Download links are available only for stable releases, SNAPSHOT dependencies need to be built based on master or release branches by yourself.**
- - [flink-sql-connector-elasticsearch7-3.0.1-1.17.jar](https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-elasticsearch7/3.0.1-1.17/flink-sql-connector-elasticsearch7-3.0.1-1.17.jar)
- - flink-sql-connector-mysql-cdc-3.0-SNAPSHOT.jar
- - flink-sql-connector-postgres-cdc-3.0-SNAPSHOT.jar
+ **下载链接只对已发布的版本有效, SNAPSHOT 版本需要本地编译**
+ - [flink-sql-connector-elasticsearch7-3.0.1-1.17.jar](https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-elasticsearch7/3.0.1-1.17/flink-sql-connector-elasticsearch7-3.0.1-1.17.jar)
+ - [flink-sql-connector-mysql-cdc-2.4.0.jar](https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.4.0/flink-sql-connector-mysql-cdc-2.4.0.jar)
+ - [flink-sql-connector-postgres-cdc-2.4.0.jar](https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-postgres-cdc/2.4.0/flink-sql-connector-postgres-cdc-2.4.0.jar)
-### Preparing data in databases
-#### Preparing data in MySQL
-1. Enter mysql's container:
+### 准备数据
+#### 在 MySQL 数据库中准备数据
+1. 进入 MySQL 容器
```shell
docker-compose exec mysql mysql -uroot -p123456
```
-2. Create tables and populate data:
+2. 创建数据库和表 `products`，`orders`，并插入数据
```sql
-- MySQL
CREATE DATABASE mydb;
@@ -150,12 +146,12 @@ We can also visit [http://localhost:5601/](http://localhost:5601/) to see if Kib
(default, '2020-07-30 10:11:09', 'Sally', 15.00, 105, false),
(default, '2020-07-30 12:00:30', 'Edward', 25.25, 106, false);
```
-#### Preparing data in Postgres
-1. Enter postgres's container:
+#### 在 Postgres 数据库中准备数据
+1. 进入 Postgres 容器
```shell
docker-compose exec postgres psql -h localhost -U postgres
```
-2. Create tables and populate data
+2. 创建表 `shipments`，并插入数据
```sql
-- PG
CREATE TABLE shipments (
@@ -172,38 +168,38 @@ We can also visit [http://localhost:5601/](http://localhost:5601/) to see if Kib
(default,10002,'Hangzhou','Shanghai',false),
(default,10003,'Shanghai','Hangzhou',false);
```
-
-## Starting Flink cluster and Flink SQL CLI
-1. Use the following command to change to the Flink directory:
+## 启动 Flink 集群和 Flink SQL CLI
+
+1. 使用下面的命令跳转至 Flink 目录下
```
- cd flink-1.18.0
+ cd flink-1.17.0
```
-
-2. Use the following command to start a Flink cluster:
+
+2. 使用下面的命令启动 Flink 集群
```shell
./bin/start-cluster.sh
```
- Then we can visit [http://localhost:8081/](http://localhost:8081/) to see if Flink is running normally, and the web page looks like:
+ 启动成功的话，可以在 [http://localhost:8081/](http://localhost:8081/) 访问到 Flink Web UI，如下所示：
{{< img src="/fig/mysql-postgres-tutorial/flink-ui.png" width="700px" alt="Flink UI" >}}
-3. Use the following command to start a Flink SQL CLI:
+3. 使用下面的命令启动 Flink SQL CLI
```shell
./bin/sql-client.sh
```
- We should see the welcome screen of the CLI client.
+ 启动成功后，可以看到如下的页面：
{{< img src="/fig/mysql-postgres-tutorial/flink-sql-client.png" alt="Flink SQL Client" >}}
-## Creating tables using Flink DDL in Flink SQL CLI
-First, enable checkpoints every 3 seconds
+## 在 Flink SQL CLI 中使用 Flink DDL 创建表
+首先，开启 checkpoint，每隔3秒做一次 checkpoint
```sql
-- Flink SQL
Flink SQL> SET execution.checkpointing.interval = 3s;
```
-Then, create tables that capture the change data from the corresponding database tables.
+然后, 对于数据库中的表 `products`, `orders`, `shipments`， 使用 Flink SQL CLI 创建对应的表，用于同步这些底层数据库表的数据
```sql
-- Flink SQL
Flink SQL> CREATE TABLE products (
@@ -259,7 +255,7 @@ Flink SQL> CREATE TABLE shipments (
);
```
-Finally, create `enriched_orders` table that is used to load data to the Elasticsearch.
+最后，创建 `enriched_orders` 表， 用来将关联后的订单数据写入 Elasticsearch 中
```sql
-- Flink SQL
Flink SQL> CREATE TABLE enriched_orders (
@@ -283,8 +279,8 @@ Flink SQL> CREATE TABLE enriched_orders (
);
```
-## Enriching orders and load to ElasticSearch
-Use Flink SQL to join the `order` table with the `products` and `shipments` table to enrich orders and write to the Elasticsearch.
+## 关联订单数据并且将其写入 Elasticsearch 中
+使用 Flink SQL 将订单表 `order` 与 商品表 `products`，物流信息表 `shipments` 关联，并将关联后的订单信息写入 Elasticsearch 中
```sql
-- Flink SQL
Flink SQL> INSERT INTO enriched_orders
@@ -293,52 +289,53 @@ Flink SQL> INSERT INTO enriched_orders
LEFT JOIN products AS p ON o.product_id = p.id
LEFT JOIN shipments AS s ON o.order_id = s.order_id;
```
-Now, the enriched orders should be shown in Kibana.
-Visit [http://localhost:5601/app/kibana#/management/kibana/index_pattern](http://localhost:5601/app/kibana#/management/kibana/index_pattern) to create an index pattern `enriched_orders`.
+现在，就可以在 Kibana 中看到包含商品和物流信息的订单数据。
+
+首先访问 [http://localhost:5601/app/kibana#/management/kibana/index_pattern](http://localhost:5601/app/kibana#/management/kibana/index_pattern) 创建 index pattern `enriched_orders`.
{{< img src="/fig/mysql-postgres-tutorial/kibana-create-index-pattern.png" width="700px" alt="Create Index Pattern" >}}
-Visit [http://localhost:5601/app/kibana#/discover](http://localhost:5601/app/kibana#/discover) to find the enriched orders.
+然后就可以在 [http://localhost:5601/app/kibana#/discover](http://localhost:5601/app/kibana#/discover) 看到写入的数据了.
{{< img src="/fig/mysql-postgres-tutorial/kibana-detailed-orders.png" width="700px" alt="Find enriched Orders" >}}
-Next, do some change in the databases, and then the enriched orders shown in Kibana will be updated after each step in real time.
-1. Insert a new order in MySQL
+接下来，修改 MySQL 和 Postgres 数据库中表的数据，Kibana中显示的订单数据也将实时更新：
+1. 在 MySQL 的 `orders` 表中插入一条数据
```sql
--MySQL
INSERT INTO orders
VALUES (default, '2020-07-30 15:22:00', 'Jark', 29.71, 104, false);
```
-2. Insert a shipment in Postgres
+2. 在 Postgres 的 `shipment` 表中插入一条数据
```sql
--PG
INSERT INTO shipments
VALUES (default,10004,'Shanghai','Beijing',false);
```
-3. Update the order status in MySQL
+3. 在 MySQL 的 `orders` 表中更新订单的状态
```sql
--MySQL
UPDATE orders SET order_status = true WHERE order_id = 10004;
```
-4. Update the shipment status in Postgres
+4. 在 Postgres 的 `shipment` 表中更新物流的状态
```sql
--PG
UPDATE shipments SET is_arrived = true WHERE shipment_id = 1004;
```
-5. Delete the order in MySQL
+5. 在 MYSQL 的 `orders` 表中删除一条数据
```sql
--MySQL
DELETE FROM orders WHERE order_id = 10004;
```
- The changes of enriched orders in Kibana are as follows:
+ 每执行一步就刷新一次 Kibana，可以看到 Kibana 中显示的订单数据将实时更新，如下所示：
{{< img src="/fig/mysql-postgres-tutorial/kibana-detailed-orders-changes.gif" width="700px" alt="Enriched Orders Changes" >}}
-
-## Clean up
-After finishing the tutorial, run the following command to stop all containers in the directory of `docker-compose.yml`:
+
+## 环境清理
+本教程结束后，在 `docker-compose.yml` 文件所在的目录下执行如下命令停止所有容器：
```shell
docker-compose down
```
-Run the following command to stop the Flink cluster in the directory of Flink `flink-1.18.0`:
+在 Flink 所在目录 `flink-1.17.0` 下执行如下命令停止 Flink 集群：
```shell
./bin/stop-cluster.sh
```
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md
index 768355154..722da0ae0 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md
@@ -1,6 +1,6 @@
---
title: "Db2 教程"
-weight: 8
+weight: 2
type: docs
aliases:
- /connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.html
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/mongodb-tutorial.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/mongodb-tutorial.md
index 6c9cb46cb..b8a33e217 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/mongodb-tutorial.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/mongodb-tutorial.md
@@ -24,9 +24,9 @@ specific language governing permissions and limitations
under the License.
-->
-# Demo: MongoDB CDC to Elasticsearch
+# 演示: MongoDB CDC 导入 Elasticsearch
-1. Create `docker-compose.yml` file using following contents:
+1. 下载 `docker-compose.yml`
```
version: '2.1'
@@ -62,20 +62,20 @@ services:
- "5601:5601"
```
-2. Enter Mongodb's container and initialize replica set and data:
+2. 进入 MongoDB 容器，初始化副本集和数据:
```
docker-compose exec mongo /usr/bin/mongo -u mongouser -p mongopw
```
```javascript
-// 1. initialize replica set
+// 1. 初始化副本集
rs.initiate();
rs.status();
-// 2. switch database
+// 2. 切换数据库
use mgdb;
-// 3. initialize data
+// 3. 初始化数据
db.orders.insertMany([
{
order_id: 101,
@@ -131,21 +131,21 @@ db.customers.insertMany([
]);
```
-3. Download following JAR package to `<FLINK_HOME>/lib/`:
+3. 下载以下 jar 包到 `<FLINK_HOME>/lib/`:
-```Download links are available only for stable releases, SNAPSHOT dependencies need to be built based on master or release branches by yourself. ```
+```下载链接只对已发布的版本有效, SNAPSHOT 版本需要本地编译```
- [flink-sql-connector-elasticsearch7-3.0.1-1.17.jar](https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-elasticsearch7/3.0.1-1.17/flink-sql-connector-elasticsearch7-3.0.1-1.17.jar)
-- [flink-sql-connector-mongodb-cdc-3.0-SNAPSHOT.jar](https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-mongodb-cdc/3.0-SNAPSHOT/flink-sql-connector-mongodb-cdc-3.0-SNAPSHOT.jar)
+- [flink-sql-connector-mongodb-cdc-2.4.0.jar](https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mongodb-cdc/2.4.0/flink-sql-connector-mongodb-cdc-2.4.0.jar)
-4. Launch a Flink cluster, then start a Flink SQL CLI and execute following SQL statements inside:
+4. 然后启动 Flink 集群，再启动 SQL CLI.
```sql
-- Flink SQL
--- checkpoint every 3000 milliseconds
+-- 设置间隔时间为3秒
Flink SQL> SET execution.checkpointing.interval = 3s;
--- set local time zone as Asia/Shanghai
+-- 设置本地时区为 Asia/Shanghai
Flink SQL> SET table.local-time-zone = Asia/Shanghai;
Flink SQL> CREATE TABLE orders (
@@ -210,34 +210,34 @@ Flink SQL> INSERT INTO enriched_orders
LEFT JOIN customers AS c ON o.customer_id = c.customer_id;
```
-5. Make some changes in MongoDB, then check the result in Elasticsearch:
+5. 修改 MongoDB 里面的数据，观察 elasticsearch 里的结果。
```javascript
-db.orders.insert({
- order_id: 104,
- order_date: ISODate("2020-07-30T12:00:30.001Z"),
- customer_id: 1004,
- price: NumberDecimal("25.25"),
- product: {
- name: 'rocks',
- description: 'box of assorted rocks'
- },
- order_status: false
+db.orders.insert({
+ order_id: 104,
+ order_date: ISODate("2020-07-30T12:00:30.001Z"),
+ customer_id: 1004,
+ price: NumberDecimal("25.25"),
+ product: {
+ name: 'rocks',
+ description: 'box of assorted rocks'
+ },
+ order_status: false
});
-db.customers.insert({
- customer_id: 1004,
- name: 'Jacob',
- address: 'Shanghai'
+db.customers.insert({
+ customer_id: 1004,
+ name: 'Jacob',
+ address: 'Shanghai'
});
db.orders.updateOne(
- { order_id: 104 },
- { $set: { order_status: true } }
+ { order_id: 104 },
+ { $set: { order_status: true } }
);
db.orders.deleteOne(
- { order_id : 104 }
+ { order_id : 104 }
);
```
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/oceanbase-tutorial.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/oceanbase-tutorial.md
index 1211470c6..d0a06fd62 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/oceanbase-tutorial.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/oceanbase-tutorial.md
@@ -24,33 +24,29 @@ specific language governing permissions and limitations
under the License.
-->
-# Demo: OceanBase CDC to ElasticSearch
+# 演示: OceanBase CDC 导入 Elasticsearch
-## Video tutorial
+## 视频教程
- [YouTube](https://www.youtube.com/watch?v=ODGE-73Dntg&t=2s)
- [Bilibili](https://www.bilibili.com/video/BV1Zg411a7ZB/?spm_id_from=333.999.0.0)
-### Preparation
-#### Configure and start the components
+### 准备教程所需要的组件
-Create `docker-compose.yml`.
+#### 配置并启动容器
-*Note*: `host` network mode is required in this demo, so it can only work on Linux, see [network-tutorial-host](https://docs.docker.com/network/network-tutorial-host/).
+配置 `docker-compose.yml`。
```yaml
version: '2.1'
services:
observer:
- image: oceanbase/oceanbase-ce:4.2.0.0
+ image: oceanbase/oceanbase-ce:4.0.0.0
container_name: observer
- environment:
- - 'MODE=slim'
- - 'OB_ROOT_PASSWORD=pswd'
network_mode: "host"
oblogproxy:
- image: whhe/oblogproxy:1.1.3_4x
+ image: whhe/oblogproxy:1.1.0_4x
container_name: oblogproxy
environment:
- 'OB_SYS_USERNAME=root'
@@ -83,40 +79,50 @@ services:
- '/var/run/docker.sock:/var/run/docker.sock'
```
-Execute the following command in the directory where `docker-compose.yml` is located.
+在 `docker-compose.yml` 所在目录下执行下面的命令来启动本教程需要的组件：
```shell
docker-compose up -d
```
-### Set password
+### 设置密码
+
+OceanBase 中 root 用户默认是没有密码的，但是 oblogproxy 需要配置一个使用非空密码的系统租户用户，因此这里我们需要先为 root@sys 用户设置一个密码。
-From OceanBase 4.0.0.0 CE, we can only fetch the commit log of non-sys tenant.
+登陆 sys 租户的 root 用户：
-Here we use the 'test' tenant for example.
+```shell
+docker-compose exec observer obclient -h127.0.0.1 -P2881 -uroot@sys
+```
-Login with 'root' user of 'test' tenant:
+设置密码，注意这里的密码需要与上一步中 oblogproxy 服务的环境变量 'OB_SYS_PASSWORD' 保持一样。
+
+```mysql
+ALTER USER root IDENTIFIED BY 'pswd';
+```
+
+OceanBase 从社区版 4.0.0.0 开始只支持对非 sys 租户的增量数据拉取，这里我们使用 test 租户的 root 用户作为示例。
+
+登陆 test 租户的 root 用户：
```shell
docker-compose exec observer obclient -h127.0.0.1 -P2881 -uroot@test
```
-Set a password:
+设置密码:
```mysql
ALTER USER root IDENTIFIED BY 'test';
```
-### Create data for reading snapshot
+### 准备数据
-Login 'root' user of 'test' tenant.
+使用 'root@test' 用户登陆。
```shell
docker-compose exec observer obclient -h127.0.0.1 -P2881 -uroot@test -ptest
```
-Insert data:
-
```sql
CREATE DATABASE ob;
USE ob;
@@ -154,23 +160,23 @@ VALUES (default, '2020-07-30 10:08:22', 'Jark', 50.50, 102, false),
(default, '2020-07-30 12:00:30', 'Edward', 25.25, 106, false);
```
-### Download the libraries required
+### 下载所需要的依赖包
-```Download links are only available for stable releases.```
+```下载链接只对已发布的版本有效, SNAPSHOT 版本需要本地编译```
- [flink-sql-connector-elasticsearch7-3.0.1-1.17.jar](https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-elasticsearch7/3.0.1-1.17/flink-sql-connector-elasticsearch7-3.0.1-1.17.jar)
-- [flink-sql-connector-oceanbase-cdc-3.0-SNAPSHOT.jar](https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-oceanbase-cdc/3.0-SNAPSHOT/flink-sql-connector-oceanbase-cdc-3.0-SNAPSHOT.jar)
+- [flink-sql-connector-oceanbase-cdc-2.4.0.jar](https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-oceanbase-cdc/2.4.0/flink-sql-connector-oceanbase-cdc-2.4.0.jar)
-### Use Flink DDL to create dynamic table in Flink SQL CLI
+### 在 Flink SQL CLI 中使用 Flink DDL 创建表
```sql
--- checkpoint every 3000 milliseconds
+-- 设置间隔时间为3秒
Flink SQL> SET execution.checkpointing.interval = 3s;
--- set local time zone as Asia/Shanghai
+-- 设置本地时区为 Asia/Shanghai
Flink SQL> SET table.local-time-zone = Asia/Shanghai;
--- create orders table
+-- 创建订单表
Flink SQL> CREATE TABLE orders (
order_id INT,
order_date TIMESTAMP(0),
@@ -195,7 +201,7 @@ Flink SQL> CREATE TABLE orders (
'working-mode' = 'memory'
);
--- create products table
+-- 创建商品表
Flink SQL> CREATE TABLE products (
id INT,
name STRING,
@@ -215,9 +221,9 @@ Flink SQL> CREATE TABLE products (
'logproxy.host' = 'localhost',
'logproxy.port' = '2983',
'working-mode' = 'memory'
- );
+ );
--- create flat table enriched_orders
+-- 创建关联后的订单数据表
Flink SQL> CREATE TABLE enriched_orders (
order_id INT,
order_date TIMESTAMP(0),
@@ -233,7 +239,7 @@ Flink SQL> CREATE TABLE enriched_orders (
'hosts' = 'http://localhost:9200',
'index' = 'enriched_orders');
--- Start the reading and writing job
+-- 执行读取和写入
Flink SQL> INSERT INTO enriched_orders
SELECT o.order_id,
o.order_date,
@@ -247,13 +253,13 @@ Flink SQL> INSERT INTO enriched_orders
LEFT JOIN products AS p ON o.product_id = p.id;
```
-### Check data on Kibana
+### 在 Kibana 中查看数据
-Open [http://localhost:5601/app/kibana#/management/kibana/index_pattern](http://localhost:5601/app/kibana#/management/kibana/index_pattern) and create index pattern `enriched_orders`, then go to [http://localhost:5601/app/kibana#/discover](http://localhost:5601/app/kibana#/discover), and you will see the data of `enriched_orders`.
+访问 [http://localhost:5601/app/kibana#/management/kibana/index_pattern](http://localhost:5601/app/kibana#/management/kibana/index_pattern) 创建 index pattern `enriched_orders`，之后可以在 [http://localhost:5601/app/kibana#/discover](http://localhost:5601/app/kibana#/discover) 看到写入的数据了。
-### Check data changes
+### 修改监听表数据，查看增量数据变动
-Execute the following sql in OceanBase under `ob` database, you will find records in Kibana be updated after each step in real time.
+在OceanBase中依次执行如下修改操作，每执行一步就刷新一次 Kibana，可以看到 Kibana 中显示的订单数据将实时更新。
```sql
INSERT INTO orders VALUES (default, '2020-07-30 15:22:00', 'Jark', 29.71, 104, false);
@@ -261,15 +267,15 @@ UPDATE orders SET order_status = true WHERE order_id = 10004;
DELETE FROM orders WHERE order_id = 10004;
```
-### Clean up
+### 环境清理
-Execute the following command to stop all containers in the directory where `docker-compose.yml` is located.
+在 `docker-compose.yml` 文件所在的目录下执行如下命令停止所有容器：
```shell
docker-compose down
```
-Stop the flink cluster by following command.
+进入Flink的部署目录，停止 Flink 集群：
```shell
./bin/stop-cluster.sh
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/oracle-tutorial.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/oracle-tutorial.md
index ab5fc3cef..649f34976 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/oracle-tutorial.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/oracle-tutorial.md
@@ -24,9 +24,9 @@ specific language governing permissions and limitations
under the License.
-->
-# Demo: Oracle CDC to Elasticsearch
+# 演示: Oracle CDC 导入 Elasticsearch
-**Create `docker-compose.yml` file using following contents:**
+**创建`docker-compose.yml`文件，内容如下所示:**
```
version: '2.1'
@@ -58,35 +58,36 @@ services:
- "5601:5601"
volumes:
- /var/run/docker.sock:/var/run/docker.sock
-```
-The Docker Compose environment consists of the following containers:
-- Oracle: Oracle 19c database.
-- Elasticsearch: store the join result of the `orders` and `products` table.
-- Kibana: mainly used to visualize the data in Elasticsearch
+```
+该 Docker Compose 中包含的容器有:
+- Oracle: Oracle 19c 数据库
+- Elasticsearch: `orders` 表将和 `products` 表进行join，join的结果写入Elasticsearch中
+- Kibana: 可视化 Elasticsearch 中的数据
-To start all containers, run the following command in the directory that contains the docker-compose.yml file.
+在 docker-compose.yml 所在目录下运行如下命令以启动所有容器：
```shell
docker-compose up -d
```
-This command automatically starts all the containers defined in the Docker Compose configuration in a detached mode.
-Run docker ps to check whether these containers are running properly. You can also visit http://localhost:5601/ to see if Kibana is running normally.
+该命令会以 detached 模式自动启动 Docker Compose 配置中定义的所有容器。
+你可以通过 docker ps 来观察上述的容器是否正常启动了。 也可以访问 http://localhost:5601/ 来查看 Kibana 是否运行正常。
+另外可以通过如下命令停止所有的容器：
-Don’t forget to run the following command to stop all containers after you finished the tutorial:
```shell
docker-compose down
-```
+````
-**Download following JAR package to `<FLINK_HOME>/lib`**
+**下载以下 jar 包到 `<FLINK_HOME>/lib/`:**
-**Download links are available only for stable releases, SNAPSHOT dependencies need to be built based on master or release-branches by yourself.**
+*下载链接只对已发布的版本有效, SNAPSHOT 版本需要本地编译*
- [flink-sql-connector-elasticsearch7-3.0.1-1.17.jar](https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-elasticsearch7/3.0.1-1.17/flink-sql-connector-elasticsearch7-3.0.1-1.17.jar)
-- [flink-sql-connector-oracle-cdc-3.0-SNAPSHOT.jar](https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-oracle-cdc/3.0-SNAPSHOT/flink-sql-connector-oracle-cdc-3.0-SNAPSHOT.jar)
+- [flink-sql-connector-oracle-cdc-2.4.0.jar](https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-oracle-cdc/2.4.0/flink-sql-connector-oracle-cdc-2.4.0.jar)
-**Preparing data in Oracle database**
+**在 Oracle 数据库中准备数据**
+
+创建数据库和表 `products`，`orders`，并插入数据：
-Introduce the tables in Oracle:
```shell
docker-compose exec oracle sqlplus debezium/dbz@localhost:1521/ORCLCDB
```
@@ -147,9 +148,7 @@ INSERT INTO DEBEZIUM.ORDERS VALUES (1003, TO_TIMESTAMP('2020-07-30 12:00:30.0010
INSERT INTO DEBEZIUM.ORDERS VALUES (1004, TO_TIMESTAMP('2020-07-30 15:22:00.001000', 'YYYY-MM-DD HH24:MI:SS.FF'), 'Jark', 1, 104);
```
-**Launch a Flink cluster and start a Flink SQL CLI**
-
-Execute following SQL statements in the Flink SQL CLI:
+**然后启动 Flink 集群，再启动 SQL CLI:**
```sql
-- Flink SQL
@@ -210,13 +209,13 @@ Flink SQL> INSERT INTO enriched_orders
LEFT JOIN products AS p ON o.PRODUCT_ID = p.ID;
```
-**Check result in Elasticsearch**
+**检查 ElasticSearch 中的结果**
-Check the data has been written to Elasticsearch successfully, you can visit [Kibana](http://localhost:5601/) to see the data.
+检查最终的结果是否写入ElasticSearch中, 可以在[Kibana](http://localhost:5601/)看到ElasticSearch中的数据
-**Make changes in Oracle and watch result in Elasticsearch**
+**在 Oracle 制造一些变更，观察 ElasticSearch 中的结果**
-Enter Oracle's container to make some changes in Oracle, then you can see the result in Elasticsearch will change after executing every SQL statement:
+进入Oracle容器中并通过如下的SQL语句对Oracle数据库进行一些修改, 然后就可以看到每执行一条SQL语句，Elasticsearch中的数据都会实时更新。
```shell
docker-compose exec oracle sqlplus debezium/dbz@localhost:1521/ORCLCDB
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/polardbx-tutorial.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/polardbx-tutorial.md
index 71c66d924..fce4be25d 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/polardbx-tutorial.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/polardbx-tutorial.md
@@ -24,24 +24,18 @@ specific language governing permissions and limitations
under the License.
-->
-# Demo: PolarDB-X CDC to Elasticsearch
+# 演示: PolarDB-X CDC 导入 Elasticsearch
-This tutorial is to show how to quickly build streaming ETL for PolarDB-X with Flink CDC.
+本示例我们通过演示 PolarDB-X 借助 Flink-CDC 将数据导入至 Elasticsearch 来介绍 PolarDB-X 的增量订阅能力，你可以前往:[PolarDB-X](https://github.com/ApsaraDB/galaxysql) 了解更多细节。
-Assuming we are running an e-commerce business. The product and order data stored in PolarDB-X.
-We want to enrich the orders using the product table, and then load the enriched orders to ElasticSearch in real time.
+### 准备教程所需要的组件
+我们假设你运行在一台 MacOS 或者 Linux 机器上，并且已经安装 docker.
-In the following sections, we will describe how to use Flink PolarDB-X CDC to implement it.
-All exercises in this tutorial are performed in the Flink SQL CLI, and the entire process uses standard SQL syntax, without a single line of Java/Scala code or IDE installation.
+#### 配置并启动容器
-## Preparation
-Prepare a Linux or MacOS computer with Docker installed.
+配置 `docker-compose.yml`。
-### Starting components required
-The components required in this demo are all managed in containers, so we will use `docker-compose` to start them.
-
-Create `docker-compose.yml` file using following contents:
-```
+```yaml
version: '2.1'
services:
polardbx:
@@ -76,105 +70,96 @@ services:
volumes:
- '/var/run/docker.sock:/var/run/docker.sock'
```
-The Docker Compose environment consists of the following containers:
-- PolarDB-X: the `products`,`orders` tables will be store in the database. They will be joined enrich the orders.
-- Elasticsearch: mainly used as a data sink to store enriched orders.
-- Kibana: used to visualize the data in Elasticsearch.
+该 Docker Compose 中包含的容器有：
+- PolarDB-X: 商品表 `products` 和 订单表 `orders` 将存储在该数据库中， 这两张表将进行关联，得到一张包含更多信息的订单表 `enriched_orders`
+- Elasticsearch: 最终的订单表 `enriched_orders` 将写到 Elasticsearch
+- Kibana: 用来可视化 ElasticSearch 的数据
+
+在 `docker-compose.yml` 所在目录下执行下面的命令来启动本教程需要的组件：
-To start all containers, run the following command in the directory that contains the `docker-compose.yml` file.
```shell
docker-compose up -d
```
-This command automatically starts all the containers defined in the Docker Compose configuration in a detached mode. Run docker ps to check whether these containers are running properly.
-We can also visit [http://localhost:5601/](http://localhost:5601/) to see if Kibana is running normally.
-
-### Preparing Flink and JAR package required
-1. Download [Flink 1.18.0](https://archive.apache.org/dist/flink/flink-1.18.0/flink-1.18.0-bin-scala_2.12.tgz) and unzip it to the directory `flink-1.18.0`
-2. Download following JAR package required and put them under `flink-1.18.0/lib/`:
-
- **Download links are available only for stable releases, SNAPSHOT dependencies need to be built based on master or release branches by yourself.**
- - flink-sql-connector-mysql-cdc-3.0-SNAPSHOT.jar
- - [flink-sql-connector-elasticsearch7-3.0.1-1.17.jar](https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-elasticsearch7/3.0.1-1.17/flink-sql-connector-elasticsearch7-3.0.1-1.17.jar)
-
-### Preparing data in databases
-#### Preparing data in PolarDB-X
-1. Enter PolarDB-X Database:
- ```shell
- mysql -h127.0.0.1 -P8527 -upolardbx_root -p"123456"
- ```
-2. Create tables and populate data:
- ```sql
- -- PolarDB-X
- CREATE TABLE products (
- id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
- name VARCHAR(255) NOT NULL,
- description VARCHAR(512)
- ) AUTO_INCREMENT = 101;
-
- INSERT INTO products
- VALUES (default,"scooter","Small 2-wheel scooter"),
- (default,"car battery","12V car battery"),
- (default,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3"),
- (default,"hammer","12oz carpenter's hammer"),
- (default,"hammer","14oz carpenter's hammer"),
- (default,"hammer","16oz carpenter's hammer"),
- (default,"rocks","box of assorted rocks"),
- (default,"jacket","water resistent black wind breaker"),
- (default,"spare tire","24 inch spare tire");
-
-
- CREATE TABLE orders (
- order_id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
- order_date DATETIME NOT NULL,
- customer_name VARCHAR(255) NOT NULL,
- price DECIMAL(10, 5) NOT NULL,
- product_id INTEGER NOT NULL,
- order_status BOOLEAN NOT NULL -- Whether order has been placed
- ) AUTO_INCREMENT = 10001;
-
- INSERT INTO orders
- VALUES (default, '2020-07-30 10:08:22', 'Jark', 50.50, 102, false),
- (default, '2020-07-30 10:11:09', 'Sally', 15.00, 105, false),
- (default, '2020-07-30 12:00:30', 'Edward', 25.25, 106, false);
- ```
-
-## Starting Flink cluster and Flink SQL CLI
-
-1. Use the following command to change to the Flink directory:
- ```
- cd flink-1.18.0
- ```
-
-2. Use the following command to start a Flink cluster:
- ```shell
- ./bin/start-cluster.sh
- ```
-
- Then we can visit [http://localhost:8081/](http://localhost:8081/) to see if Flink is running normally, and the web page looks like:
-
- {{< img src="/fig/mysql-postgres-tutorial/flink-ui.png" alt="Flink UI" >}}
-
-3. Use the following command to start a Flink SQL CLI:
- ```shell
- ./bin/sql-client.sh
- ```
- We should see the welcome screen of the CLI client.
-
- {{< img src="/fig/mysql-postgres-tutorial/flink-sql-client.png" alt="Flink SQL Client" >}}
-
-## Creating tables using Flink DDL in Flink SQL CLI
-First, enable checkpoints every 3 seconds
+该命令将以 detached 模式自动启动 Docker Compose 配置中定义的所有容器。你可以通过 docker ps 来观察上述的容器是否正常启动了，也可以通过访问 [http://localhost:5601/](http://localhost:5601/) 来查看 Kibana 是否运行正常
+
+
+### 准备数据：
+
+使用已创建的用户名和密码进行登陆PolarDB-X。
+
+```shell
+mysql -h127.0.0.1 -P8527 -upolardbx_root -p"123456"
+```
+
```sql
--- Flink SQL
-Flink SQL> SET execution.checkpointing.interval = 3s;
+CREATE DATABASE mydb;
+USE mydb;
+
+-- 创建一张产品表，并写入一些数据
+CREATE TABLE products (
+ id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
+ name VARCHAR(255) NOT NULL,
+ description VARCHAR(512)
+) AUTO_INCREMENT = 101;
+
+INSERT INTO products
+VALUES (default,"scooter","Small 2-wheel scooter"),
+ (default,"car battery","12V car battery"),
+ (default,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3"),
+ (default,"hammer","12oz carpenter's hammer"),
+ (default,"hammer","14oz carpenter's hammer"),
+ (default,"hammer","16oz carpenter's hammer"),
+ (default,"rocks","box of assorted rocks"),
+ (default,"jacket","water resistent black wind breaker"),
+ (default,"spare tire","24 inch spare tire");
+
+
+-- 创建一张订单表，并写入一些数据
+CREATE TABLE orders (
+ order_id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
+ order_date DATETIME NOT NULL,
+ customer_name VARCHAR(255) NOT NULL,
+ price DECIMAL(10, 5) NOT NULL,
+ product_id INTEGER NOT NULL,
+ order_status BOOLEAN NOT NULL -- Whether order has been placed
+) AUTO_INCREMENT = 10001;
+
+INSERT INTO orders
+VALUES (default, '2020-07-30 10:08:22', 'Jark', 50.50, 102, false),
+ (default, '2020-07-30 10:11:09', 'Sally', 15.00, 105, false),
+ (default, '2020-07-30 12:00:30', 'Edward', 25.25, 106, false);
+```
+
+### 下载 Flink 和所需要的依赖包
+1. 下载 [Flink 1.17.0](https://archive.apache.org/dist/flink/flink-1.17.0/flink-1.17.0-bin-scala_2.12.tgz) 并将其解压至目录 `flink-1.17.0`
+2. 下载下面列出的依赖包，并将它们放到目录 `flink-1.17.0/lib/` 下
+
+```下载链接只对已发布的版本有效, SNAPSHOT 版本需要本地编译```
+- 用于订阅PolarDB-X Binlog: [flink-sql-connector-mysql-cdc-2.4.0.jar](https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.4.0/flink-sql-connector-mysql-cdc-2.4.0.jar)
+- 用于写入Elasticsearch: [flink-sql-connector-elasticsearch7-3.0.1-1.17.jar](https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-elasticsearch7/3.0.1-1.17/flink-sql-connector-elasticsearch7-3.0.1-1.17.jar)
+3. 启动flink服务:
+```shell
+./bin/start-cluster.sh
+```
+
+我们可以访问 [http://localhost:8081/](http://localhost:8081/) 看到Flink正常运行:
+
+{{< img src="/fig/mysql-postgres-tutorial/flink-ui.png" alt="Flink UI" >}}
+
+
+4. 启动Flink SQL CLI:
+```shell
+./bin/sql-client.sh
```
-Then, create tables that capture the change data from the corresponding database tables.
+### 在 Flink SQL CLI 中使用 Flink DDL 创建表
+
+
```sql
--- Flink SQL
+-- 设置间隔时间为3秒
Flink SQL> SET execution.checkpointing.interval = 3s;
--- create source table2 - orders
+-- 创建source1 -订单表
Flink SQL> CREATE TABLE orders (
order_id INT,
order_date TIMESTAMP(0),
@@ -193,7 +178,7 @@ Flink SQL> CREATE TABLE orders (
'table-name' = 'orders'
);
--- create source table2 - products
+-- 创建source2 -产品表
CREATE TABLE products (
id INT,
name STRING,
@@ -208,12 +193,8 @@ CREATE TABLE products (
'database-name' = 'mydb',
'table-name' = 'products'
);
-```
-Finally, create `enriched_orders` table that is used to load data to the Elasticsearch.
-```sql
--- Flink SQL
--- create sink table - enrich_orders
+-- 创建sink - 关联后的结果表
Flink SQL> CREATE TABLE enriched_orders (
order_id INT,
order_date TIMESTAMP(0),
@@ -229,12 +210,8 @@ Flink SQL> CREATE TABLE enriched_orders (
'hosts' = 'http://localhost:9200',
'index' = 'enriched_orders'
);
-```
-## Enriching orders and load to ElasticSearch
-Use Flink SQL to join the `order` table with the `products` table to enrich orders and write to the Elasticsearch.
-```sql
--- Flink SQL
+-- 执行读取和写入
Flink SQL> INSERT INTO enriched_orders
SELECT o.order_id,
o.order_date,
@@ -247,41 +224,35 @@ Flink SQL> INSERT INTO enriched_orders
FROM orders AS o
LEFT JOIN products AS p ON o.product_id = p.id;
```
-Now, the enriched orders should be shown in Kibana.
-Visit [http://localhost:5601/app/kibana#/management/kibana/index_pattern](http://localhost:5601/app/kibana#/management/kibana/index_pattern) to create an index pattern `enriched_orders`.
-
-{{< img src="/fig/mysql-postgres-tutorial/kibana-create-index-pattern.png" alt="Create Index Pattern" >}}
-
-Visit [http://localhost:5601/app/kibana#/discover](http://localhost:5601/app/kibana#/discover) to find the enriched orders.
-
-{{< img src="/fig/mysql-postgres-tutorial/kibana-detailed-orders.png" alt="Find enriched Orders" >}}
-
-Next, do some change in the databases, and then the enriched orders shown in Kibana will be updated after each step in real time.
-1. Insert a new order in PolarDB-X
- ```sql
- --PolarDB-X
- INSERT INTO orders
- VALUES (default, '2020-07-30 15:22:00', 'Jark', 29.71, 104, false);
- ```
-2. Update the order status in PolarDB-X
- ```sql
- --PolarDB-X
- UPDATE orders SET order_status = true WHERE order_id = 10004;
- ```
-3. Delete the order in PolarDB-X
- ```sql
- --PolarDB-X
- DELETE FROM orders WHERE order_id = 10004;
- ```
- The changes of enriched orders in Kibana are as follows:
- {{< img src="/fig/mysql-postgres-tutorial/kibana-detailed-orders-changes.gif" alt="Enriched Orders Changes" >}}
-
-## Clean up
-After finishing the tutorial, run the following command to stop all containers in the directory of `docker-compose.yml`:
+
+### 在 Kibana 中查看数据
+
+访问 [http://localhost:5601/app/kibana#/management/kibana/index_pattern](http://localhost:5601/app/kibana#/management/kibana/index_pattern)
+
+创建 index pattern `enriched_orders`，之后可以在 [http://localhost:5601/app/kibana#/discover](http://localhost:5601/app/kibana#/discover) 看到写入的数据了。
+
+### 修改监听表数据，查看增量数据变动
+
+在PolarDB-X中依次执行如下修改操作，每执行一步就刷新一次 Kibana，可以看到 Kibana 中显示的订单数据将实时更新。
+
+```sql
+INSERT INTO orders VALUES (default, '2020-07-30 15:22:00', 'Jark', 29.71, 104, false);
+
+UPDATE orders SET order_status = true WHERE order_id = 10004;
+
+DELETE FROM orders WHERE order_id = 10004;
+```
+
+### 环境清理
+
+在 `docker-compose.yml` 文件所在的目录下执行如下命令停止所有容器：
+
```shell
docker-compose down
```
-Run the following command to stop the Flink cluster in the directory of Flink `flink-1.18.0`:
+
+进入Flink的部署目录，停止 Flink 集群：
+
```shell
./bin/stop-cluster.sh
```
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/sqlserver-tutorial.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/sqlserver-tutorial.md
index 31ca52a08..89f22a31f 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/sqlserver-tutorial.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/sqlserver-tutorial.md
@@ -24,9 +24,9 @@ specific language governing permissions and limitations
under the License.
-->
-# Demo: SqlServer CDC to Elasticsearch
+# 演示: SqlServer CDC 导入 Elasticsearch
-**Create `docker-compose.yml` file using following contents:**
+**创建 `docker-compose.yml` 文件，内容如下所示：**
```
version: '2.1'
@@ -67,35 +67,35 @@ services:
volumes:
- /var/run/docker.sock:/var/run/docker.sock
```
-The Docker Compose environment consists of the following containers:
-- SqlServer: SqlServer database.
-- Elasticsearch: store the join result of the `orders` and `products` table.
-- Kibana: mainly used to visualize the data in Elasticsearch.
+该 Docker Compose 中包含的容器有：
+- SqlServer：SqlServer 数据库。
+- Elasticsearch：`orders` 表将和 `products` 表进行 join，join 的结果写入 Elasticsearch 中。
+- Kibana：可视化 Elasticsearch 中的数据。
-To start all containers, run the following command in the directory that contains the docker-compose.yml file:
+在 docker-compose.yml 所在目录下运行如下命令以启动所有容器：
```shell
docker-compose up -d
```
-This command automatically starts all the containers defined in the Docker Compose configuration in a detached mode.
-Run docker ps to check whether these containers are running properly. You can also visit http://localhost:5601/ to see if Kibana is running normally.
+该命令会以 detached 模式自动启动 Docker Compose 配置中定义的所有容器。
+你可以通过 docker ps 来观察上述的容器是否正常启动了。 也可以访问 http://localhost:5601/ 来查看 Kibana 是否运行正常。
-Don’t forget to run the following command to stop and remove all containers after you finished the tutorial:
+另外可以通过如下命令停止并删除所有的容器：
```shell
docker-compose down
````
-**Download following JAR package to `<FLINK_HOME>/lib`:**
+**下载以下 jar 包到 `<FLINK_HOME>/lib/`：**
-**Download links are available only for stable releases, SNAPSHOT dependencies need to be built based on master or release branches by yourself.**
+```下载链接只对已发布的版本有效, SNAPSHOT 版本需要本地编译```
- [flink-sql-connector-elasticsearch7-3.0.1-1.17.jar](https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-elasticsearch7/3.0.1-1.17/flink-sql-connector-elasticsearch7-3.0.1-1.17.jar)
-- [flink-sql-connector-sqlserver-cdc-3.0-SNAPSHOT.jar](https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-sqlserver-cdc/3.0-SNAPSHOT/flink-sql-connector-sqlserver-cdc-3.0-SNAPSHOT.jar)
+- [flink-sql-connector-sqlserver-cdc-2.4.0.jar](https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-sqlserver-cdc/2.4.0/flink-sql-connector-sqlserver-cdc-2.4.0.jar)
-**Preparing data in SqlServer database**
+**在 SqlServer 数据库中准备数据**
-Create databases/tables and populate data
+创建数据库和表 `products`，`orders`，并插入数据：
```sql
-- Sqlserver
@@ -150,7 +150,7 @@ Create databases/tables and populate data
EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', @source_name = 'orders', @role_name = NULL, @supports_net_changes = 0;
GO
```
-**Launch a Flink cluster and start a Flink SQL CLI:**
+**然后启动 Flink 集群，再启动 SQL CLI：**
```sql
-- Flink SQL
@@ -209,13 +209,13 @@ Flink SQL> INSERT INTO enriched_orders
LEFT JOIN products AS p ON o.product_id = p.id;
```
-**Check result in Elasticsearch**
+**检查 ElasticSearch 中的结果**
-Check the data has been written to Elasticsearch successfully, you can visit [Kibana](http://localhost:5601/) to see the data.
+检查最终的结果是否写入 ElasticSearch 中，可以在 [Kibana](http://localhost:5601/) 看到 ElasticSearch 中的数据。
-**Make changes in SqlServer and watch result in Elasticsearch**
+**在 SqlServer 制造一些变更，观察 ElasticSearch 中的结果**
-Do some changes in the databases, and then the enriched orders shown in Kibana will be updated after each step in real time.
+通过如下的 SQL 语句对 SqlServer 数据库进行一些修改，然后就可以看到每执行一条 SQL 语句，Elasticsearch 中的数据都会实时更新。
```sql
INSERT INTO orders(order_date,purchaser,quantity,product_id) VALUES ('22-FEB-2016', 1006, 22, 107);
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/tidb-tutorial.md b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/tidb-tutorial.md
index fbb8700a4..b4994d2ea 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/tidb-tutorial.md
+++ b/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/tidb-tutorial.md
@@ -24,14 +24,14 @@ specific language governing permissions and limitations
under the License.
-->
-# Demo: TiDB CDC to Elasticsearch
+# 演示: TiDB CDC 导入 Elasticsearch
-**First,we will start TiDB cluster with docker.**
+**首先我们得通过 docker 来启动 TiDB 集群。**
```shell
$ git clone https://github.com/pingcap/tidb-docker-compose.git
```
-**Next,replace `docker-compose.yml` file using following contents in directory `tidb-docker-compose`:**
+**其次替换目录 `tidb-docker-compose` 里面的 `docker-compose.yml` 文件，内容如下所示：**
```
version: "2.1"
@@ -118,37 +118,37 @@ services:
- /var/run/docker.sock:/var/run/docker.sock
```
-The Docker Compose environment consists of the following containers:
-- TiDB cluster: tikv、pd、tidb.
-- Elasticsearch: store the join result of the `orders` and `products` table.
-- Kibana: mainly used to visualize the data in Elasticsearch.
+该 Docker Compose 中包含的容器有：
+- TiDB 集群: tikv、pd、tidb。
+- Elasticsearch：`orders` 表将和 `products` 表进行 join，join 的结果写入 Elasticsearch 中。
+- Kibana：可视化 Elasticsearch 中的数据。
-Add `pd` and `tikv` mapping to `127.0.0.1` in `host` file.
-To start all containers, run the following command in the directory that contains the docker-compose.yml file:
+本机添加 host 映射 `pd` 和 `tikv` 映射 `127.0.0.1`。
+在 docker-compose.yml 所在目录下运行如下命令以启动所有容器：
```shell
docker-compose up -d
mysql -h 127.0.0.1 -P 4000 -u root # Just test tidb cluster is ready,if you have install mysql local.
```
-This command automatically starts all the containers defined in the Docker Compose configuration in a detached mode.
-Run docker ps to check whether these containers are running properly. You can also visit http://localhost:5601/ to see if Kibana is running normally.
+该命令会以 detached 模式自动启动 Docker Compose 配置中定义的所有容器。
+你可以通过 docker ps 来观察上述的容器是否正常启动了。 也可以访问 http://localhost:5601/ 来查看 Kibana 是否运行正常。
-Don’t forget to run the following command to stop and remove all containers after you finished the tutorial:
+另外可以通过如下命令停止并删除所有的容器：
```shell
docker-compose down
````
-**Download following JAR package to `<FLINK_HOME>/lib`:**
+**下载以下 jar 包到 `<FLINK_HOME>/lib/`：**
-**Download links are available only for stable releases, SNAPSHOT dependencies need to be built based on master or release branches by yourself.**
+```下载链接只对已发布的版本有效, SNAPSHOT 版本需要本地编译```
- [flink-sql-connector-elasticsearch7-3.0.1-1.17.jar](https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-elasticsearch7/3.0.1-1.17/flink-sql-connector-elasticsearch7-3.0.1-1.17.jar)
-- [flink-sql-connector-tidb-cdc-3.0-SNAPSHOT.jar](https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-tidb-cdc/3.0-SNAPSHOT/flink-sql-connector-tidb-cdc-3.0-SNAPSHOT.jar)
+- [flink-sql-connector-tidb-cdc-2.4.0.jar](https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-tidb-cdc/2.4.0/flink-sql-connector-tidb-cdc-2.4.0.jar)
-**Preparing data in TiDB database**
+**在 TiDB 数据库中准备数据**
-Create databases/tables and populate data
+创建数据库和表 `products`，`orders`，并插入数据：
```sql
-- TiDB
@@ -185,7 +185,7 @@ VALUES (default, '2020-07-30 10:08:22', 'Jark', 50.50, 102, false),
(default, '2020-07-30 10:11:09', 'Sally', 15.00, 105, false),
(default, '2020-07-30 12:00:30', 'Edward', 25.25, 106, false);
```
-**Launch a Flink cluster and start a Flink SQL CLI:**
+**然后启动 Flink 集群，再启动 SQL CLI：**
```sql
-- Flink SQL
@@ -241,13 +241,13 @@ Flink SQL> INSERT INTO enriched_orders
LEFT JOIN products AS p ON o.product_id = p.id;
```
-**Check result in Elasticsearch**
+**检查 ElasticSearch 中的结果**
-Check the data has been written to Elasticsearch successfully, you can visit [Kibana](http://localhost:5601/) to see the data.
+检查最终的结果是否写入 ElasticSearch 中，可以在 [Kibana](http://localhost:5601/) 看到 ElasticSearch 中的数据。
-**Make changes in TiDB and watch result in Elasticsearch**
+**在 TiDB 制造一些变更，观察 ElasticSearch 中的结果**
-Do some changes in the databases, and then the enriched orders shown in Kibana will be updated after each step in real time.
+通过如下的 SQL 语句对 TiDB 数据库进行一些修改，然后就可以看到每执行一条 SQL 语句，Elasticsearch 中的数据都会实时更新。
```sql
INSERT INTO orders
diff --git a/docs/content/docs/connectors/legacy-flink-cdc-sources/overview.md b/docs/content/docs/connectors/legacy-flink-cdc-sources/overview.md
index 4311ed961..b718aec23 100644
--- a/docs/content/docs/connectors/legacy-flink-cdc-sources/overview.md
+++ b/docs/content/docs/connectors/legacy-flink-cdc-sources/overview.md
@@ -24,12 +24,12 @@ specific language governing permissions and limitations
under the License.
-->
-# Legacy CDC sources for Apache Flink
+# Legacy CDC sources
-Flink CDC sources is a set of source connectors for <a href="https://flink.apache.org/">Apache Flink<sup>®</sup></a>, ingesting changes from different databases using change data capture (CDC).
+Legacy CDC sources is a set of source connectors for <a href="https://flink.apache.org/">Apache Flink<sup>®</sup></a>, ingesting changes from different databases using change data capture (CDC).
Some CDC sources integrate Debezium as the engine to capture data changes. So it can fully leverage the ability of Debezium. See more about what is [Debezium](https://github.com/debezium/debezium).
-You can also read [tutorials]({{< ref "docs/connectors/legacy-flink-cdc-sources/tutorials/mysql-postgres-tutorial" >}}) about how to use these sources.
+You can also read [tutorials]({{< ref "docs/connectors/legacy-flink-cdc-sources/tutorials/build-streaming-etl-tutorial" >}}) about how to use these sources.
{{< img src="/fig/cdc-flow.png" width="600px" alt="Flink CDC" >}}
diff --git a/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/build-real-time-data-lake-tutorial.md b/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/build-real-time-data-lake-tutorial.md
index 8f43f0995..6408a9aa5 100644
--- a/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/build-real-time-data-lake-tutorial.md
+++ b/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/build-real-time-data-lake-tutorial.md
@@ -1,6 +1,6 @@
---
title: "Building a Real-time Data Lake with Flink CDC"
-weight: 999
+weight: 998
type: docs
aliases:
- /connectors/legacy-flink-cdc-sources/tutorials/build-real-time-data-lake-tutorial.html
@@ -26,7 +26,7 @@ specific language governing permissions and limitations
under the License.
-->
-# Using Flink CDC to synchronize data from MySQL sharding tables and build real-time data lake
+# Building a Real-time Data Lake with Flink CDC
For OLTP databases, to deal with a huge number of data in a single table, we usually do database and table sharding to get better throughput.
But sometimes, for convenient analysis, we need to merge them into one table when loading them to data warehouse or data lake.
diff --git a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/mysql-postgres-tutorial.md b/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/build-streaming-etl-tutorial.md
similarity index 99%
rename from docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/mysql-postgres-tutorial.md
rename to docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/build-streaming-etl-tutorial.md
index c124bead9..a357570bf 100644
--- a/docs/content.zh/docs/connectors/legacy-flink-cdc-sources/tutorials/mysql-postgres-tutorial.md
+++ b/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/build-streaming-etl-tutorial.md
@@ -1,6 +1,6 @@
---
-title: "MySQL & PostgreSQL 教程"
-weight: 2
+title: "Building a Streaming ETL with Flink CDC"
+weight: 999
type: docs
aliases:
- /connectors/legacy-flink-cdc-sources/tutorials/mysql-postgres-tutorial.html
@@ -24,7 +24,7 @@ specific language governing permissions and limitations
under the License.
-->
-# Streaming ETL for MySQL and PostgreSQL with Flink CDC
+# Building a Streaming ETL with Flink CDC
This tutorial is to show how to quickly build streaming ETL for MySQL and Postgres with Flink CDC.
diff --git a/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md b/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md
index 9232aeb0b..54e69b7f6 100644
--- a/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md
+++ b/docs/content/docs/connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.md
@@ -1,6 +1,6 @@
---
title: "Db2 Tutorial"
-weight: 8
+weight: 2
type: docs
aliases:
- /connectors/legacy-flink-cdc-sources/tutorials/db2-tutorial.html