Skip to content

[Feature][Format] Improve maxwell_json,canal_json,debezium_json format support merge update_before and update_after #9805

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 27 commits into
base: dev
Choose a base branch
from
Open

[Feature][Format] Improve maxwell_json,canal_json,debezium_json format support merge update_before and update_after #9805

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
6cb4abf
init
Aug 19, 2025
e7f0632
init
Aug 20, 2025
1e48dc4
init
Aug 20, 2025
0868921
init
Aug 20, 2025
25cbac9
init
Aug 20, 2025
41c06cf
init
Aug 20, 2025
f780aa6
init
Aug 20, 2025
02ec69b
init
dyp12 Aug 21, 2025
4574f83
Merge branch 'apache:dev' into dev
dyp12 Aug 23, 2025
24d6f1a
init
dyp12 Aug 25, 2025
51ad4e9
code format
dyp12 Aug 25, 2025
72199d7
Merge branch 'apache:dev' into dev
dyp12 Aug 28, 2025
979fb5e
init
Aug 29, 2025
4157583
init
Aug 29, 2025
f098cac
init
Aug 29, 2025
529dc6c
init
Aug 29, 2025
0e7ee4c
init
Aug 29, 2025
73bec44
init
Aug 29, 2025
b263e7a
init
Aug 30, 2025
bf3e6ec
Merge branch 'apache:dev' into dev
dyp12 Aug 30, 2025
660a21e
init
Aug 30, 2025
1add2d1
init
Aug 30, 2025
d4bfa25
init
Aug 31, 2025
16d2a33
feat:update doc
Sep 4, 2025
76647a5
feat:update
Sep 5, 2025
79e9efe
feat:update
Sep 5, 2025
5ec634d
Merge branch 'apache:dev' into dev
dyp12 Sep 5, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
83 changes: 46 additions & 37 deletions docs/en/connector-v2/sink/CosFile.md
View file
Open in desktop

Large diffs are not rendered by default.

9 changes: 9 additions & 0 deletions docs/en/connector-v2/sink/HdfsFile.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,9 @@ By default, we use 2PC commit to ensure `exactly-once`
- [x] binary
- [x] compress codec
- [x] lzo
- [x] canal_json
- [x] debezium_json
- [x] maxwell_json

## Description

Expand Down Expand Up @@ -85,11 +88,17 @@ Output data to hdfs file
| enable_header_write | boolean | no | false | Only used when file_format_type is text,csv.<br/> false:don't write header,true:write header. |
| encoding | string | no | "UTF-8" | Only used when file_format_type is json,text,csv,xml. |
| remote_user | string | no | - | The remote user name of hdfs. |
| merge_update_event | boolean | no | false | Only used when file_format_type is canal_json,debezium_json or maxwell_json. When value is true, the UPDATE_AFTER and UPDATE_BEFORE event will be merged into UPDATE event data |

### Tips

> If you use spark/flink, In order to use this connector, You must ensure your spark/flink cluster already integrated hadoop. The tested hadoop version is 2.x. If you use SeaTunnel Engine, It automatically integrated the hadoop jar when you download and install SeaTunnel Engine. You can check the jar package under ${SEATUNNEL_HOME}/lib to confirm this.

### merge_update_event [boolean]

Only used when file_format_type is canal_json,debezium_json or maxwell_json.
When value is true, the UPDATE_AFTER and UPDATE_BEFORE event will be merged into UPDATE event data

## Task Example

### Simple
Expand Down
81 changes: 45 additions & 36 deletions docs/en/connector-v2/sink/LocalFile.md
View file
Open in desktop

Large diffs are not rendered by default.

65 changes: 37 additions & 28 deletions docs/en/connector-v2/sink/ObsFile.md
View file
Open in desktop

Large diffs are not rendered by default.

89 changes: 49 additions & 40 deletions docs/en/connector-v2/sink/OssFile.md
View file
Open in desktop

Large diffs are not rendered by default.

84 changes: 47 additions & 37 deletions docs/en/connector-v2/sink/OssJindoFile.md
View file
Open in desktop

Large diffs are not rendered by default.

93 changes: 51 additions & 42 deletions docs/en/connector-v2/sink/S3File.md
View file
Open in desktop

Large diffs are not rendered by default.

