You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "JingsongLi (via GitHub)" <gi...@apache.org> on 2023/05/23 08:27:18 UTC
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #1200: [flink] Supports Kafka with canal CDC format ingestion (#911)
JingsongLi commented on code in PR #1200:
URL: https://github.com/apache/incubator-paimon/pull/1200#discussion_r1201741990
##########
docs/content/how-to/cdc-ingestion.md:
##########
@@ -157,3 +157,187 @@ Example
--table-conf changelog-producer=input \
--table-conf sink.parallelism=4
```
+
+## Kafka
+
+### Prepare Kafka Bundled Jar
+
+```
+flink-sql-connector-kafka-*.jar
+```
+### Supported Formats
+Flink provides several Kafka CDC formats :canal-json、debezium-json,ogg-json,maxwell-json.
+If a message in a Kafka topic is a change event captured from another database using the Change Data Capture (CDC) tool, then you can use the Paimon Kafka CDC. Write the INSERT, UPDATE, DELETE messages parsed into the paimon table.
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left">Formats</th>
+ <th class="text-left">Supported</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/canal/">Canal CDC</a></td>
+ <td>True</td>
+ </tr>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/debezium/">Debezium CDC</a></td>
+ <td>False</td>
+ </tr>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/maxwell/ >}}">Maxwell CDC</a></td>
+ <td>False</td>
+ </tr>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/ogg/">OGG CDC</a></td>
+ <td>False</td>
+ </tr>
+ </tbody>
+</table>
+
+### Synchronizing Tables
+
+By using [KafkaSyncTableAction](/docs/{{< param Branch >}}/api/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction) in a Flink DataStream job or directly through `flink run`, users can synchronize one or multiple tables from Kafka's one topic into one Paimon table.
+
+To use this feature through `flink run`, run the following shell command.
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ -c org.apache.paimon.flink.action.FlinkActions \
+ /path/to/paimon-flink-**-{{< version >}}.jar \
+ kafka-sync-table
+ --warehouse <warehouse-path> \
+ --database <database-name> \
+ --table <table-name> \
+ [--partition-keys <partition-keys>] \
+ [--primary-keys <primary-keys>] \
+ [--computed-column <'column-name=expr-name(args[, ...])'> [--computed-column ...]] \
+ [--kafka-conf <kafka-source-conf> [--kafka-conf <kafka-source-conf> ...]] \
+ [--catalog-conf <paimon-catalog-conf> [--catalog-conf <paimon-catalog-conf> ...]] \
+ [--table-conf <paimon-table-sink-conf> [--table-conf <paimon-table-sink-conf> ...]]
+```
+
+* `--warehouse` is the path to Paimon warehouse.
+* `--database` is the database name in Paimon catalog.
+* `--table` is the Paimon table name.
+* `--partition-keys` are the partition keys for Paimon table. If there are multiple partition keys, connect them with comma, for example `dt,hh,mm`.
+* `--primary-keys` are the primary keys for Paimon table. If there are multiple primary keys, connect them with comma, for example `buyer_id,seller_id`.
+* `--computed-column` are the definitions of computed columns. The argument field is from Kafka topic's table field name. Supported expressions are:
+ * year(date-column): Extract year from a DATE, DATETIME or TIMESTAMP. Output is an INT value represent the year.
+ * substring(column,beginInclusive): Get column.substring(beginInclusive). Output is a STRING.
+ * substring(column,beginInclusive,endExclusive): Get column.substring(beginInclusive,endExclusive). Output is a STRING.
+ * truncate(column,width): truncate column by width. Output type is same with column.
+ * If the column is a STRING, truncate(column,width) will truncate the string to width characters, namely `value.substring(0, width)`.
+ * If the column is an INT or LONG, truncate(column,width) will truncate the number with the algorithm `v - (((v % W) + W) % W)`. The `redundant` compute part is to keep the result always positive.
+ * If the column is a DECIMAL, truncate(column,width) will truncate the decimal with the algorithm: let `scaled_W = decimal(W, scale(v))`, then return `v - (v % scaled_W)`.
+* `--kafka-conf` is the configuration for Flink Kafka sources. Each configuration should be specified in the format `key=value`. `properties.bootstrap.servers`, `topic`, `properties.group.id`, and `value.format` are required configurations, others are optional. See its [document](https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/kafka/#%e8%bf%9e%e6%8e%a5%e5%99%a8%e5%8f%82%e6%95%b0) for a complete list of configurations.
+* `--catalog-conf` is the configuration for Paimon catalog. Each configuration should be specified in the format `key=value`. See [here]({{< ref "maintenance/configurations" >}}) for a complete list of catalog configurations.
+* `--table-conf` is the configuration for Paimon table sink. Each configuration should be specified in the format `key=value`. See [here]({{< ref "maintenance/configurations" >}}) for a complete list of table configurations.
+
+If the Paimon table you specify does not exist, this action will automatically create the table. Its schema will be derived from all specified Kafka topic's tables. If the Paimon table already exists, its schema will be compared against the schema of all specified Kafka topic's tables.
Review Comment:
> Its schema will be derived from all specified Kafka topic's tables
How to infer schema from topic, we can document this.
##########
docs/content/how-to/cdc-ingestion.md:
##########
@@ -157,3 +157,187 @@ Example
--table-conf changelog-producer=input \
--table-conf sink.parallelism=4
```
+
+## Kafka
+
+### Prepare Kafka Bundled Jar
+
+```
+flink-sql-connector-kafka-*.jar
+```
+### Supported Formats
+Flink provides several Kafka CDC formats :canal-json、debezium-json,ogg-json,maxwell-json.
+If a message in a Kafka topic is a change event captured from another database using the Change Data Capture (CDC) tool, then you can use the Paimon Kafka CDC. Write the INSERT, UPDATE, DELETE messages parsed into the paimon table.
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left">Formats</th>
+ <th class="text-left">Supported</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/canal/">Canal CDC</a></td>
+ <td>True</td>
+ </tr>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/debezium/">Debezium CDC</a></td>
+ <td>False</td>
+ </tr>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/maxwell/ >}}">Maxwell CDC</a></td>
+ <td>False</td>
+ </tr>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/ogg/">OGG CDC</a></td>
+ <td>False</td>
+ </tr>
+ </tbody>
+</table>
+
+### Synchronizing Tables
+
+By using [KafkaSyncTableAction](/docs/{{< param Branch >}}/api/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction) in a Flink DataStream job or directly through `flink run`, users can synchronize one or multiple tables from Kafka's one topic into one Paimon table.
+
+To use this feature through `flink run`, run the following shell command.
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ -c org.apache.paimon.flink.action.FlinkActions \
+ /path/to/paimon-flink-**-{{< version >}}.jar \
+ kafka-sync-table
+ --warehouse <warehouse-path> \
+ --database <database-name> \
+ --table <table-name> \
+ [--partition-keys <partition-keys>] \
+ [--primary-keys <primary-keys>] \
+ [--computed-column <'column-name=expr-name(args[, ...])'> [--computed-column ...]] \
+ [--kafka-conf <kafka-source-conf> [--kafka-conf <kafka-source-conf> ...]] \
+ [--catalog-conf <paimon-catalog-conf> [--catalog-conf <paimon-catalog-conf> ...]] \
+ [--table-conf <paimon-table-sink-conf> [--table-conf <paimon-table-sink-conf> ...]]
+```
+
+* `--warehouse` is the path to Paimon warehouse.
+* `--database` is the database name in Paimon catalog.
+* `--table` is the Paimon table name.
+* `--partition-keys` are the partition keys for Paimon table. If there are multiple partition keys, connect them with comma, for example `dt,hh,mm`.
+* `--primary-keys` are the primary keys for Paimon table. If there are multiple primary keys, connect them with comma, for example `buyer_id,seller_id`.
+* `--computed-column` are the definitions of computed columns. The argument field is from Kafka topic's table field name. Supported expressions are:
+ * year(date-column): Extract year from a DATE, DATETIME or TIMESTAMP. Output is an INT value represent the year.
+ * substring(column,beginInclusive): Get column.substring(beginInclusive). Output is a STRING.
+ * substring(column,beginInclusive,endExclusive): Get column.substring(beginInclusive,endExclusive). Output is a STRING.
+ * truncate(column,width): truncate column by width. Output type is same with column.
+ * If the column is a STRING, truncate(column,width) will truncate the string to width characters, namely `value.substring(0, width)`.
+ * If the column is an INT or LONG, truncate(column,width) will truncate the number with the algorithm `v - (((v % W) + W) % W)`. The `redundant` compute part is to keep the result always positive.
+ * If the column is a DECIMAL, truncate(column,width) will truncate the decimal with the algorithm: let `scaled_W = decimal(W, scale(v))`, then return `v - (v % scaled_W)`.
+* `--kafka-conf` is the configuration for Flink Kafka sources. Each configuration should be specified in the format `key=value`. `properties.bootstrap.servers`, `topic`, `properties.group.id`, and `value.format` are required configurations, others are optional. See its [document](https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/kafka/#%e8%bf%9e%e6%8e%a5%e5%99%a8%e5%8f%82%e6%95%b0) for a complete list of configurations.
+* `--catalog-conf` is the configuration for Paimon catalog. Each configuration should be specified in the format `key=value`. See [here]({{< ref "maintenance/configurations" >}}) for a complete list of catalog configurations.
+* `--table-conf` is the configuration for Paimon table sink. Each configuration should be specified in the format `key=value`. See [here]({{< ref "maintenance/configurations" >}}) for a complete list of table configurations.
+
+If the Paimon table you specify does not exist, this action will automatically create the table. Its schema will be derived from all specified Kafka topic's tables. If the Paimon table already exists, its schema will be compared against the schema of all specified Kafka topic's tables.
+
+This action supports a limited number of schema changes. Currently, the framework can not drop columns, so the behaviors of `DROP` will be ignored, `RENAME` will add a new column. Currently supported schema changes includes:
+
+* Adding columns.
+
+* Altering column types. More specifically,
+
+ * altering from a string type (char, varchar, text) to another string type with longer length,
+ * altering from a binary type (binary, varbinary, blob) to another binary type with longer length,
+ * altering from an integer type (tinyint, smallint, int, bigint) to another integer type with wider range,
+ * altering from a floating-point type (float, double) to another floating-point type with wider range,
+
+ are supported.
+
+Example
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ -c org.apache.paimon.flink.action.FlinkActions \
+ /path/to/paimon-flink-**-{{< version >}}.jar \
+ kafka-sync-table \
+ --warehouse hdfs:///path/to/warehouse \
+ --database test_db \
+ --table test_table \
+ --partition-keys pt \
+ --primary-keys pt,uid \
+ --computed-columns '_year=year(age)' \
+ --kafka-conf properties.bootstrap.servers=127.0.0.1:9020 \
+ --kafka-conf topic=order \
+ --kafka-conf properties.group.id=123456 \
+ --kafka-conf value.format=canal-json \
+ --catalog-conf metastore=hive \
+ --catalog-conf uri=thrift://hive-metastore:9083 \
+ --table-conf bucket=4 \
+ --table-conf changelog-producer=input \
+ --table-conf sink.parallelism=4
+```
+
+### Synchronizing Databases
+
+By using [KafkaSyncDatabaseAction](/docs/{{< param Branch >}}/api/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction) in a Flink DataStream job or directly through `flink run`, users can synchronize the multi kafka topic into one Paimon database.
+
+To use this feature through `flink run`, run the following shell command.
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ -c org.apache.paimon.flink.action.FlinkActions \
+ /path/to/paimon-flink-**-{{< version >}}.jar \
+ kafka-sync-database
+ --warehouse <warehouse-path> \
+ --database <database-name> \
+ [--ignore-incompatible <true/false>] \
+ [--table-prefix <paimon-table-prefix>] \
+ [--table-suffix <paimon-table-suffix>] \
+ [--kafka-conf <kafka-source-conf> [--kafka-conf <kafka-source-conf> ...]] \
+ [--catalog-conf <paimon-catalog-conf> [--catalog-conf <paimon-catalog-conf> ...]] \
+ [--table-conf <paimon-table-sink-conf> [--table-conf <paimon-table-sink-conf> ...]]
+```
+
+* `--warehouse` is the path to Paimon warehouse.
+* `--database` is the database name in Paimon catalog.
+* `--ignore-incompatible` is default false, in this case, if Kafka topic's table name exists in Paimon and their schema is incompatible,
+ an exception will be thrown. You can specify it to true explicitly to ignore the incompatible tables and exception.
+* `--table-prefix` is the prefix of all Paimon tables to be synchronized. For example, if you want all synchronized tables
+ to have "ods_" as prefix, you can specify `--table-prefix ods_`.
+* `--table-suffix` is the suffix of all Paimon tables to be synchronized. The usage is same as `--table-prefix`.
+* `--kafka-conf` is the configuration for Flink Kafka sources. Each configuration should be specified in the format `key=value`. `properties.bootstrap.servers`, `topic`, `properties.group.id`, and `value.format` are required configurations, others are optional.Note `topic` must contain multiple topics. See its [document](https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/kafka/#%e8%bf%9e%e6%8e%a5%e5%99%a8%e5%8f%82%e6%95%b0) for a complete list of configurations.
+* `--catalog-conf` is the configuration for Paimon catalog. Each configuration should be specified in the format `key=value`. See [here]({{< ref "maintenance/configurations" >}}) for a complete list of catalog configurations.
+* `--table-conf` is the configuration for Paimon table sink. Each configuration should be specified in the format `key=value`. All Paimon sink table will be applied the same set of configurations. See [here]({{< ref "maintenance/configurations" >}}) for a complete list of table configurations.
+
+Only tables with primary keys will be synchronized.
+
+For each Kafka topic's table to be synchronized, if the corresponding Paimon table does not exist, this action will automatically create the table. Its schema will be derived from all specified Kafka topic's tables. If the Paimon table already exists, its schema will be compared against the schema of all specified Kafka topic's tables.
Review Comment:
> Its schema will be derived from all specified Kafka topic's tables.
How to infer schema from topic, we can document this.
##########
paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/cdc/kafka/canal/CanalJsonEventParser.java:
##########
@@ -0,0 +1,346 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.flink.action.cdc.kafka.canal;
+
+import org.apache.paimon.flink.action.cdc.ComputedColumn;
+import org.apache.paimon.flink.action.cdc.TableNameConverter;
+import org.apache.paimon.flink.action.cdc.kafka.TypeUtil;
+import org.apache.paimon.flink.action.cdc.mysql.MySqlTypeUtils;
+import org.apache.paimon.flink.sink.cdc.CdcRecord;
+import org.apache.paimon.flink.sink.cdc.EventParser;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.utils.StringUtils;
+
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.ArrayNode;
+import org.apache.paimon.shade.jackson2.com.fasterxml.jackson.databind.node.NullNode;
+
+import com.alibaba.druid.DbType;
+import com.alibaba.druid.sql.SQLUtils;
+import com.alibaba.druid.sql.ast.SQLName;
+import com.alibaba.druid.sql.ast.SQLStatement;
+import com.alibaba.druid.sql.ast.statement.SQLAlterTableAddColumn;
+import com.alibaba.druid.sql.ast.statement.SQLAlterTableDropColumnItem;
+import com.alibaba.druid.sql.ast.statement.SQLAlterTableItem;
+import com.alibaba.druid.sql.ast.statement.SQLAlterTableStatement;
+import com.alibaba.druid.sql.ast.statement.SQLColumnDefinition;
+import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlAlterTableChangeColumn;
+import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlAlterTableModifyColumn;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+import static org.apache.paimon.utils.Preconditions.checkArgument;
+
+/** {@link EventParser} for Canal-json. */
+public class CanalJsonEventParser implements EventParser<String> {
Review Comment:
Can we have test for this class? Maybe you can create a separate PR for this to accelerate review.
##########
docs/content/how-to/cdc-ingestion.md:
##########
@@ -157,3 +157,187 @@ Example
--table-conf changelog-producer=input \
--table-conf sink.parallelism=4
```
+
+## Kafka
+
+### Prepare Kafka Bundled Jar
+
+```
+flink-sql-connector-kafka-*.jar
+```
+### Supported Formats
+Flink provides several Kafka CDC formats :canal-json、debezium-json,ogg-json,maxwell-json.
+If a message in a Kafka topic is a change event captured from another database using the Change Data Capture (CDC) tool, then you can use the Paimon Kafka CDC. Write the INSERT, UPDATE, DELETE messages parsed into the paimon table.
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left">Formats</th>
+ <th class="text-left">Supported</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/canal/">Canal CDC</a></td>
+ <td>True</td>
+ </tr>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/debezium/">Debezium CDC</a></td>
+ <td>False</td>
+ </tr>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/maxwell/ >}}">Maxwell CDC</a></td>
+ <td>False</td>
+ </tr>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/ogg/">OGG CDC</a></td>
+ <td>False</td>
+ </tr>
+ </tbody>
+</table>
+
+### Synchronizing Tables
+
+By using [KafkaSyncTableAction](/docs/{{< param Branch >}}/api/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction) in a Flink DataStream job or directly through `flink run`, users can synchronize one or multiple tables from Kafka's one topic into one Paimon table.
+
+To use this feature through `flink run`, run the following shell command.
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ -c org.apache.paimon.flink.action.FlinkActions \
+ /path/to/paimon-flink-**-{{< version >}}.jar \
+ kafka-sync-table
+ --warehouse <warehouse-path> \
+ --database <database-name> \
+ --table <table-name> \
+ [--partition-keys <partition-keys>] \
+ [--primary-keys <primary-keys>] \
+ [--computed-column <'column-name=expr-name(args[, ...])'> [--computed-column ...]] \
+ [--kafka-conf <kafka-source-conf> [--kafka-conf <kafka-source-conf> ...]] \
+ [--catalog-conf <paimon-catalog-conf> [--catalog-conf <paimon-catalog-conf> ...]] \
+ [--table-conf <paimon-table-sink-conf> [--table-conf <paimon-table-sink-conf> ...]]
+```
+
+* `--warehouse` is the path to Paimon warehouse.
+* `--database` is the database name in Paimon catalog.
+* `--table` is the Paimon table name.
+* `--partition-keys` are the partition keys for Paimon table. If there are multiple partition keys, connect them with comma, for example `dt,hh,mm`.
+* `--primary-keys` are the primary keys for Paimon table. If there are multiple primary keys, connect them with comma, for example `buyer_id,seller_id`.
+* `--computed-column` are the definitions of computed columns. The argument field is from Kafka topic's table field name. Supported expressions are:
+ * year(date-column): Extract year from a DATE, DATETIME or TIMESTAMP. Output is an INT value represent the year.
+ * substring(column,beginInclusive): Get column.substring(beginInclusive). Output is a STRING.
+ * substring(column,beginInclusive,endExclusive): Get column.substring(beginInclusive,endExclusive). Output is a STRING.
+ * truncate(column,width): truncate column by width. Output type is same with column.
+ * If the column is a STRING, truncate(column,width) will truncate the string to width characters, namely `value.substring(0, width)`.
+ * If the column is an INT or LONG, truncate(column,width) will truncate the number with the algorithm `v - (((v % W) + W) % W)`. The `redundant` compute part is to keep the result always positive.
+ * If the column is a DECIMAL, truncate(column,width) will truncate the decimal with the algorithm: let `scaled_W = decimal(W, scale(v))`, then return `v - (v % scaled_W)`.
+* `--kafka-conf` is the configuration for Flink Kafka sources. Each configuration should be specified in the format `key=value`. `properties.bootstrap.servers`, `topic`, `properties.group.id`, and `value.format` are required configurations, others are optional. See its [document](https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/kafka/#%e8%bf%9e%e6%8e%a5%e5%99%a8%e5%8f%82%e6%95%b0) for a complete list of configurations.
+* `--catalog-conf` is the configuration for Paimon catalog. Each configuration should be specified in the format `key=value`. See [here]({{< ref "maintenance/configurations" >}}) for a complete list of catalog configurations.
+* `--table-conf` is the configuration for Paimon table sink. Each configuration should be specified in the format `key=value`. See [here]({{< ref "maintenance/configurations" >}}) for a complete list of table configurations.
+
+If the Paimon table you specify does not exist, this action will automatically create the table. Its schema will be derived from all specified Kafka topic's tables. If the Paimon table already exists, its schema will be compared against the schema of all specified Kafka topic's tables.
+
+This action supports a limited number of schema changes. Currently, the framework can not drop columns, so the behaviors of `DROP` will be ignored, `RENAME` will add a new column. Currently supported schema changes includes:
+
+* Adding columns.
+
+* Altering column types. More specifically,
+
+ * altering from a string type (char, varchar, text) to another string type with longer length,
+ * altering from a binary type (binary, varbinary, blob) to another binary type with longer length,
+ * altering from an integer type (tinyint, smallint, int, bigint) to another integer type with wider range,
+ * altering from a floating-point type (float, double) to another floating-point type with wider range,
+
+ are supported.
+
+Example
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ -c org.apache.paimon.flink.action.FlinkActions \
+ /path/to/paimon-flink-**-{{< version >}}.jar \
+ kafka-sync-table \
+ --warehouse hdfs:///path/to/warehouse \
+ --database test_db \
+ --table test_table \
+ --partition-keys pt \
+ --primary-keys pt,uid \
+ --computed-columns '_year=year(age)' \
+ --kafka-conf properties.bootstrap.servers=127.0.0.1:9020 \
+ --kafka-conf topic=order \
+ --kafka-conf properties.group.id=123456 \
+ --kafka-conf value.format=canal-json \
+ --catalog-conf metastore=hive \
+ --catalog-conf uri=thrift://hive-metastore:9083 \
+ --table-conf bucket=4 \
+ --table-conf changelog-producer=input \
+ --table-conf sink.parallelism=4
+```
+
+### Synchronizing Databases
+
+By using [KafkaSyncDatabaseAction](/docs/{{< param Branch >}}/api/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction) in a Flink DataStream job or directly through `flink run`, users can synchronize the multi kafka topic into one Paimon database.
+
+To use this feature through `flink run`, run the following shell command.
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ -c org.apache.paimon.flink.action.FlinkActions \
+ /path/to/paimon-flink-**-{{< version >}}.jar \
+ kafka-sync-database
+ --warehouse <warehouse-path> \
+ --database <database-name> \
+ [--ignore-incompatible <true/false>] \
+ [--table-prefix <paimon-table-prefix>] \
+ [--table-suffix <paimon-table-suffix>] \
+ [--kafka-conf <kafka-source-conf> [--kafka-conf <kafka-source-conf> ...]] \
+ [--catalog-conf <paimon-catalog-conf> [--catalog-conf <paimon-catalog-conf> ...]] \
+ [--table-conf <paimon-table-sink-conf> [--table-conf <paimon-table-sink-conf> ...]]
+```
+
+* `--warehouse` is the path to Paimon warehouse.
+* `--database` is the database name in Paimon catalog.
+* `--ignore-incompatible` is default false, in this case, if Kafka topic's table name exists in Paimon and their schema is incompatible,
+ an exception will be thrown. You can specify it to true explicitly to ignore the incompatible tables and exception.
+* `--table-prefix` is the prefix of all Paimon tables to be synchronized. For example, if you want all synchronized tables
+ to have "ods_" as prefix, you can specify `--table-prefix ods_`.
+* `--table-suffix` is the suffix of all Paimon tables to be synchronized. The usage is same as `--table-prefix`.
+* `--kafka-conf` is the configuration for Flink Kafka sources. Each configuration should be specified in the format `key=value`. `properties.bootstrap.servers`, `topic`, `properties.group.id`, and `value.format` are required configurations, others are optional.Note `topic` must contain multiple topics. See its [document](https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/kafka/#%e8%bf%9e%e6%8e%a5%e5%99%a8%e5%8f%82%e6%95%b0) for a complete list of configurations.
+* `--catalog-conf` is the configuration for Paimon catalog. Each configuration should be specified in the format `key=value`. See [here]({{< ref "maintenance/configurations" >}}) for a complete list of catalog configurations.
+* `--table-conf` is the configuration for Paimon table sink. Each configuration should be specified in the format `key=value`. All Paimon sink table will be applied the same set of configurations. See [here]({{< ref "maintenance/configurations" >}}) for a complete list of table configurations.
+
+Only tables with primary keys will be synchronized.
+
+For each Kafka topic's table to be synchronized, if the corresponding Paimon table does not exist, this action will automatically create the table. Its schema will be derived from all specified Kafka topic's tables. If the Paimon table already exists, its schema will be compared against the schema of all specified Kafka topic's tables.
+
Review Comment:
Do not support one topic multiple tables? What happen in this case? Document this.
##########
docs/content/how-to/cdc-ingestion.md:
##########
@@ -157,3 +157,187 @@ Example
--table-conf changelog-producer=input \
--table-conf sink.parallelism=4
```
+
+## Kafka
+
+### Prepare Kafka Bundled Jar
+
+```
+flink-sql-connector-kafka-*.jar
+```
+### Supported Formats
+Flink provides several Kafka CDC formats :canal-json、debezium-json,ogg-json,maxwell-json.
+If a message in a Kafka topic is a change event captured from another database using the Change Data Capture (CDC) tool, then you can use the Paimon Kafka CDC. Write the INSERT, UPDATE, DELETE messages parsed into the paimon table.
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left">Formats</th>
+ <th class="text-left">Supported</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/canal/">Canal CDC</a></td>
+ <td>True</td>
+ </tr>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/debezium/">Debezium CDC</a></td>
+ <td>False</td>
+ </tr>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/maxwell/ >}}">Maxwell CDC</a></td>
+ <td>False</td>
+ </tr>
+ <tr>
+ <td><a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/zh/docs/connectors/table/formats/ogg/">OGG CDC</a></td>
+ <td>False</td>
+ </tr>
+ </tbody>
+</table>
+
+### Synchronizing Tables
+
+By using [KafkaSyncTableAction](/docs/{{< param Branch >}}/api/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction) in a Flink DataStream job or directly through `flink run`, users can synchronize one or multiple tables from Kafka's one topic into one Paimon table.
+
+To use this feature through `flink run`, run the following shell command.
+
+```bash
+<FLINK_HOME>/bin/flink run \
+ -c org.apache.paimon.flink.action.FlinkActions \
+ /path/to/paimon-flink-**-{{< version >}}.jar \
+ kafka-sync-table
+ --warehouse <warehouse-path> \
+ --database <database-name> \
+ --table <table-name> \
+ [--partition-keys <partition-keys>] \
+ [--primary-keys <primary-keys>] \
+ [--computed-column <'column-name=expr-name(args[, ...])'> [--computed-column ...]] \
+ [--kafka-conf <kafka-source-conf> [--kafka-conf <kafka-source-conf> ...]] \
+ [--catalog-conf <paimon-catalog-conf> [--catalog-conf <paimon-catalog-conf> ...]] \
+ [--table-conf <paimon-table-sink-conf> [--table-conf <paimon-table-sink-conf> ...]]
+```
+
+* `--warehouse` is the path to Paimon warehouse.
+* `--database` is the database name in Paimon catalog.
+* `--table` is the Paimon table name.
+* `--partition-keys` are the partition keys for Paimon table. If there are multiple partition keys, connect them with comma, for example `dt,hh,mm`.
+* `--primary-keys` are the primary keys for Paimon table. If there are multiple primary keys, connect them with comma, for example `buyer_id,seller_id`.
+* `--computed-column` are the definitions of computed columns. The argument field is from Kafka topic's table field name. Supported expressions are:
+ * year(date-column): Extract year from a DATE, DATETIME or TIMESTAMP. Output is an INT value represent the year.
Review Comment:
We can have a separate chapter to describe `omputed-column`. For example, a `computed-column` in the last.
```
## Computed Functions
.......
```
--
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@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org