You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2020/11/23 03:22:52 UTC
[flink] branch master updated: [FLINK-20150][debezium-avro] Add
documentation for the debezium-avro-confluent format
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 0c962c3 [FLINK-20150][debezium-avro] Add documentation for the debezium-avro-confluent format
0c962c3 is described below
commit 0c962c3f148641c67f8e8b5e73e86c131137919f
Author: Jark Wu <ja...@apache.org>
AuthorDate: Mon Nov 23 11:18:18 2020 +0800
[FLINK-20150][debezium-avro] Add documentation for the debezium-avro-confluent format
This closes #14151
---
docs/_data/sql-connectors.yml | 8 ++-
docs/dev/table/connectors/formats/debezium.md | 77 +++++++++++++++++++----
docs/dev/table/connectors/formats/debezium.zh.md | 78 ++++++++++++++++++++----
docs/dev/table/connectors/kinesis.md | 5 --
4 files changed, 140 insertions(+), 28 deletions(-)
diff --git a/docs/_data/sql-connectors.yml b/docs/_data/sql-connectors.yml
index 57e0c73..2683cda 100644
--- a/docs/_data/sql-connectors.yml
+++ b/docs/_data/sql-connectors.yml
@@ -60,7 +60,13 @@ parquet:
category: format
sql-url: https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-parquet{{site.scala_version_suffix}}/{{site.version}}/flink-sql-parquet{{site.scala_version_suffix}}-{{site.version}}.jar
-debezium:
+debezium-avro-confluent:
+ name: Debezium
+ maven: flink-avro-confluent-registry
+ category: format
+ sql-url: https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-avro-confluent-registry/{{site.version}}/flink-sql-avro-confluent-registry-{{site.version}}.jar
+
+debezium-json:
name: Debezium
maven: flink-json
category: format
diff --git a/docs/dev/table/connectors/formats/debezium.md b/docs/dev/table/connectors/formats/debezium.md
index 8c1c646..42b8e3f 100644
--- a/docs/dev/table/connectors/formats/debezium.md
+++ b/docs/dev/table/connectors/formats/debezium.md
@@ -32,32 +32,40 @@ under the License.
[Debezium](https://debezium.io/) is a CDC (Changelog Data Capture) tool that can stream changes in real-time from MySQL, PostgreSQL, Oracle, Microsoft SQL Server and many other databases into Kafka. Debezium provides a unified format schema for changelog and supports to serialize messages using JSON and [Apache Avro](https://avro.apache.org/).
-Flink supports to interpret Debezium JSON messages as INSERT/UPDATE/DELETE messages into Flink SQL system. This is useful in many cases to leverage this feature, such as
+Flink supports to interpret Debezium JSON and Avro messages as INSERT/UPDATE/DELETE messages into Flink SQL system. This is useful in many cases to leverage this feature, such as
- synchronizing incremental data from databases to other systems
- auditing logs
- real-time materialized views on databases
- temporal join changing history of a database table and so on.
-Flink also supports to encode the INSERT/UPDATE/DELETE messages in Flink SQL as Debezium JSON messages, and emit to storage like Kafka.
+Flink also supports to encode the INSERT/UPDATE/DELETE messages in Flink SQL as Debezium JSON or Avro messages, and emit to external systems like Kafka.
However, currently Flink can't combine UPDATE_BEFORE and UPDATE_AFTER into a single UPDATE message. Therefore, Flink encodes UPDATE_BEFORE and UDPATE_AFTER as DELETE and INSERT Debezium messages.
-*Note: Support for interpreting Debezium Avro messages and emitting Debezium messages is on the roadmap.*
-
Dependencies
------------
-{% assign connector = site.data.sql-connectors['debezium'] %}
+<div class="codetabs" markdown="1">
+<div data-lang="Debezium Avro" markdown="1">
+{% assign connector = site.data.sql-connectors['debezium-avro-confluent'] %}
{% include sql-connector-download-table.html
connector=connector
%}
+</div>
+<div data-lang="Debezium Json" markdown="1">
+{% assign connector = site.data.sql-connectors['debezium-json'] %}
+{% include sql-connector-download-table.html
+ connector=connector
+%}
+</div>
+</div>
-*Note: please refer to [Debezium documentation](https://debezium.io/documentation/reference/1.1/index.html) about how to setup a Debezium Kafka Connect to synchronize changelog to Kafka topics.*
+*Note: please refer to [Debezium documentation](https://debezium.io/documentation/reference/1.3/index.html) about how to setup a Debezium Kafka Connect to synchronize changelog to Kafka topics.*
How to use Debezium format
----------------
-Debezium provides a unified format for changelog, here is a simple example for an update operation captured from a MySQL `products` table:
+Debezium provides a unified format for changelog, here is a simple example for an update operation captured from a MySQL `products` table in JSON format:
```json
{
@@ -80,7 +88,7 @@ Debezium provides a unified format for changelog, here is a simple example for a
}
```
-*Note: please refer to [Debezium documentation](https://debezium.io/documentation/reference/1.1/connectors/mysql.html#mysql-connector-events_debezium) about the meaning of each fields.*
+*Note: please refer to [Debezium documentation](https://debezium.io/documentation/reference/1.3/connectors/mysql.html#mysql-connector-events_debezium) about the meaning of each fields.*
The MySQL `products` table has 4 columns (`id`, `name`, `description` and `weight`). The above JSON message is an update change event on the `products` table where the `weight` value of the row with `id = 111` is changed from `5.18` to `5.15`.
Assuming this messages is synchronized to Kafka topic `products_binlog`, then we can use the following DDL to consume this topic and interpret the change events.
@@ -99,7 +107,9 @@ CREATE TABLE topic_products (
'topic' = 'products_binlog',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.group.id' = 'testGroup',
- 'format' = 'debezium-json' -- using debezium-json as the format
+ -- using 'debezium-json' as the format to interpret Debezium JSON messages
+ -- please use 'debezium-avro-confluent' if Debezium encodes messages in Avro format
+ 'format' = 'debezium-json'
)
{% endhighlight %}
</div>
@@ -154,6 +164,50 @@ SELECT * FROM topic_products;
Format Options
----------------
+Flink provides `debezium-avro-confluent` and `debezium-json` formats to interpret Avro or Json messages produced by Debezium.
+Use format `debezium-avro-confluent` to interpret Debezium Avro messages and format `debezium-json` to interpret Debezium Json messages.
+
+<div class="codetabs" markdown="1">
+<div data-lang="Debezium Avro" markdown="1">
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 25%">Option</th>
+ <th class="text-center" style="width: 8%">Required</th>
+ <th class="text-center" style="width: 7%">Default</th>
+ <th class="text-center" style="width: 10%">Type</th>
+ <th class="text-center" style="width: 50%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>format</h5></td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Specify what format to use, here should be <code>'debezium-avro-confluent'</code>.</td>
+ </tr>
+ <tr>
+ <td><h5>debezium-avro-confluent.schema-registry.url</h5></td>
+ <td>required</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The URL of the Confluent Schema Registry to fetch/register schemas.</td>
+ </tr>
+ <tr>
+ <td><h5>debezium-avro-confluent.schema-registry.subject</h5></td>
+ <td>required by sink</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>The Confluent Schema Registry subject under which to register the schema used by this format during serialization.</td>
+ </tr>
+ </tbody>
+</table>
+
+</div>
+<div data-lang="Debezium Json" markdown="1">
+
<table class="table table-bordered">
<thead>
<tr>
@@ -220,10 +274,11 @@ Format Options
<td>String</td>
<td>Specify string literal to replace null key when <code>'debezium-json.map-null-key.mode'</code> is LITERAL.</td>
</tr>
- <tr>
</tbody>
</table>
+</div>
+
Caveats
----------------
@@ -239,5 +294,5 @@ See more details in [Debezium Documentation for PostgreSQL REPLICA IDENTITY](htt
Data Type Mapping
----------------
-Currently, the Debezium format uses JSON format for serialization and deserialization. Please refer to [JSON format documentation]({% link dev/table/connectors/formats/json.md %}#data-type-mapping) for more details about the data type mapping.
+Currently, the Debezium format uses JSON and Avro format for serialization and deserialization. Please refer to [JSON Format documentation]({% link dev/table/connectors/formats/json.md %}#data-type-mapping) and [Confluent Avro Format documentation]({% link dev/table/connectors/formats/avro-confluent.md %}#data-type-mapping) for more details about the data type mapping.
diff --git a/docs/dev/table/connectors/formats/debezium.zh.md b/docs/dev/table/connectors/formats/debezium.zh.md
index b04f567..3558cf1 100644
--- a/docs/dev/table/connectors/formats/debezium.zh.md
+++ b/docs/dev/table/connectors/formats/debezium.zh.md
@@ -32,33 +32,41 @@ under the License.
[Debezium](https://debezium.io/) 是一个 CDC（Changelog Data Capture，变更数据捕获）的工具，可以把来自 MySQL、PostgreSQL、Oracle、Microsoft SQL Server 和许多其他数据库的更改实时流式传输到 Kafka 中。 Debezium 为变更日志提供了统一的格式结构，并支持使用 JSON 和 Apache Avro 序列化消息。
-Flink 支持将 Debezium JSON 消息解析为 INSERT / UPDATE / DELETE 消息到 Flink SQL 系统中。在很多情况下，利用这个特性非常的有用，例如
+Flink 支持将 Debezium JSON 和 Avro 消息解析为 INSERT / UPDATE / DELETE 消息到 Flink SQL 系统中。在很多情况下，利用这个特性非常的有用，例如
- 将增量数据从数据库同步到其他系统
- 日志审计
- 数据库的实时物化视图
- 关联维度数据库的变更历史，等等。
-Flink 还支持将 Flink SQL 中的 INSERT / UPDATE / DELETE 消息编码为 Debezium 格式的 JSON 消息，输出到 Kafka 等存储中。
+Flink 还支持将 Flink SQL 中的 INSERT / UPDATE / DELETE 消息编码为 Debezium 格式的 JSON 或 Avro 消息，输出到 Kafka 等存储中。
但需要注意的是，目前 Flink 还不支持将 UPDATE_BEFORE 和 UPDATE_AFTER 合并为一条 UPDATE 消息。因此，Flink 将 UPDATE_BEFORE 和 UPDATE_AFTER 分别编码为 DELETE 和 INSERT 类型的 Debezium 消息。
-*注意: 支持解析 Debezium Avro 消息和输出 Debezium 消息已经规划在路线图上了。*
-
依赖
------------
-{% assign connector = site.data.sql-connectors['debezium'] %}
-{% include sql-connector-download-table.html
+<div class="codetabs" markdown="1">
+<div data-lang="Debezium Avro" markdown="1">
+{% assign connector = site.data.sql-connectors['debezium-avro-confluent'] %}
+{% include sql-connector-download-table.html
+ connector=connector
+%}
+</div>
+<div data-lang="Debezium Json" markdown="1">
+{% assign connector = site.data.sql-connectors['debezium-json'] %}
+{% include sql-connector-download-table.html
connector=connector
%}
+</div>
+</div>
-*注意: 请参考 [Debezium 文档](https://debezium.io/documentation/reference/1.1/index.html)，了解如何设置 Debezium Kafka Connect 用来将变更日志同步到 Kafka 主题。*
+*注意: 请参考 [Debezium 文档](https://debezium.io/documentation/reference/1.3/index.html)，了解如何设置 Debezium Kafka Connect 用来将变更日志同步到 Kafka 主题。*
如何使用 Debezium Format
----------------
-Debezium 为变更日志提供了统一的格式，这是一个从 MySQL product 表捕获的更新操作的简单示例:
+Debezium 为变更日志提供了统一的格式，这是一个 JSON 格式的从 MySQL product 表捕获的更新操作的简单示例:
```json
{
@@ -81,7 +89,7 @@ Debezium 为变更日志提供了统一的格式，这是一个从 MySQL product
}
```
-*注意: 请参考 [Debezium 文档](https://debezium.io/documentation/reference/1.1/connectors/mysql.html#mysql-connector-events_debezium)，了解每个字段的含义。*
+*注意: 请参考 [Debezium 文档](https://debezium.io/documentation/reference/1.3/connectors/mysql.html#mysql-connector-events_debezium)，了解每个字段的含义。*
MySQL 产品表有4列（`id`、`name`、`description`、`weight`）。上面的 JSON 消息是 `products` 表上的一条更新事件，其中 `id = 111` 的行的 `weight` 值从 `5.18` 更改为 `5.15`。假设此消息已同步到 Kafka 主题 `products_binlog`，则可以使用以下 DDL 来使用此主题并解析更改事件。
@@ -99,7 +107,9 @@ CREATE TABLE topic_products (
'topic' = 'products_binlog',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.group.id' = 'testGroup',
- 'format' = 'debezium-json' -- 使用 debezium-json 作为 format
+ -- 使用 'debezium-json' format 来解析 Debezium 的 JSON 消息
+ -- 如果 Debezium 用 Avro 编码消息，请使用 'debezium-avro-confluent'
+ 'format' = 'debezium-json' -- 如果 Debezium 用 Avro 编码消息，请使用 'debezium-avro-confluent'
)
{% endhighlight %}
</div>
@@ -154,6 +164,50 @@ SELECT * FROM topic_products;
Format 参数
----------------
+Flink 提供了 `debezium-avro-confluent` 和 `debezium-json` 两种 format 来解析 Debezium 生成的 JSON 格式和 Avro 格式的消息。
+请使用 `debezium-avro-confluent` 来解析 Debezium 的 Avro 消息，使用 `debezium-json` 来解析 Debezium 的 JSON 消息。
+
+<div class="codetabs" markdown="1">
+<div data-lang="Debezium Avro" markdown="1">
+
+<table class="table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 25%">参数</th>
+ <th class="text-center" style="width: 10%">是否必选</th>
+ <th class="text-center" style="width: 10%">默认值</th>
+ <th class="text-center" style="width: 10%">类型</th>
+ <th class="text-center" style="width: 45%">描述</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>format</h5></td>
+ <td>必选</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>指定要使用的格式，此处应为 <code>'debezium-avro-confluent'</code>。</td>
+ </tr>
+ <tr>
+ <td><h5>debezium-avro-confluent.schema-registry.url</h5></td>
+ <td>必选</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>用于获取/注册 schemas 的 Confluent Schema Registry 的 URL。</td>
+ </tr>
+ <tr>
+ <td><h5>debezium-avro-confluent.schema-registry.subject</h5></td>
+ <td>sink 必选</td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>String</td>
+ <td>Confluent Schema Registry主题，用于在序列化期间注册此格式使用的 schema。</td>
+ </tr>
+ </tbody>
+</table>
+
+</div>
+<div data-lang="Debezium Json" markdown="1">
+
<table class="table table-bordered">
<thead>
<tr>
@@ -221,6 +275,8 @@ Format 参数
</tbody>
</table>
+</div>
+
注意事项
----------------
@@ -236,5 +292,5 @@ Format 参数
数据类型映射
----------------
-目前，Debezium Format 使用 JSON Format 进行序列化和反序列化。有关数据类型映射的更多详细信息，请参考 [JSON Format 文档]({% link dev/table/connectors/formats/json.zh.md %}#data-type-mapping)。
+目前，Debezium Format 使用 JSON Format 进行序列化和反序列化。有关数据类型映射的更多详细信息，请参考 [JSON Format 文档]({% link dev/table/connectors/formats/json.zh.md %}#data-type-mapping) 和 [Confluent Avro Format 文档]({% link dev/table/connectors/formats/avro-confluent.md %}#data-type-mapping)。
diff --git a/docs/dev/table/connectors/kinesis.md b/docs/dev/table/connectors/kinesis.md
index 85dc0a3..353654f 100644
--- a/docs/dev/table/connectors/kinesis.md
+++ b/docs/dev/table/connectors/kinesis.md
@@ -39,11 +39,6 @@ Dependencies
connector=connector
%}
-{% assign connector = site.data.sql-connectors['kinesis'] %}
-{% include sql-connector-download-table.html
- connector=connector
-%}
-
How to create a Kinesis data stream table
-----------------------------------------