You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Sukang1002 (via GitHub)" <gi...@apache.org> on 2025/06/02 11:09:07 UTC
[PR] [Flink-36980] [cdc-connect] Build a new introductory practice "Streaming ELT from MySQL to Kafka" [flink-cdc]
Sukang1002 opened a new pull request, #4033:
URL: https://github.com/apache/flink-cdc/pull/4033
Desc: [FLINK-36980](https://issues.apache.org/jira/browse/FLINK-36980)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-36980] [cdc-connect] Build a new introductory practice "Streaming ELT from MySQL to Kafka" [flink-cdc]
Posted by "MOBIN-F (via GitHub)" <gi...@apache.org>.
MOBIN-F commented on PR #4033:
URL: https://github.com/apache/flink-cdc/pull/4033#issuecomment-2934465483
Hi @Sukang1002, Thanks for your contribution, I left some comments~
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-36980] [cdc-connect] Build a new introductory practice "Streaming ELT from MySQL to Kafka" [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on code in PR #4033:
URL: https://github.com/apache/flink-cdc/pull/4033#discussion_r2265446885
##########
docs/content/docs/get-started/quickstart/mysql-to-kafka.md:
##########
@@ -0,0 +1,591 @@
+---
+title: "MySQL to Kafka"
+weight: 2
+type: docs
+aliases:
+- /try-flink-cdc/pipeline-connectors/mysql-Kafka-pipeline-tutorial.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Streaming ELT from MySQL to Kafka
+
+This tutorial is to show how to quickly build a Streaming ELT job from MySQL to StarRocks using Flink CDC, including the
Review Comment:
```suggestion
This tutorial is to show how to quickly build a Streaming ELT job from MySQL to Kafka using Flink CDC, including the
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-36980] [cdc-connect] Build a new introductory practice "Streaming ELT from MySQL to Kafka" [flink-cdc]
Posted by "Sukang1002 (via GitHub)" <gi...@apache.org>.
Sukang1002 commented on PR #4033:
URL: https://github.com/apache/flink-cdc/pull/4033#issuecomment-3289441118
Thank you both for your guidance. Please take the time to review the article again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-36980] [cdc-connect] Build a new introductory practice "Streaming ELT from MySQL to Kafka" [flink-cdc]
Posted by "MOBIN-F (via GitHub)" <gi...@apache.org>.
MOBIN-F commented on code in PR #4033:
URL: https://github.com/apache/flink-cdc/pull/4033#discussion_r2123339621
##########
docs/content.zh/docs/get-started/quickstart/mysql-to-kafka.md:
##########
@@ -0,0 +1,588 @@
+---
+title: "MySQL 同步到 Kafka"
+weight: 2
+type: docs
+aliases:
+- /try-flink-cdc/pipeline-connectors/mysql-Kafka-pipeline-tutorial.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Streaming ELT 同步 MySQL 到 Kafka
+
+这篇教程将展示如何基于 Flink CDC 快速构建 MySQL 到 Kafka 的 Streaming ELT 作业，包含整库同步、表结构变更同步和分库分表同步的功能。
+本教程的演示都将在 Flink CDC CLI 中进行，无需一行 Java/Scala 代码，也无需安装 IDE。
+
+## 准备阶段
+准备一台已经安装了 Docker 的 Linux 或者 MacOS 电脑。
+
+### 准备 Flink Standalone 集群
+1. 下载 [Flink 1.20.1](https://archive.apache.org/dist/flink/flink-1.20.1/flink-1.20.1-bin-scala_2.12.tgz) ，解压后得到 flink-1.20.1 目录。
+ 使用下面的命令跳转至 Flink 目录下，并且设置 FLINK_HOME 为 flink-1.20.1 所在目录。
+
+ ```shell
+ tar -zxvf flink-1.20.1-bin-scala_2.12.tgz
+ exprot FLINK_HOME=$(pwd)/flink-1.20.1
+ cd flink-1.20.1
+ ```
+
+2. 通过在 conf/flink-conf.yaml 配置文件追加下列参数开启 checkpoint，每隔 3 秒做一次 checkpoint。
+
+ ```yaml
+ execution:
+ checkpointing:
+ interval: 3000
+ ```
+
+3. 使用下面的命令启动 Flink 集群。
+
+ ```shell
+ ./bin/start-cluster.sh
+ ```
+
+启动成功的话，可以在 [http://localhost:8081/](http://localhost:8081/) 访问到 Flink Web UI，如下所示：
+
+{{< img src="/fig/mysql-Kafka-tutorial/flink-ui.png" alt="Flink UI" >}}
+
+多次执行 start-cluster.sh 可以拉起多个 TaskManager。
+注：如果你是云服务器，无法访问本地，需要将 conf/config.yaml 里面 rest.bind-address 和 rest.address的 localhost 改成0.0.0.0，然后使用 公网IP:8081 即可访问。
+
+### 准备 Docker 环境
+使用下面的内容创建一个 `docker-compose.yml` 文件：
+
+ ```yaml
+ version: '2.1'
+ services:
+ Zookeeper:
Review Comment:
Unified indentation style？
##########
docs/content.zh/docs/get-started/quickstart/mysql-to-kafka.md:
##########
@@ -0,0 +1,588 @@
+---
+title: "MySQL 同步到 Kafka"
+weight: 2
+type: docs
+aliases:
+- /try-flink-cdc/pipeline-connectors/mysql-Kafka-pipeline-tutorial.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Streaming ELT 同步 MySQL 到 Kafka
+
+这篇教程将展示如何基于 Flink CDC 快速构建 MySQL 到 Kafka 的 Streaming ELT 作业，包含整库同步、表结构变更同步和分库分表同步的功能。
+本教程的演示都将在 Flink CDC CLI 中进行，无需一行 Java/Scala 代码，也无需安装 IDE。
+
+## 准备阶段
+准备一台已经安装了 Docker 的 Linux 或者 MacOS 电脑。
+
+### 准备 Flink Standalone 集群
+1. 下载 [Flink 1.20.1](https://archive.apache.org/dist/flink/flink-1.20.1/flink-1.20.1-bin-scala_2.12.tgz) ，解压后得到 flink-1.20.1 目录。
+ 使用下面的命令跳转至 Flink 目录下，并且设置 FLINK_HOME 为 flink-1.20.1 所在目录。
+
+ ```shell
+ tar -zxvf flink-1.20.1-bin-scala_2.12.tgz
+ exprot FLINK_HOME=$(pwd)/flink-1.20.1
+ cd flink-1.20.1
+ ```
+
+2. 通过在 conf/flink-conf.yaml 配置文件追加下列参数开启 checkpoint，每隔 3 秒做一次 checkpoint。
+
+ ```yaml
+ execution:
+ checkpointing:
+ interval: 3000
+ ```
+
+3. 使用下面的命令启动 Flink 集群。
+
+ ```shell
+ ./bin/start-cluster.sh
+ ```
+
+启动成功的话，可以在 [http://localhost:8081/](http://localhost:8081/) 访问到 Flink Web UI，如下所示：
+
+{{< img src="/fig/mysql-Kafka-tutorial/flink-ui.png" alt="Flink UI" >}}
+
+多次执行 start-cluster.sh 可以拉起多个 TaskManager。
+注：如果你是云服务器，无法访问本地，需要将 conf/config.yaml 里面 rest.bind-address 和 rest.address的 localhost 改成0.0.0.0，然后使用 公网IP:8081 即可访问。
+
+### 准备 Docker 环境
+使用下面的内容创建一个 `docker-compose.yml` 文件：
+
+ ```yaml
+ version: '2.1'
+ services:
+ Zookeeper:
+ image: zookeeper:3.7.1
+ ports:
+ - "2181:2181"
+ environment:
+ - ALLOW_ANONYMOUS_LOGIN=yes
+ Kafka:
+ image: bitnami/kafka:2.8.1
+ ports:
+ - "9092:9092"
+ - "9093:9093"
+ environment:
+ - ALLOW_PLAINTEXT_LISTENER=yes
+ - KAFKA_LISTENERS=PLAINTEXT://:9092
+ - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.67.2:9092
+ - KAFKA_ZOOKEEPER_CONNECT=192.168.67.2:2181
+ MySQL:
+ image: debezium/example-mysql:1.1
+ ports:
+ - "3306:3306"
+ environment:
+ - MYSQL_ROOT_PASSWORD=123456
+ - MYSQL_USER=mysqluser
+ - MYSQL_PASSWORD=mysqlpw
+ ```
+注意：文件里面的 192.168.67.2 为内网 IP，可通过 ifconfig 查找。
+该 Docker Compose 中包含的容器有：
+- MySQL: 包含商品信息的数据库 `app_db`
+- Kafka: 存储从 MySQL 中根据规则映射过来的结果表
+- Zookeeper：主要用于进行Kafka集群管理和协调
+
+在 `docker-compose.yml` 所在目录下执行下面的命令来启动本教程需要的组件：
+
+ ```shell
+ docker-compose up -d
+ ```
+
+该命令将以 detached 模式自动启动 Docker Compose 配置中定义的所有容器。你可以通过 docker ps 来观察上述的容器是否正常启动了,如下所示。
+{{< img src="/fig/mysql-Kafka-tutorial/docker-ps.png" alt="Docker ps" >}}
+#### 在 MySQL 数据库中准备数据
+1. 进入 MySQL 容器
+
+ ```shell
+ docker-compose exec MySQL mysql -uroot -p123456
+ ```
+
+2. 创建数据库 `app_db` 和表 `orders`,`products`,`shipments`，并插入数据
+
+ ```sql
+ -- 创建数据库
+ CREATE DATABASE app_db;
+
+ USE app_db;
+
+ -- 创建 orders 表
+ CREATE TABLE `orders` (
+ `id` INT NOT NULL,
+ `price` DECIMAL(10,2) NOT NULL,
+ PRIMARY KEY (`id`)
+ );
+
+ -- 插入数据
+ INSERT INTO `orders` (`id`, `price`) VALUES (1, 4.00);
+ INSERT INTO `orders` (`id`, `price`) VALUES (2, 100.00);
+
+ -- 创建 shipments 表
+ CREATE TABLE `shipments` (
+ `id` INT NOT NULL,
+ `city` VARCHAR(255) NOT NULL,
+ PRIMARY KEY (`id`)
+ );
+
+ -- 插入数据
+ INSERT INTO `shipments` (`id`, `city`) VALUES (1, 'beijing');
+ INSERT INTO `shipments` (`id`, `city`) VALUES (2, 'xian');
+
+ -- 创建 products 表
+ CREATE TABLE `products` (
+ `id` INT NOT NULL,
+ `product` VARCHAR(255) NOT NULL,
+ PRIMARY KEY (`id`)
+ );
+
+ -- 插入数据
+ INSERT INTO `products` (`id`, `product`) VALUES (1, 'Beer');
+ INSERT INTO `products` (`id`, `product`) VALUES (2, 'Cap');
+ INSERT INTO `products` (`id`, `product`) VALUES (3, 'Peanut');
+ ```
+
+## 通过 Flink CDC CLI 提交任务
+1. 下载下面列出的二进制压缩包，并解压得到目录 `flink-cdc-{{< param Version >}}`；
+ [flink-cdc-{{< param Version >}}-bin.tar.gz](https://www.apache.org/dyn/closer.lua/flink/flink-cdc-{{< param Version >}}/flink-cdc-{{< param Version >}}-bin.tar.gz)
+ flink-cdc-{{< param Version >}} 下会包含 `bin`、`lib`、`log`、`conf` 四个目录。
+
+2. 下载下面列出的 connector 包，并且移动到 lib 目录下；
+ **下载链接只对已发布的版本有效, SNAPSHOT 版本需要本地基于 master 或 release- 分支编译。**
+ **请注意，您需要将 jar 移动到 Flink CDC Home 的 lib 目录，而非 Flink Home 的 lib 目录下。**
+ - [MySQL pipeline connector {{< param Version >}}](https://repo1.maven.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-mysql/{{< param Version >}}/flink-cdc-pipeline-connector-mysql-{{< param Version >}}.jar)
+ - [Kafka pipeline connector {{< param Version >}}](https://repo1.maven.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-kafka/{{< param Version >}}/flink-cdc-pipeline-connector-kafka-{{< param Version >}}.jar)
+
+ 您还需要将下面的 Driver 包放在 Flink `lib` 目录下，或通过 `--jar` 参数将其传入 Flink CDC CLI，因为 CDC Connectors 不再包含这些 Drivers：
+ - [MySQL Connector Java](https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.27/mysql-connector-java-8.0.27.jar)
+
+3. 编写任务配置 yaml 文件。
+ 下面给出了一个整库同步的示例文件 mysql-to-Kafka.yaml：
+
+ ```yaml
+ ################################################################################
+ # Description: Sync MySQL all tables to Kafka
+ ################################################################################
+ source:
+ type: mysql
+ hostname: 0.0.0.0
+ port: 3306
+ username: root
+ password: 123456
+ tables: app_db.\.*
+ server-id: 5400-5404
+ server-time-zone: UTC
+
+ sink:
+ type: kafka
+ name: Kafka Sink
+ properties.bootstrap.servers: 0.0.0.0:9092
+ topic: yaml-mysql-kafka
+
+
+ pipeline:
+ name: MySQL to Kafka Pipeline
+ parallelism: 1
+
+
+ ```
Review Comment:
Remove extra blank lines
##########
docs/content.zh/docs/get-started/quickstart/mysql-to-kafka.md:
##########
@@ -0,0 +1,588 @@
+---
+title: "MySQL 同步到 Kafka"
+weight: 2
+type: docs
+aliases:
+- /try-flink-cdc/pipeline-connectors/mysql-Kafka-pipeline-tutorial.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Streaming ELT 同步 MySQL 到 Kafka
+
+这篇教程将展示如何基于 Flink CDC 快速构建 MySQL 到 Kafka 的 Streaming ELT 作业，包含整库同步、表结构变更同步和分库分表同步的功能。
+本教程的演示都将在 Flink CDC CLI 中进行，无需一行 Java/Scala 代码，也无需安装 IDE。
+
+## 准备阶段
+准备一台已经安装了 Docker 的 Linux 或者 MacOS 电脑。
+
+### 准备 Flink Standalone 集群
+1. 下载 [Flink 1.20.1](https://archive.apache.org/dist/flink/flink-1.20.1/flink-1.20.1-bin-scala_2.12.tgz) ，解压后得到 flink-1.20.1 目录。
+ 使用下面的命令跳转至 Flink 目录下，并且设置 FLINK_HOME 为 flink-1.20.1 所在目录。
+
+ ```shell
+ tar -zxvf flink-1.20.1-bin-scala_2.12.tgz
+ exprot FLINK_HOME=$(pwd)/flink-1.20.1
+ cd flink-1.20.1
+ ```
+
+2. 通过在 conf/flink-conf.yaml 配置文件追加下列参数开启 checkpoint，每隔 3 秒做一次 checkpoint。
+
+ ```yaml
+ execution:
+ checkpointing:
+ interval: 3000
+ ```
+
+3. 使用下面的命令启动 Flink 集群。
+
+ ```shell
+ ./bin/start-cluster.sh
+ ```
+
+启动成功的话，可以在 [http://localhost:8081/](http://localhost:8081/) 访问到 Flink Web UI，如下所示：
+
+{{< img src="/fig/mysql-Kafka-tutorial/flink-ui.png" alt="Flink UI" >}}
+
+多次执行 start-cluster.sh 可以拉起多个 TaskManager。
+注：如果你是云服务器，无法访问本地，需要将 conf/config.yaml 里面 rest.bind-address 和 rest.address的 localhost 改成0.0.0.0，然后使用 公网IP:8081 即可访问。
+
+### 准备 Docker 环境
+使用下面的内容创建一个 `docker-compose.yml` 文件：
+
+ ```yaml
+ version: '2.1'
+ services:
+ Zookeeper:
+ image: zookeeper:3.7.1
+ ports:
+ - "2181:2181"
+ environment:
+ - ALLOW_ANONYMOUS_LOGIN=yes
+ Kafka:
+ image: bitnami/kafka:2.8.1
+ ports:
+ - "9092:9092"
+ - "9093:9093"
+ environment:
+ - ALLOW_PLAINTEXT_LISTENER=yes
+ - KAFKA_LISTENERS=PLAINTEXT://:9092
+ - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.67.2:9092
+ - KAFKA_ZOOKEEPER_CONNECT=192.168.67.2:2181
+ MySQL:
+ image: debezium/example-mysql:1.1
+ ports:
+ - "3306:3306"
+ environment:
+ - MYSQL_ROOT_PASSWORD=123456
+ - MYSQL_USER=mysqluser
+ - MYSQL_PASSWORD=mysqlpw
+ ```
+注意：文件里面的 192.168.67.2 为内网 IP，可通过 ifconfig 查找。
Review Comment:
192.168.67.2 -> kafka or zookeeper?
In the same docker network, we can use service names to communicate directly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-36980] [cdc-connect] Build a new introductory practice "Streaming ELT from MySQL to Kafka" [flink-cdc]
Posted by "Sukang1002 (via GitHub)" <gi...@apache.org>.
Sukang1002 commented on code in PR #4033:
URL: https://github.com/apache/flink-cdc/pull/4033#discussion_r2348664016
##########
docs/content/docs/get-started/quickstart/mysql-to-kafka.md:
##########
@@ -0,0 +1,590 @@
+---
+title: "MySQL to Kafka"
+weight: 2
Review Comment:
Thank you for your correction.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-36980] [cdc-connect] Build a new introductory practice "Streaming ELT from MySQL to Kafka" [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on PR #4033:
URL: https://github.com/apache/flink-cdc/pull/4033#issuecomment-3292905947
Just polished docs for a little bit, PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-36980] [cdc-connect] Build a new introductory practice "Streaming ELT from MySQL to Kafka" [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on code in PR #4033:
URL: https://github.com/apache/flink-cdc/pull/4033#discussion_r2265447889
##########
docs/content/docs/get-started/quickstart/mysql-to-kafka.md:
##########
@@ -0,0 +1,591 @@
+---
+title: "MySQL to Kafka"
+weight: 2
+type: docs
+aliases:
+- /try-flink-cdc/pipeline-connectors/mysql-Kafka-pipeline-tutorial.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Streaming ELT from MySQL to Kafka
+
+This tutorial is to show how to quickly build a Streaming ELT job from MySQL to StarRocks using Flink CDC, including the
+feature of sync all table of one database, schema change evolution and sync sharding tables into one table.
+All exercises in this tutorial are performed in the Flink CDC 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.
+
+
+### Prepare Flink Standalone cluster
+1. Download [Flink 1.20.1](https://archive.apache.org/dist/flink/flink-1.20.1/flink-1.20.1-bin-scala_2.12.tgz) ，unzip and get flink-1.20.1 directory.
+ Use the following command to navigate to the Flink directory and set FLINK_HOME to the directory where flink-1.20.1 is located.
+
+ ```shell
+ tar -zxvf flink-1.20.1-bin-scala_2.12.tgz
+ exprot FLINK_HOME=$(pwd)/flink-1.20.1
+ cd flink-1.20.1
+ ```
+
+2. Enable checkpointing by appending the following parameters to the conf/flink-conf.yaml configuration file to perform a checkpoint every 3 seconds.
+
+ ```yaml
+ execution:
+ checkpointing:
+ interval: 3000
+ ```
+
+3. Start the Flink cluster using the following command.
+
+ ```shell
+ ./bin/start-cluster.sh
+ ```
+
+If successfully started, you can access the Flink Web UI at [http://localhost:8081/](http://localhost:8081/), as shown below.
+
+{{< img src="/fig/mysql-Kafka-tutorial/flink-ui.png" alt="Flink UI" >}}
+
+Executing `start-cluster.sh` multiple times can start multiple TaskManager‘s.
+
+Note: If you are a cloud server and cannot access the local area, you need to change the localhost of rest.bd-address and rest.address in conf/config.yaml to 0.0.0.0, and then use the public IP address:8081 to access it.
+### Prepare docker compose
+The following tutorial will prepare the required components using `docker-compose`.
+Create a `docker-compose.yml` file using the content provided below:
+
+ ```yaml
+ version: '2.1'
+ services:
+ Zookeeper:
+ image: zookeeper:3.7.1
+ ports:
+ - "2181:2181"
+ environment:
+ - ALLOW_ANONYMOUS_LOGIN=yes
+ Kafka:
+ image: bitnami/kafka:2.8.1
+ ports:
+ - "9092:9092"
+ - "9093:9093"
+ environment:
+ - ALLOW_PLAINTEXT_LISTENER=yes
+ - KAFKA_LISTENERS=PLAINTEXT://:9092
+ - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.67.2:9092
+ - KAFKA_ZOOKEEPER_CONNECT=192.168.67.2:2181
+ MySQL:
+ image: debezium/example-mysql:1.1
+ ports:
+ - "3306:3306"
+ environment:
+ - MYSQL_ROOT_PASSWORD=123456
+ - MYSQL_USER=mysqluser
+ - MYSQL_PASSWORD=mysqlpw
+ ```
+Note: The 192.168.67.2 in the file is an internal network IP and can be found through ifconfig.
+The Docker Compose should include the following services (containers):
+- MySQL: include a database named `app_db`
+- Kafka: Store the result table mapped from MySQL according to the rules
+- Zookeeper：It is mainly used for Kafka cluster management and coordination
+
+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.
+{{< img src="/fig/mysql-Kafka-tutorial/docker-ps.png" alt="Docker ps" >}}
+#### Prepare records for MySQL
+1. Enter MySQL container
+
+ ```shell
+ docker-compose exec MySQL mysql -uroot -p123456
+ ```
+
+2. create `app_db` database and `orders`,`products`,`shipments` tables, then insert records
+
+ ```sql
+ -- create database
+ CREATE DATABASE app_db;
+
+ USE app_db;
+
+ -- create orders table
+ CREATE TABLE `orders` (
+ `id` INT NOT NULL,
+ `price` DECIMAL(10,2) NOT NULL,
+ PRIMARY KEY (`id`)
+ );
+
+ -- insert records
+ INSERT INTO `orders` (`id`, `price`) VALUES (1, 4.00);
+ INSERT INTO `orders` (`id`, `price`) VALUES (2, 100.00);
+
+ -- create shipments table
+ CREATE TABLE `shipments` (
+ `id` INT NOT NULL,
+ `city` VARCHAR(255) NOT NULL,
+ PRIMARY KEY (`id`)
+ );
+
+ -- insert records
+ INSERT INTO `shipments` (`id`, `city`) VALUES (1, 'beijing');
+ INSERT INTO `shipments` (`id`, `city`) VALUES (2, 'xian');
+
+ -- create products table
+ CREATE TABLE `products` (
+ `id` INT NOT NULL,
+ `product` VARCHAR(255) NOT NULL,
+ PRIMARY KEY (`id`)
+ );
+
+ -- insert records
+ INSERT INTO `products` (`id`, `product`) VALUES (1, 'Beer');
+ INSERT INTO `products` (`id`, `product`) VALUES (2, 'Cap');
+ INSERT INTO `products` (`id`, `product`) VALUES (3, 'Peanut');
+ ```
+
+## Submit job with Flink CDC CLI
+1. Download the binary compressed packages listed below and extract them to the directory `flink cdc-{{< param Version >}}'`：
+ [flink-cdc-{{< param Version >}}-bin.tar.gz](https://www.apache.org/dyn/closer.lua/flink/flink-cdc-{{< param Version >}}/flink-cdc-{{< param Version >}}-bin.tar.gz)
+ flink-cdc-{{< param Version >}} directory will contain four directory: `bin`, `lib`, `log`, and `conf`.
+
+2. Download the connector package listed below and move it to the `lib` directory
+ **Download links are available only for stable releases, SNAPSHOT dependencies need to be built based on master or release branches by yourself.**
+ **Please note that you need to move the jar to the lib directory of Flink CDC Home, not to the lib directory of Flink Home.**
+ - [MySQL pipeline connector {{< param Version >}}](https://repo1.maven.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-mysql/{{< param Version >}}/flink-cdc-pipeline-connector-mysql-{{< param Version >}}.jar)
+ - [Kafka pipeline connector {{< param Version >}}](https://repo1.maven.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-kafka/{{< param Version >}}/flink-cdc-pipeline-connector-kafka-{{< param Version >}}.jar)
+
+ You also need to place MySQL connector into Flink `lib` folder or pass it with `--jar` argument, since they're no longer packaged with CDC connectors:
+ - [MySQL Connector Java](https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.27/mysql-connector-java-8.0.27.jar)
+
+3. Write task configuration yaml file.
+ Here is an example file for synchronizing the entire database `mysql-to-kafka.yaml`：
+
+ ```yaml
+ ################################################################################
+ # Description: Sync MySQL all tables to Kafka
+ ################################################################################
+ source:
+ type: mysql
+ hostname: 0.0.0.0
+ port: 3306
+ username: root
+ password: 123456
+ tables: app_db.\.*
+ server-id: 5400-5404
+ server-time-zone: UTC
+
+ sink:
+ type: kafka
+ name: Kafka Sink
+ properties.bootstrap.servers: 0.0.0.0:9092
+ topic: yaml-mysql-kafka
+
+ pipeline:
+ name: MySQL to Kafka Pipeline
+ parallelism: 1
+ ```
+
+Notice that:
+* `tables: app_db.\.*` in source synchronize all tables in `app_db` through Regular Matching.
+
+4. Finally, submit job to Flink Standalone cluster using Cli.
+
+ ```shell
+ bash bin/flink-cdc.sh mysql-to-kafka.yaml
+ #For reference, some examples of custom paths are mainly used in situations such as multiple versions of flink and inconsistent mysql drivers, as follows
+ #bash /root/flink-cdc-3.4.0/bin/flink-cdc.sh /root/flink-cdc-3.4.0/bin/mysql-to-kafka.yaml --flink-home /root/flink-1.20.1 --jar /root/flink-cdc-3.4.0/lib/mysql-connector-java-8.0.27.jar
+ ```
+
+After successful submission, the return information is as follows：
+
+ ```shell
+ Pipeline has been submitted to cluster.
+ Job ID: 04fd88ccb96c789dce2bf0b3a541d626
+ Job Description: MySQL to Kafka Pipeline
+ ```
+
+We can find a job named `Sync MySQL Database to Kafka` is running through Flink Web UI.
+
+{{< img src="/fig/mysql-Kafka-tutorial/mysql-to-Kafka.png" alt="MySQL-to-Kafka" >}}
+
+The Topic situation can be viewed through the built-in client of kafka to obtain the content in debezium-json format:
+```shell
+ docker-compose exec Kafka kafka-console-consumer.sh --bootstrap-server 192.168.67.2:9092 --topic yaml-mysql-kafka --from-beginning
+```
+The debezium-json format contains several elements such as before,after,op, and source. The demonstration example is as follows:
+```json
+{
+ "before": null,
+ "after": {
+ "id": 1,
+ "price": 4
+ },
+ "op": "c",
+ "source": {
+ "db": "app_db",
+ "table": "orders"
+ }
+}
+...
+{
+ "before": null,
+ "after": {
+ "id": 1,
+ "product": "Beer"
+ },
+ "op": "c",
+ "source": {
+ "db": "app_db",
+ "table": "products"
+ }
+}
+...
+{
+ "before": null,
+ "after": {
+ "id": 2,
+ "city": "xian"
+ },
+ "op": "c",
+ "source": {
+ "db": "app_db",
+ "table": "shipments"
+ }
+}
+```
+### Synchronize Schema and Data changes
+Enter MySQL container
+
+ ```shell
+ docker-compose exec mysql mysql -uroot -p123456
+ ```
+
+Then, modify schema and record in MySQL, and the tables of StarRocks will change the same in real time：
Review Comment:
should it be StarRocks?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-36980] [cdc-connect] Build a new introductory practice "Streaming ELT from MySQL to Kafka" [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #4033:
URL: https://github.com/apache/flink-cdc/pull/4033#discussion_r2347814005
##########
docs/content.zh/docs/get-started/quickstart/mysql-to-kafka.md:
##########
@@ -0,0 +1,586 @@
+---
+title: "MySQL 同步到 Kafka"
+weight: 2
Review Comment:
Set weight to 4? Weight 2 has been occupied by `mysql-to-starrocks`.
##########
docs/content/docs/get-started/quickstart/mysql-to-kafka.md:
##########
@@ -0,0 +1,590 @@
+---
+title: "MySQL to Kafka"
+weight: 2
Review Comment:
Ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Flink-36980] [cdc-connect] Build a new introductory practice "Streaming ELT from MySQL to Kafka" [flink-cdc]
Posted by "mielientiev (via GitHub)" <gi...@apache.org>.
mielientiev commented on code in PR #4033:
URL: https://github.com/apache/flink-cdc/pull/4033#discussion_r2265450418
##########
docs/content/docs/get-started/quickstart/mysql-to-kafka.md:
##########
@@ -0,0 +1,591 @@
+---
+title: "MySQL to Kafka"
+weight: 2
+type: docs
+aliases:
+- /try-flink-cdc/pipeline-connectors/mysql-Kafka-pipeline-tutorial.html
+---
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Streaming ELT from MySQL to Kafka
+
+This tutorial is to show how to quickly build a Streaming ELT job from MySQL to StarRocks using Flink CDC, including the
+feature of sync all table of one database, schema change evolution and sync sharding tables into one table.
+All exercises in this tutorial are performed in the Flink CDC 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.
+
+
+### Prepare Flink Standalone cluster
+1. Download [Flink 1.20.1](https://archive.apache.org/dist/flink/flink-1.20.1/flink-1.20.1-bin-scala_2.12.tgz) ，unzip and get flink-1.20.1 directory.
+ Use the following command to navigate to the Flink directory and set FLINK_HOME to the directory where flink-1.20.1 is located.
+
+ ```shell
+ tar -zxvf flink-1.20.1-bin-scala_2.12.tgz
+ exprot FLINK_HOME=$(pwd)/flink-1.20.1
+ cd flink-1.20.1
+ ```
+
+2. Enable checkpointing by appending the following parameters to the conf/flink-conf.yaml configuration file to perform a checkpoint every 3 seconds.
+
+ ```yaml
+ execution:
+ checkpointing:
+ interval: 3000
+ ```
+
+3. Start the Flink cluster using the following command.
+
+ ```shell
+ ./bin/start-cluster.sh
+ ```
+
+If successfully started, you can access the Flink Web UI at [http://localhost:8081/](http://localhost:8081/), as shown below.
+
+{{< img src="/fig/mysql-Kafka-tutorial/flink-ui.png" alt="Flink UI" >}}
+
+Executing `start-cluster.sh` multiple times can start multiple TaskManager‘s.
+
+Note: If you are a cloud server and cannot access the local area, you need to change the localhost of rest.bd-address and rest.address in conf/config.yaml to 0.0.0.0, and then use the public IP address:8081 to access it.
+### Prepare docker compose
+The following tutorial will prepare the required components using `docker-compose`.
+Create a `docker-compose.yml` file using the content provided below:
+
+ ```yaml
+ version: '2.1'
+ services:
+ Zookeeper:
+ image: zookeeper:3.7.1
+ ports:
+ - "2181:2181"
+ environment:
+ - ALLOW_ANONYMOUS_LOGIN=yes
+ Kafka:
+ image: bitnami/kafka:2.8.1
+ ports:
+ - "9092:9092"
+ - "9093:9093"
+ environment:
+ - ALLOW_PLAINTEXT_LISTENER=yes
+ - KAFKA_LISTENERS=PLAINTEXT://:9092
+ - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.67.2:9092
+ - KAFKA_ZOOKEEPER_CONNECT=192.168.67.2:2181
+ MySQL:
+ image: debezium/example-mysql:1.1
+ ports:
+ - "3306:3306"
+ environment:
+ - MYSQL_ROOT_PASSWORD=123456
+ - MYSQL_USER=mysqluser
+ - MYSQL_PASSWORD=mysqlpw
+ ```
+Note: The 192.168.67.2 in the file is an internal network IP and can be found through ifconfig.
+The Docker Compose should include the following services (containers):
+- MySQL: include a database named `app_db`
+- Kafka: Store the result table mapped from MySQL according to the rules
+- Zookeeper：It is mainly used for Kafka cluster management and coordination
+
+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.
+{{< img src="/fig/mysql-Kafka-tutorial/docker-ps.png" alt="Docker ps" >}}
+#### Prepare records for MySQL
+1. Enter MySQL container
+
+ ```shell
+ docker-compose exec MySQL mysql -uroot -p123456
+ ```
+
+2. create `app_db` database and `orders`,`products`,`shipments` tables, then insert records
+
+ ```sql
+ -- create database
+ CREATE DATABASE app_db;
+
+ USE app_db;
+
+ -- create orders table
+ CREATE TABLE `orders` (
+ `id` INT NOT NULL,
+ `price` DECIMAL(10,2) NOT NULL,
+ PRIMARY KEY (`id`)
+ );
+
+ -- insert records
+ INSERT INTO `orders` (`id`, `price`) VALUES (1, 4.00);
+ INSERT INTO `orders` (`id`, `price`) VALUES (2, 100.00);
+
+ -- create shipments table
+ CREATE TABLE `shipments` (
+ `id` INT NOT NULL,
+ `city` VARCHAR(255) NOT NULL,
+ PRIMARY KEY (`id`)
+ );
+
+ -- insert records
+ INSERT INTO `shipments` (`id`, `city`) VALUES (1, 'beijing');
+ INSERT INTO `shipments` (`id`, `city`) VALUES (2, 'xian');
+
+ -- create products table
+ CREATE TABLE `products` (
+ `id` INT NOT NULL,
+ `product` VARCHAR(255) NOT NULL,
+ PRIMARY KEY (`id`)
+ );
+
+ -- insert records
+ INSERT INTO `products` (`id`, `product`) VALUES (1, 'Beer');
+ INSERT INTO `products` (`id`, `product`) VALUES (2, 'Cap');
+ INSERT INTO `products` (`id`, `product`) VALUES (3, 'Peanut');
+ ```
+
+## Submit job with Flink CDC CLI
+1. Download the binary compressed packages listed below and extract them to the directory `flink cdc-{{< param Version >}}'`：
+ [flink-cdc-{{< param Version >}}-bin.tar.gz](https://www.apache.org/dyn/closer.lua/flink/flink-cdc-{{< param Version >}}/flink-cdc-{{< param Version >}}-bin.tar.gz)
+ flink-cdc-{{< param Version >}} directory will contain four directory: `bin`, `lib`, `log`, and `conf`.
+
+2. Download the connector package listed below and move it to the `lib` directory
+ **Download links are available only for stable releases, SNAPSHOT dependencies need to be built based on master or release branches by yourself.**
+ **Please note that you need to move the jar to the lib directory of Flink CDC Home, not to the lib directory of Flink Home.**
+ - [MySQL pipeline connector {{< param Version >}}](https://repo1.maven.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-mysql/{{< param Version >}}/flink-cdc-pipeline-connector-mysql-{{< param Version >}}.jar)
+ - [Kafka pipeline connector {{< param Version >}}](https://repo1.maven.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-kafka/{{< param Version >}}/flink-cdc-pipeline-connector-kafka-{{< param Version >}}.jar)
+
+ You also need to place MySQL connector into Flink `lib` folder or pass it with `--jar` argument, since they're no longer packaged with CDC connectors:
+ - [MySQL Connector Java](https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.27/mysql-connector-java-8.0.27.jar)
+
+3. Write task configuration yaml file.
+ Here is an example file for synchronizing the entire database `mysql-to-kafka.yaml`：
+
+ ```yaml
+ ################################################################################
+ # Description: Sync MySQL all tables to Kafka
+ ################################################################################
+ source:
+ type: mysql
+ hostname: 0.0.0.0
+ port: 3306
+ username: root
+ password: 123456
+ tables: app_db.\.*
+ server-id: 5400-5404
+ server-time-zone: UTC
+
+ sink:
+ type: kafka
+ name: Kafka Sink
+ properties.bootstrap.servers: 0.0.0.0:9092
+ topic: yaml-mysql-kafka
+
+ pipeline:
+ name: MySQL to Kafka Pipeline
+ parallelism: 1
+ ```
+
+Notice that:
+* `tables: app_db.\.*` in source synchronize all tables in `app_db` through Regular Matching.
+
+4. Finally, submit job to Flink Standalone cluster using Cli.
+
+ ```shell
+ bash bin/flink-cdc.sh mysql-to-kafka.yaml
+ #For reference, some examples of custom paths are mainly used in situations such as multiple versions of flink and inconsistent mysql drivers, as follows
+ #bash /root/flink-cdc-3.4.0/bin/flink-cdc.sh /root/flink-cdc-3.4.0/bin/mysql-to-kafka.yaml --flink-home /root/flink-1.20.1 --jar /root/flink-cdc-3.4.0/lib/mysql-connector-java-8.0.27.jar
+ ```
+
+After successful submission, the return information is as follows：
+
+ ```shell
+ Pipeline has been submitted to cluster.
+ Job ID: 04fd88ccb96c789dce2bf0b3a541d626
+ Job Description: MySQL to Kafka Pipeline
+ ```
+
+We can find a job named `Sync MySQL Database to Kafka` is running through Flink Web UI.
+
+{{< img src="/fig/mysql-Kafka-tutorial/mysql-to-Kafka.png" alt="MySQL-to-Kafka" >}}
+
+The Topic situation can be viewed through the built-in client of kafka to obtain the content in debezium-json format:
+```shell
+ docker-compose exec Kafka kafka-console-consumer.sh --bootstrap-server 192.168.67.2:9092 --topic yaml-mysql-kafka --from-beginning
+```
+The debezium-json format contains several elements such as before,after,op, and source. The demonstration example is as follows:
+```json
+{
+ "before": null,
+ "after": {
+ "id": 1,
+ "price": 4
+ },
+ "op": "c",
+ "source": {
+ "db": "app_db",
+ "table": "orders"
+ }
+}
+...
+{
+ "before": null,
+ "after": {
+ "id": 1,
+ "product": "Beer"
+ },
+ "op": "c",
+ "source": {
+ "db": "app_db",
+ "table": "products"
+ }
+}
+...
+{
+ "before": null,
+ "after": {
+ "id": 2,
+ "city": "xian"
+ },
+ "op": "c",
+ "source": {
+ "db": "app_db",
+ "table": "shipments"
+ }
+}
+```
+### Synchronize Schema and Data changes
+Enter MySQL container
+
+ ```shell
+ docker-compose exec mysql mysql -uroot -p123456
+ ```
+
+Then, modify schema and record in MySQL, and the tables of StarRocks will change the same in real time：
+1. insert one record in `orders` from MySQL:
+
+ ```sql
+ INSERT INTO app_db.orders (id, price) VALUES (3, 100.00);
+ ```
+
+2. add one column in `orders` from MySQL:
+
+ ```sql
+ ALTER TABLE app_db.orders ADD amount varchar(100) NULL;
+ ```
+
+3. update one record in `orders` from MySQL:
+
+ ```sql
+ UPDATE app_db.orders SET price=100.00, amount=100.00 WHERE id=1;
+ ```
+4. delete one record in `orders` from MySQL:
+
+ ```sql
+ DELETE FROM app_db.orders WHERE id=2;
+ ```
+By monitoring the topic through consumers, we can see that these changes are also taking place in real time on Kafka:
+```json
+{
+ "before": {
+ "id": 1,
+ "price": 4,
+ "amount": null
+ },
+ "after": {
+ "id": 1,
+ "price": 100,
+ "amount": "100.00"
+ },
+ "op": "u",
+ "source": {
+ "db": "app_db",
+ "table": "orders"
+ }
+}
+```
+Similarly, by modifying the `shipments`,`products` table, you can see the results of the synchronized changes in real time at the corresponding topic in Kafka.
+
+### Route the changes
+Flink CDC provides the configuration to route the table structure/data of the source table to other table names.
+With this ability, we can achieve functions such as table name, database name replacement, and whole database synchronization.
+Here is an example file for using `route` feature:
+ ```yaml
+ ################################################################################
+ # Description: Sync MySQL all tables to Kafka
+ ################################################################################
+ source:
+ type: mysql
+ hostname: localhost
+ port: 3306
+ username: root
+ password: 123456
+ tables: app_db.\.*
+ server-id: 5400-5404
+ server-time-zone: UTC
+
+ sink:
+ type: kafka
+ name: Kafka Sink
+ properties.bootstrap.servers: 0.0.0.0:9092
+ pipeline:
+ name: MySQL to Kafka Pipeline
+ parallelism: 1
+ route:
+ - source-table: app_db.orders
+ sink-table: kafka_ods_orders
+ - source-table: app_db.shipments
+ sink-table: kafka_ods_shipments
+ - source-table: app_db.products
+ sink-table: kafka_ods_products
+ ```
+
+Using the upper `route` configuration, we can synchronize the table schema and data of `app_db.orders` to `kafka_ods_orders`, thus achieving the function of database migration.
+Specifically, `source-table` support regular expression matching with multiple tables to synchronize sharding databases and tables. like the following：
+
+ ```yaml
+ route:
+ - source-table: app_db.order\.*
+ sink-table: kafka_ods_orders
+ ```
+
+In this way, we can synchronize sharding tables like `app_db.order01`、`app_db.order02`、`app_db.order03` into one kafka_ods_orders tables.By using the built-in tools of kafka, you can view the successful establishment of the corresponding Topic. Data details can be queried using kafka-console-Consumer.sh:
Review Comment:
```suggestion
In this way, we can synchronize sharding tables like `app_db.order01`、`app_db.order02`、`app_db.order03` into one kafka_ods_orders topic. By using the built-in tools of kafka, you can view the successful establishment of the corresponding Topic. Data details can be queried using kafka-console-Consumer.sh:
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org