89 changes: 49 additions & 40 deletions docs/en/connector-v2/sink/SftpFile.md
View file
Open in desktop

Large diffs are not rendered by default.

7 changes: 7 additions & 0 deletions docs/zh/connector-v2/sink/CosFile.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,7 @@ import ChangeLog from '../changelog/connector-file-cos.md';
| parquet_avro_write_timestamp_as_int96 | boolean | 否 | false | 仅在file_format为parquet时使用. |
| parquet_avro_write_fixed_as_int96 | array | 否 | - | 仅在file_format为parquet时使用. |
| encoding | string | 否 | "UTF-8" | 仅当file_format_type为json、text、csv、xml时使用. |
| merge_update_event | boolean | 否 | false | 仅当file_format_type为canal_json、debezium_json、maxwell_json. |

### path [string]

Expand Down Expand Up @@ -242,6 +243,12 @@ Tips: excel 类型不支持任何压缩格式
仅当file_format_type为json、text、csv、xml时使用.
要写入的文件的编码。此参数将由`Charset.forName(encoding)` 解析.

### merge_update_event [boolean]

仅当file_format_type为canal_json、debezium_json、maxwell_json时使用.
设置成true,序列化数据时,UPDATE_AFTER 和 UPDATE_BEFORE 会合并成 UPDATE;
设置成false,序列化数据时,UPDATE_AFTER 和 UPDATE_BEFORE 不会合并;

## 示例

对于具有 `have_partition` 、 `custom_filename` 和 `sink_columns` 的文本文件格式
Expand Down
11 changes: 9 additions & 2 deletions docs/zh/connector-v2/sink/HdfsFile.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,10 +55,10 @@ import ChangeLog from '../changelog/connector-file-hadoop.md';
| custom_filename | boolean | 否 | false | 是否需要自定义文件名 |
| file_name_expression | string | 否 | "${transactionId}" | 仅在 `custom_filename` 为 `true` 时使用。`file_name_expression` 描述将创建到 `path` 中的文件表达式。我们可以在 `file_name_expression` 中添加变量 `${now}` 或 `${uuid}`，例如 `test_${uuid}_${now}`，`${now}` 表示当前时间，其格式可以通过指定选项 `filename_time_format` 来定义。请注意，如果 `is_enable_transaction` 为 `true`，我们将在文件头部自动添加 `${transactionId}_`。 |
| filename_time_format | string | 否 | "yyyy.MM.dd" | 仅在 `custom_filename` 为 `true` 时使用。当 `file_name_expression` 参数中的格式为 `xxxx-${now}` 时，`filename_time_format` 可以指定路径的时间格式，默认值为 `yyyy.MM.dd`。常用的时间格式如下所示：[y:年,M:月,d:月中的一天,H:一天中的小时（0-23），m:小时中的分钟，s:分钟中的秒] |
| file_format_type | string | 否 | "csv" | 我们支持以下文件类型：`text` `json` `csv` `orc` `parquet` `excel` `canal_json` `debezium_json` `maxwell_json`。请注意，最终文件名将以文件格式的后缀结束，文本文件的后缀是 `txt`。 |
| file_format_type | string | 否 | "csv" | 我们支持以下文件类型：`text` `json` `csv` `orc` `parquet` `excel` `canal_json` `debezium_json` `maxwell_json`。请注意，最终文件名将以文件格式的后缀结束，文本文件的后缀是 `txt`。 |
| filename_extension | string | 否 | - | 使用自定义的文件扩展名覆盖默认的文件扩展名。 例如：`.xml`, `.json`, `dat`, `.customtype` |
| field_delimiter | string | 否 | '\001' | 仅在 file_format 为 text 时使用，数据行中列之间的分隔符。仅需要 `text` 文件格式。 |
| row_delimiter | string | 否 | "\n" | 仅在 file_format 为 text 时使用，文件中行之间的分隔符。仅需要 `text`、`csv`、`json` 文件格式。 |
| row_delimiter | string | 否 | "\n" | 仅在 file_format 为 text 时使用，文件中行之间的分隔符。仅需要 `text`、`csv`、`json` 文件格式。 |
| have_partition | boolean | 否 | false | 是否需要处理分区。 |
| partition_by | array | 否 | - | 仅在 have_partition 为 true 时使用，根据选定的字段对数据进行分区。 |
| partition_dir_expression | string | 否 | "${k0}=${v0}/${k1}=${v1}/.../${kn}=${vn}/" | 仅在 have_partition 为 true 时使用，如果指定了 `partition_by`，我们将根据分区信息生成相应的分区目录，并将最终文件放置在分区目录中。默认 `partition_dir_expression` 为 `${k0}=${v0}/${k1}=${v1}/.../${kn}=${vn}/`。`k0` 是第一个分区字段，`v0` 是第一个分区字段的值。 |
Expand All @@ -79,13 +79,20 @@ import ChangeLog from '../changelog/connector-file-hadoop.md';
| max_rows_in_memory | int | 否 | - | 仅当 file_format 为 excel 时使用。当文件格式为 Excel 时，可以缓存在内存中的最大数据项数。 |
| sheet_name | string | 否 | Sheet${Random number} | 仅当 file_format 为 excel 时使用。将工作簿的表写入指定的表名 |
| remote_user | string | 否 | - | Hdfs的远端用户名。 |
| merge_update_event | boolean | 否 | false | 仅当file_format_type为canal_json、debezium_json、maxwell_json. |

