This is an automated email from the ASF dual-hosted git repository.
zirui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/inlong-website.git
The following commit(s) were added to refs/heads/master by this push:
new 61143f35ad [INLONG-591][Doc] Add document for multiple sink of iceberg
(#592)
61143f35ad is described below
commit 61143f35ad2db13ada07a905ac88cee50d90e32b
Author: thesumery <[email protected]>
AuthorDate: Thu Nov 17 14:52:14 2022 +0800
[INLONG-591][Doc] Add document for multiple sink of iceberg (#592)
---
docs/data_node/load_node/iceberg.md | 101 +++++++++++++++++++++
.../current/data_node/load_node/iceberg.md | 97 ++++++++++++++++++++
2 files changed, 198 insertions(+)
diff --git a/docs/data_node/load_node/iceberg.md
b/docs/data_node/load_node/iceberg.md
index 4b15e24abd..a1ade23add 100644
--- a/docs/data_node/load_node/iceberg.md
+++ b/docs/data_node/load_node/iceberg.md
@@ -147,6 +147,103 @@ TODO
### Usage for InLong Manager Client
TODO
+## Feature
+### Multiple table sink
+Currently Iceberg support multiple table sinking, it require FLINK SQL create
table parameters add
+`'sink.multiple.enable' = 'true'` and target table schema can only be defined
as `BYTES` or `STRING`
+Examples as follows:
+```
+CREATE TABLE `table_2`(
+ `data` STRING)
+WITH (
+ 'connector'='iceberg-inlong',
+ 'catalog-name'='hive_prod',
+ 'uri'='thrift://localhost:9083',
+ 'warehouse'='hdfs://localhost:8020/hive/warehouse',
+ 'sink.multiple.enable' = 'true',
+ 'sink.multiple.format' = 'canal-json',
+ 'sink.multiple.add-column.policy' = 'TRY_IT_BEST',
+ 'sink.multiple.database-pattern' = '${database}',
+ 'sink.multiple.table-pattern' = 'test_${table}'
+);
+```
+To support multiple sink, it is necessary to set the serialization format of
upstream data
+(Via option 'sink.multiple.format' to set, currently only supports
[canal-json|debezium-json]).
+
+### dynamic dababase/table Extraction
+Iceberg can customize mapping rules for database names and table names, it can
fill in placeholders and add prefixes
+and suffixes to modify the mapped target table name. Iceberg Load Node will
extract `'sink.multiple.database-pattern'`
+as target database name, extract `'sink.multiple.table-pattern'` as target
table name,
+The placeholder is parsed from the data, the variable is strictly represented
by '${VARIABLE_NAME}',
+the value of the variable comes from the data itself, it can be a metadata
field of a Format specified by
+`'sink.multiple.format'`, or it can be a physical field in the data.
+Examples of 'topic-parttern' are as follows:
+- 'sink.multiple.format' is 'canal-json':
+
+The upstream data is:
+```
+{
+ "data": [
+ {
+ "id": "111",
+ "name": "scooter",
+ "description": "Big 2-wheel scooter",
+ "weight": "5.18"
+ }
+ ],
+ "database": "inventory",
+ "es": 1589373560000,
+ "id": 9,
+ "isDdl": false,
+ "mysqlType": {
+ "id": "INTEGER",
+ "name": "VARCHAR(255)",
+ "description": "VARCHAR(512)",
+ "weight": "FLOAT"
+ },
+ "old": [
+ {
+ "weight": "5.15"
+ }
+ ],
+ "pkNames": [
+ "id"
+ ],
+ "sql": "",
+ "sqlType": {
+ "id": 4,
+ "name": 12,
+ "description": 12,
+ "weight": 7
+ },
+ "table": "products",
+ "ts": 1589373560798,
+ "type": "UPDATE"
+}
+```
+'topic-pattern' is '{database}_${table}', and the extracted topic is
'inventory_products'
+('source.db', 'source.table' are metadata fields, and 'id' are physical fields)
+
+'topic-pattern' is '{database}_${table}_${id}', and the extracted topic is
'inventory_products_111'
+('source.db', 'source.table' are metadata fields, and 'id' are physical fields)
+
+
+### Auto create database/table
+Iceberg can auto create database and auto create table in multiple sink scenes
if database and table not exists, and it supports capture new table at runtime。
+default Iceberg table parameters: `'format-version' =
'2'`、`'write.upsert.enabled' = 'true''`、`'engine.hive.enabled' = 'true'`
+
+### Dynamic schema evolution
+Iceberg support schema evolution from source table to target table in multiple
sink scenes(DDL synchronize), supported schema evolution：
+
+| schema evolution type | supported |
+| -------------- | ----------- |
+| Column add | true |
+| Column delete | false |
+| Column reorder | false |
+| Column rename | false |
+| Column type update | false |
+
+
## Iceberg Load Node Options
| Option | Required | Default |
Type | Description |
@@ -163,6 +260,10 @@ TODO
| warehouse | optional for hadoop catalog or hive catalog | (none) |
String | For Hive catalog，is the Hive warehouse location, users should specify
this path if neither set the `hive-conf-dir` to specify a location containing a
`hive-site.xml` configuration file nor add a correct `hive-site.xml` to
classpath. For hadoop catalog，The HDFS directory to store metadata files and
data files. |
| hive-conf-dir | optional for hive catalog | (none) |
String | Path to a directory containing a `hive-site.xml` configuration file
which will be used to provide custom Hive configuration values. The value of
`hive.metastore.warehouse.dir` from `<hive-conf-dir>/hive-site.xml` (or hive
configure file from classpath) will be overwrote with the `warehouse` value if
setting both `hive-conf-dir` and `warehouse` when creating iceberg catalog. |
| inlong.metric.labels | optional | (none) | String | Inlong metric label,
format of value is groupId=xxgroup&streamId=xxstream&nodeId=xxnode. |
+| sink.multiple.enable | optional | false | Boolean |
Whether to enable multiple sink |
+| sink.multiple.schema-update.policy | optional | TRY_IT_BEST | Enum
| The policy to handle the inconsistency between the schema in the data and the
schema of the target table <br/>TRY_IT_BEST: try best, deal with as much as
possible, ignore it if can't handled.<br/> IGNORE_WITH_LOG:ignore it and log
it,ignore this table later.<br/> THROW_WITH_STOP:throw exception and stop the
job, until user deal with schema conflict and job restore.
+| sink.multiple.pk-auto-generated | optional | false | Boolean
| Whether auto generate primary key, regard all field combined as primary key
in multiple sink scenes. |
+| sink.multiple.typemap-compatible-with-spark | optional | false | Boolean
| Whether to adapt spark type system in auto generate table. |
## Data Type Mapping
diff --git
a/i18n/zh-CN/docusaurus-plugin-content-docs/current/data_node/load_node/iceberg.md
b/i18n/zh-CN/docusaurus-plugin-content-docs/current/data_node/load_node/iceberg.md
index 930605b239..5243e9534e 100644
---
a/i18n/zh-CN/docusaurus-plugin-content-docs/current/data_node/load_node/iceberg.md
+++
b/i18n/zh-CN/docusaurus-plugin-content-docs/current/data_node/load_node/iceberg.md
@@ -148,6 +148,98 @@ TODO
### InLong Manager Client 用法
TODO
+## 特征
+### 多表写入
+目前 Iceberg 支持多表同时写入，需要在 FLINK SQL 的建表参数上添加 `'sink.multiple.enable' = 'true'`
并且目标表的schema
+只能定义成 `BYTES` 或者 `STRING` ，以下是一个建表语句举例：
+```
+CREATE TABLE `table_2`(
+ `data` STRING)
+WITH (
+ 'connector'='iceberg-inlong',
+ 'catalog-name'='hive_prod',
+ 'uri'='thrift://localhost:9083',
+ 'warehouse'='hdfs://localhost:8020/hive/warehouse',
+ 'sink.multiple.enable' = 'true',
+ 'sink.multiple.format' = 'canal-json',
+ 'sink.multiple.add-column.policy' = 'TRY_IT_BEST',
+ 'sink.multiple.database-pattern' = '${database}',
+ 'sink.multiple.table-pattern' = 'test_${table}'
+);
+```
+要支持多表写入同时需要设置上游数据的序列化格式(通过选项 'sink.multiple.format'
+来设置, 目前仅支持 [canal-json|debezium-json])。
+
+### 动态表名映射
+Iceberg 在多表写入的时可以自定义映射的数据库名和表名的规则，可以填充占位符然后添加前后缀来修改映射的目标表名称。
+Iceberg Load Node 会解析 `'sink.multiple.database-pattern'` 作为目的端的 数据库名, 解析
`'sink.multiple.table-pattern'`
+作为目的端的表名，占位符是从数据中解析出来的，变量是严格通过 '${VARIABLE_NAME}' 来表示, 变量的取值来自于数据本身,
+即可以是通过 `'sink.multiple.format'` 指定的某种 Format 的元数据字段, 也可以是数据中的物理字段。
+关于 'topic-parttern' 的例子如下:
+- 'sink.multiple.format' 为 'canal-json':
+
+上游数据为:
+```
+{
+ "data": [
+ {
+ "id": "111",
+ "name": "scooter",
+ "description": "Big 2-wheel scooter",
+ "weight": "5.18"
+ }
+ ],
+ "database": "inventory",
+ "es": 1589373560000,
+ "id": 9,
+ "isDdl": false,
+ "mysqlType": {
+ "id": "INTEGER",
+ "name": "VARCHAR(255)",
+ "description": "VARCHAR(512)",
+ "weight": "FLOAT"
+ },
+ "old": [
+ {
+ "weight": "5.15"
+ }
+ ],
+ "pkNames": [
+ "id"
+ ],
+ "sql": "",
+ "sqlType": {
+ "id": 4,
+ "name": 12,
+ "description": 12,
+ "weight": 7
+ },
+ "table": "products",
+ "ts": 1589373560798,
+ "type": "UPDATE"
+}
+```
+'topic-pattern' 为 '{database}_${table}', 提取后的 Topic 为 'inventory_products'
('database', 'table' 为元数据字段,
+'id' 为物理字段)
+
+'topic-pattern' 为 '{database}_${table}_${id}', 提取后的 Topic 为
'inventory_products_111' ('database', 'table'
+为元数据字段, 'id' 为物理字段)
+
+### 动态建库、建表
+Iceberg在多表写入时遇到不存在的表和不存在的库时会自动创建数据库和数据表，并且支持在运行过程中新增捕获额外的表入库。
+默认的Iceberg表参数为：`'format-version' = '2'`、`'write.upsert.enabled' =
'true''`、`'engine.hive.enabled' = 'true'`
+
+### 动态schema变更
+Iceberg在多表写入时支持同步源表结构变更到目标表（DDL同步），支持的schema变更如下：
+
+| schema变更类型 | 是否支持 |
+| -------------- | ----------- |
+| 列增加 | 是 |
+| 列减少 | 否 |
+| 列位置变更 | 否 |
+| 列重命名 | 否 |
+| 列类型变更 | 否 |
+
## Iceberg Load 节点参数
| 选项 | 是否必须 | 默认值 | 类型 | 描述
|
@@ -164,6 +256,11 @@ TODO
| warehouse | hive catalog或hadoop catalog可选 | (none) | String | 对于
Hive 目录，是 Hive
仓库位置，如果既不设置`hive-conf-dir`指定包含`hive-site.xml`配置文件的位置也不添加正确`hive-site.xml`的类路径，用户应指定此路径。对于hadoop目录，HDFS目录存放元数据文件和数据文件
|
| hive-conf-dir | hive catalog可选 | (none) | String |
`hive-site.xml`包含将用于提供自定义 Hive
配置值的配置文件的目录的路径。如果同时设置和创建Iceberg目录时，`hive.metastore.warehouse.dir`from
`<hive-conf-dir>/hive-site.xml`（或来自类路径的 hive
配置文件）的值将被该值覆盖。`warehouse``hive-conf-dir``warehouse` |
| inlong.metric | 可选 | (none) | String | inlong metric
的标签值，该值的构成为groupId&streamId&nodeId。|
+| sink.multiple.enable | 可选 | false | Boolean |
是否开启多路写入 |
+| sink.multiple.schema-update.policy | 可选 | TRY_IT_BEST | Enum |
遇到数据中schema和目标表不一致时的处理策略<br/>TRY_IT_BEST：尽力而为，尽可能处理，处理不了的则忽略<br/>IGNORE_WITH_LOG：忽略并且记录日志，后续该表数据不再处理<br/>
THROW_WITH_STOP：抛异常并且停止任务，直到用户手动处理schema不一致的情况
+| sink.multiple.pk-auto-generated | 可选 | false | Boolean |
是否自动生成主键，对于多路写入自动建表时当源表无主键时是否将所有字段当作主键 |
+| sink.multiple.typemap-compatible-with-spark | 可选 | false | Boolean |
是否适配spark的类型系统，对于多路写入自动建表时是否需要适配spark的类型系统 |
+
## 数据类型映射