### 提示

> 如果您使用 spark/flink，为了使用此连接器，您必须确保您的 spark/flink 集群已经集成了 hadoop。测试过的 hadoop 版本是
> 2.x。如果您使用 SeaTunnel Engine，则在下载和安装 SeaTunnel Engine 时会自动集成 hadoop
> jar。您可以检查 `${SEATUNNEL_HOME}/lib` 下的 jar 包来确认这一点。

### merge_update_event [boolean]

仅当file_format_type为canal_json、debezium_json、maxwell_json时使用.
设置成true,序列化数据时,UPDATE_AFTER 和 UPDATE_BEFORE 会合并成 UPDATE;
设置成false,序列化数据时,UPDATE_AFTER 和 UPDATE_BEFORE 不会合并;

## 任务示例

### 简单示例
Expand Down
9 changes: 8 additions & 1 deletion docs/zh/connector-v2/sink/LocalFile.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ import ChangeLog from '../changelog/connector-file-local.md';
| file_format_type | string | 否 | "csv" | 文件格式类型 |
| filename_extension | string | 否 | - | 使用自定义的文件扩展名覆盖默认的文件扩展名。 例如：`.xml`, `.json`, `dat`, `.customtype` |
| field_delimiter | string | 否 | '\001' | 仅在 file_format_type 为 text 时使用 |
| row_delimiter | string | 否 | "\n" | 仅在 file_format_type 为 `text`、`csv`、`json` 时使用 |
| row_delimiter | string | 否 | "\n" | 仅在 file_format_type 为 `text`、`csv`、`json` 时使用 |
| have_partition | boolean | 否 | false | 是否需要处理分区 |
| partition_by | array | 否 | - | 仅在 have_partition 为 true 时使用 |
| partition_dir_expression | string | 否 | "${k0}=${v0}/${k1}=${v1}/.../${kn}=${vn}/" | 仅在 have_partition 为 true 时使用 |
Expand All @@ -73,6 +73,7 @@ import ChangeLog from '../changelog/connector-file-local.md';
| parquet_avro_write_fixed_as_int96 | array | 否 | - | 仅在 file_format 为 parquet 时使用 |
| enable_header_write | boolean | 否 | false | 仅在 file_format_type 为 text,csv 时使用。<br/> false:不写入表头,true:写入表头。 |
| encoding | string | 否 | "UTF-8" | 仅在 file_format_type 为 json,text,csv,xml 时使用 |
| merge_update_event | boolean | 否 | false | 仅当file_format_type为canal_json、debezium_json、maxwell_json. |

### path [string]

Expand Down Expand Up @@ -225,6 +226,12 @@ _root_tag [string]

仅在 file_format_type 为 json,text,csv,xml 时使用。文件写入的编码。该参数将通过 `Charset.forName(encoding)` 解析。

### merge_update_event [boolean]

仅当file_format_type为canal_json、debezium_json、maxwell_json时使用.
设置成true,序列化数据时,UPDATE_AFTER 和 UPDATE_BEFORE 会合并成 UPDATE;
设置成false,序列化数据时,UPDATE_AFTER 和 UPDATE_BEFORE 不会合并;

## 示例

对于 orc 文件格式的简单配置
Expand Down
Loading