You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by do...@apache.org on 2023/02/02 03:10:45 UTC
[inlong-website] branch master updated: [INLONG-675][Doc] Add document for dirty data archive (#676)
This is an automated email from the ASF dual-hosted git repository.
dockerzhang 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 e555e89336 [INLONG-675][Doc] Add document for dirty data archive (#676)
e555e89336 is described below
commit e555e893369c2ac88ef0ca2809c09b9c5301b6c7
Author: yunqingmoswu <44...@users.noreply.github.com>
AuthorDate: Thu Feb 2 11:10:39 2023 +0800
[INLONG-675][Doc] Add document for dirty data archive (#676)
Co-authored-by: yunqingmo <yu...@tencent.com>
---
docs/modules/sort/dirty_data_archive.md | 214 ++++++++++++++++++++
docs/modules/sort/overview.md | 2 +-
.../current/modules/sort/dirty_data_archive.md | 216 +++++++++++++++++++++
.../modules/sort/dirty_data_archive.md | 216 +++++++++++++++++++++
.../modules/sort/dirty_data_archive.md | 213 ++++++++++++++++++++
5 files changed, 860 insertions(+), 1 deletion(-)
diff --git a/docs/modules/sort/dirty_data_archive.md b/docs/modules/sort/dirty_data_archive.md
new file mode 100644
index 0000000000..6dbb13754e
--- /dev/null
+++ b/docs/modules/sort/dirty_data_archive.md
@@ -0,0 +1,214 @@
+---
+title: Dirty Data Archive
+sidebar_position: 5
+---
+
+## Overview
+
+Dirty data refers to data that cannot be extracted, transformed, and loaded correctly due to the quality of the data itself during the process of data extraction, transform, and loading.
+Common dirty data may have field types, lengths, and numbers that do not match. Data serialization, deserialization exception, target library, the table does not exist, etc.
+Dirty data archives can dump the dirty data during InLong data flow operation to third-party storage, which is convenient for businesses to find problems.
+Sort currently supports dirty data archiving for S3 and Log targets, as well as dirty data archiving for common data sources such as Kafka, Doris, Iceberg, HBase, Hive, Elasticsearch, and JDBC.
+
+## Supported Nodes
+
+| Type | Name | Archive Target System |
+|--------------|--------------------------------------------|-----------|
+| Extract Node | Kafka | Log, S3 |
+| Load Node | Hive | Log, S3 |
+| | Kafka | Log, S3 |
+| | HBase | Log, S3 |
+| | ClickHouse | Log, S3 |
+| | Iceberg | Log, S3 |
+| | Elasticsearch | Log, S3 |
+| | PostgreSQL | Log, S3 |
+| | HDFS | Log, S3 |
+| | TDSQL Postgres | Log, S3 |
+| | Doris | Log, S3 |
+| | SQL Server | Log, S3 |
+| | Greenplum | Log, S3 |
+
+## Dirty Data Format
+
+During the processing of dirty data archive, we define the following system variables for formatting dirty data:
+* SYSTEM_TIME: System time, the format is 'yyyy-MM-dd HH:mm:ss'
+* DIRTY_TYPE：Dirty type, common dirty types are SerializeError, DeserializeError, DataTypeMappingError, etc
+* DIRTY_MESSAGE: Dirty data message, that is the cause of dirty data, abnormal information, etc
+
+The format for archiving to Log：
+* [${dirty.side-output.log-tag}] ${value}, Among them, ${value} is the merged value of ${dirty.side-output.labels} and ${dirty data}, and formatted by 'csv' or 'json'
+
+The format for archiving to S3：
+* The filename generation format of S3: ${dirty.side-output.s3.key}/${dirty.identifier}-${randome-sequence}.${suffix}
+* The format of the data in the file of S3: it merge the ${dirty.side-output.labels} and ${dirty data} and formatted by 'csv' or 'json'
+
+**Notes**：For ${dirty.side-output.log-tag}, ${dirty.side-output.labels}, ${dirty.identifier}, ${dirty.side-output.s3.key} see the configuration details below.
+
+## Configuration
+
+### Common Configuration
+
+| Option | Required | Default | Type | Description |
+|---------|----------|---------|------|------------|
+| dirty.ignore | optional | false | Boolean | Whether to ignore dirty data, the default is 'false' |
+| dirty.side-output.enable | optional | false | Boolean | Whether to support dirty data archive, the default is 'false'|
+| dirty.side-output.connector | required | (none) | String | The connector of dirty data archive, it must be set when 'dirty.side-output.enable' is 'true', currently only 'log' and 's3' are supported. |
+| dirty.side-output.format | optional | csv | String | The format of dirty data archive, currently only supports 'csv' and 'json', defaults is 'csv'. |
+| dirty.side-output.log.enable | optional | true | Boolean |Whether to support log printing when dirty data is archived, the default is 'true'. |
+| dirty.side-output.ignore-errors | optional | true | Boolean | Whether to ignore errors in dirty data archives, defaults is 'true'. |
+| dirty.identifier | required | (none) | String | The identifier of dirty data, it will be used for filename generation of file dirty archive, topic generation of mq dirty archive, tablename generation of database, etc, and it supports variable replace like '${variable}'. There are several system variables(SYSTEM_TIME,DIRTY_TYPE,DIRTY_MESSAGE) are currently supported, and the support of other variables is determined by the connector. |
+| dirty.side-output.log-tag | optional | (none) | String | The log tag of dirty data, it will be used for log printing, and it supports variable replace like '${variable}'. There are several system variables(SYSTEM_TIME,DIRTY_TYPE,DIRTY_MESSAGE) are currently supported, and the support of other variables is determined by the connector. |
+| dirty.side-output.labels | optional | (none) | String | The labels of dirty data, it will be used for log printing and will be archived with dirty data, and it supports variable replace like '${variable}'. There are several system variables(SYSTEM_TIME,DIRTY_TYPE,DIRTY_MESSAGE) are currently supported, and the support of other variables is determined by the connector. |
+| dirty.side-output.field-delimiter | optional | , | String | The column separator of dirty data archive, it is used for 'csv' format, the default is ','. |
+| dirty.side-output.line-delimiter | optional | \n | String | The row separator of dirty data archive, it is used for 'csv' and 'json' format, the default is '\n'. |
+| dirty.side-output.batch.size | optional | 100 | Integer | The cache batch size of dirty data archive, the default is '100'.|
+| dirty.side-output.batch.bytes | optional | 10240 | Integer | The cache batch byte of dirty data archive, the unit is byte and the default is '10240'(10KB).|
+| dirty.side-output.retries | optional | 3 | Integer | The retris of dirty data archive，the default is '3'.|
+| dirty.side-output.batch.interval | optional | 60000 | Integer | The interval time of irty data archive, the unit is millisecond, the default is '60000'(60s).|
+
+### S3 Archive Configuration
+
+| Option | Required | Default | Type | Description |
+|---------|----------|---------|------|------------|
+| dirty.side-output.s3.endpoint | required | (none) | String | The endpoint of s3 archive |
+| dirty.side-output.s3.region | required | (none) | String | The region of s3 archive|
+| dirty.side-output.s3.bucket | required | (none) | String | The bucket of s3 archive|
+| dirty.side-output.s3.key | required | (none) | String | The key of s3 archive|
+| dirty.side-output.s3.access-key-id | optional | (none) | String | The access key id of s3 archive, it needs to be configured in the environment if this item is not configured. |
+| dirty.side-output.s3.secret-key-id | optional | (none) | String | The secret key id of s3 archive, it needs to be configured in the environment if this item is not configured. |
+
+## Usage
+
+One example about sync Kafka data to Kafka data and we will introduce usage of dirty data archive(it is similar to other nodes).
+
+* The useage for archive to log
+```sql
+create table `table_user_input`(
+ `id` INT,
+ `name` INT,
+ `age` STRING)
+ WITH (
+ 'dirty.side-output.connector' = 'log',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user',
+ 'inlong.metric.labels' = 'groupId=1&streamId=1&nodeId=1',
+ 'topic' = 'user_input',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'scan.startup.mode' = 'earliest-offset',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'false',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'false',
+ 'properties.group.id' = 'test_group');
+
+CREATE TABLE `table_user_output`(
+ `id` INT,
+ `name` STRING,
+ `age` INT)
+ WITH (
+ 'topic' = 'user_output',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'sink.ignore.changelog' = 'true',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'true',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'true',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.connector' = 'log',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.log.enable' = 'true',
+ 'dirty.side-output.log-tag' = 'DirtyData',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user');
+
+INSERT INTO `table_user_output`
+SELECT
+ `id`,
+ `name`,
+ `age`
+FROM `table_user_input`;
+-- In this example, we deliberately input a piece of data in non-json format, such as: 1,zhangsan,18, then the following dirty data will be printed in the log according to the configuration：
+[DirtyData] 2023-01-30 13:01:01 ValueDeserializeError,inlong,user,1,zhangsan,18
+```
+
+* The useage for archive to s3
+```sql
+create table `table_user_input`(
+ `id` INT,
+ `name` INT,
+ `age` STRING)
+ WITH (
+ 'dirty.side-output.connector' = 's3',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user',
+ 'dirty.side-output.s3.bucket' = 's3-test-bucket',
+ 'dirty.side-output.s3.endpoint' = 's3.test.endpoint',
+ 'dirty.side-output.s3.key' = 'dirty/test',
+ 'dirty.side-output.s3.region' = 'region',
+ 'dirty.side-output.s3.access-key-id' = 'access_key_id',
+ 'dirty.side-output.s3.secret-key-id' = 'secret_key_id',
+ 'dirty.identifier' = 'inlong-user-${SYSTEM_TIME}',
+ 'inlong.metric.labels' = 'groupId=1&streamId=1&nodeId=1',
+ 'topic' = 'user_input',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'scan.startup.mode' = 'earliest-offset',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'false',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'false',
+ 'properties.group.id' = 'test_group');
+
+CREATE TABLE `table_user_output`(
+ `id` INT,
+ `name` STRING,
+ `age` INT)
+ WITH (
+ 'topic' = 'user_output',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'sink.ignore.changelog' = 'true',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'true',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'true',
+ 'dirty.side-output.connector' = 's3',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user',
+ 'dirty.side-output.s3.bucket' = 's3-test-bucket',
+ 'dirty.side-output.s3.endpoint' = 's3.test.endpoint',
+ 'dirty.side-output.s3.key' = 'dirty/test',
+ 'dirty.side-output.s3.region' = 'region',
+ 'dirty.side-output.s3.access-key-id' = 'access_key_id',
+ 'dirty.side-output.s3.secret-key-id' = 'secret_key_id',
+ 'dirty.identifier' = 'inlong-user-${SYSTEM_TIME}');
+
+INSERT INTO `table_user_output`
+SELECT
+ `id`,
+ `name`,
+ `age`
+FROM `table_user_input`;
+-- In this example, we deliberately input a piece of data in non-json format, such as: 1,zhangsan,18, then the following dirty data will be written to s3 according to the configuration(the file path is: dirty/test/inlong-user-2023-01-01130101xxxx.txt, where xxxx is a 4-digit random sequence):
+[DirtyData] 2023-01-30 13:01:01 ValueDeserializeError,inlong,user,1,zhangsan,18
+```
\ No newline at end of file
diff --git a/docs/modules/sort/overview.md b/docs/modules/sort/overview.md
index d5e024b174..4585f9f629 100644
--- a/docs/modules/sort/overview.md
+++ b/docs/modules/sort/overview.md
@@ -9,7 +9,7 @@ InLong Sort is used to extract data from different source systems, then transfor
InLong Sort can be used together with the Manager to manage metadata, or it can run independently in the Flink environment.
## Features
-### Suport a variety of data nodes
+### Supports a variety of data nodes
| Type | Service |
|--------------|--------------------------------------------|
diff --git a/i18n/zh-CN/docusaurus-plugin-content-docs/current/modules/sort/dirty_data_archive.md b/i18n/zh-CN/docusaurus-plugin-content-docs/current/modules/sort/dirty_data_archive.md
new file mode 100644
index 0000000000..f040701d63
--- /dev/null
+++ b/i18n/zh-CN/docusaurus-plugin-content-docs/current/modules/sort/dirty_data_archive.md
@@ -0,0 +1,216 @@
+---
+title: 脏数据归档
+sidebar_position: 5
+---
+
+## 概览
+
+脏数据为在数据抽取、转换、加载过程中，因数据本身的质量问题导致数据无法正确抽取、转换、加载的数据，
+常见的脏数据可能有字段类型、长度、个数不匹配，数据序列化、反序列化异常，目标端库、表不存在等。
+脏数据归档可以将 InLong 数据流运行中的脏数据转存到第三方存储，便于业务查找问题。
+目前 Sort 支持了 S3、Log 两种目标端的脏数据归档，同时支持 Kafka、Doris、Iceberg、Hbase、Hive、Elasticsearch、JDBC 等常见数据源的脏数据归档。
+
+## 支持的节点
+
+| 类型 | 名称 | 归档目标端 |
+|--------------|--------------------------------------------|-----------|
+| Extract 节点 | Kafka | Log, S3 |
+| Load 节点 | Hive | Log, S3 |
+| | Kafka | Log, S3 |
+| | HBase | Log, S3 |
+| | ClickHouse | Log, S3 |
+| | Iceberg | Log, S3 |
+| | Elasticsearch | Log, S3 |
+| | PostgreSQL | Log, S3 |
+| | HDFS | Log, S3 |
+| | TDSQL Postgres | Log, S3 |
+| | Doris | Log, S3 |
+| | SQL Server | Log, S3 |
+| | Greenplum | Log, S3 |
+
+## 脏数据格式化
+
+在脏数据归档处理过程中，我们定义了下面的系统变量，用作脏数据的格式化：
+* SYSTEM_TIME: 系统时间，格式为 'yyyy-MM-dd HH:mm:ss'
+* DIRTY_TYPE：脏数据类型，常见的有 SerializeError、DeserializeError、DataTypeMappingError 等
+* DIRTY_MESSAGE: 脏数据信息，即脏数据产生的原因、异常信息等
+
+归档到 Log 的格式化为：
+* [${dirty.side-output.log-tag}] ${value}，其中 ${value} 为 ${dirty.side-output.labels} 和 ${脏数据} 合并，并按照 'csv' 或者 'json' 进行格式化后的值
+
+归档到 S3 的格式化为：
+* S3 文件名生成格式为: ${dirty.side-output.s3.key}/${dirty.identifier}-${随机序列}.${文件后缀名}
+* S3 文件内数据格式为：会将 ${dirty.side-output.labels} 和 ${脏数据} 合并，并按照 'csv' 或者 'json' 进行格式化
+
+**注**：${dirty.side-output.log-tag}、${dirty.side-output.labels}、${dirty.identifier}、${dirty.side-output.s3.key} 等见下面配置详解。
+
+## 配置
+
+### 公共配置
+
+| 参数 | 是否必选 | 默认值 | 数据类型 | 描述 |
+|---------|----------|---------|------|------------|
+| dirty.ignore | 可选 | false | Boolean | 是否忽略脏数据，只有为 'true' 才能进行脏数据归档，默认为 'false' |
+| dirty.side-output.enable | 可选 | false | Boolean | 是否支持脏数据归档，默认为 'false' |
+| dirty.side-output.connector | 必选 | (none) | String | 归档目标端名称，当 'dirty.side-output.enable' 为 'true' 时必须设置该值，目前仅支持 'log' 和 's3'。 |
+| dirty.side-output.format | 可选 | csv | String | 脏数据格式化，目前仅支持 'csv' 和 'json'，默认为 'csv'。 |
+| dirty.side-output.log.enable | 可选 | true | Boolean |脏数据归档时是否支持日志打印, 默认为 'true'。 |
+| dirty.side-output.ignore-errors | 可选 | true | Boolean | 是否忽略脏数据归档中的错误，默认为 'true'。 |
+| dirty.identifier | 必选 | (none) | String | 脏数据标识，用作 File 类型脏数据归档的文件名称生成或者 MQ 类型脏数据归档的 Topic 生成或者 数据库类型脏数据归档的 Tablename生成等。 它支持形如 ${variable} 变量替换，这里支持 SYSTEM_TIME、DIRTY_TYPE、DIRTY_MESSAGE 几种系统变量，其他的变量支持取决于具体的节点实现。 |
+| dirty.side-output.log-tag | 可选 | (none) | String | 脏数据 Tag，用作日志打印时标识等，比如 [${logTag}] ${logLabels},${dirtydata}。 它支持形如 ${variable} 变量替换，这里支持 SYSTEM_TIME、DIRTY_TYPE、DIRTY_MESSAGE 几种系统变量，其他的变量支持取决于具体的节点实现。 |
+| dirty.side-output.labels | 可选 | (none) | String | 脏数据 labels，用作日志打印标识，并将和脏数据一起归档等，比如 [${logTag}] ${logLabels},${dirtydata}。 它支持形如 ${variable} 变量替换，这里支持 SYSTEM_TIME、DIRTY_TYPE、DIRTY_MESSAGE 几种系统变量，其他的变量支持取决于具体的节点实现。 |
+| dirty.side-output.field-delimiter | 可选 | , | String | 脏数据归档列分割符，用作 'csv' 等格式化场景，默认为 ','。 |
+| dirty.side-output.line-delimiter | 可选 | \n | String | 脏数据归档行分割符，用作 'csv'、'json' 等格式化场景，默认为 '\n'。 |
+| dirty.side-output.batch.size | 可选 | 100 | Integer | 脏数据归档缓存 Batch 条数，默认为 '100'。|
+| dirty.side-output.batch.bytes | 可选 | 10240 | Integer | 脏数据归档缓存 Batch 大小，单位为 Byte, 默认为 '10240' 即 10KB。|
+| dirty.side-output.retries | 可选 | 3 | Integer | 脏数据归档发生错误时重试次数，默认为 '3'。 |
+| dirty.side-output.batch.interval | 可选 | 60000 | Integer | 脏数据归档间隔时间，单位为毫秒，默认为 '60000' 即 60s。|
+
+### S3 归档配置
+
+| 参数 | 是否必选 | 默认值 | 数据类型 | 描述 |
+|---------|----------|---------|------|------------|
+| dirty.side-output.s3.endpoint | 必选 | (none) | String | S3 归档的 Endpoint。 |
+| dirty.side-output.s3.region | 必选 | (none) | String | S3 归档的 Region。 |
+| dirty.side-output.s3.bucket | 必选 | (none) | String | S3 归档的 Bucket。 |
+| dirty.side-output.s3.key | 必选 | (none) | String | S3 归档的 Key。 |
+| dirty.side-output.s3.access-key-id | 可选 | (none) | String | S3 归档的 Access Key Id，若不配置该项，则需要在环境中配置好。 |
+| dirty.side-output.s3.secret-key-id | 可选 | (none) | String | S3 归档的 Secret Key Id，若不配置该项，则需要在环境中配置好。 |
+
+## 用法
+
+这里将介绍一个同步 Kafka 数据到 Kafka 的例子，同时介绍脏数据归档的使用，其他节点类似。
+
+* 归档到 Log 的使用
+```sql
+ create table `table_user_input`(
+ `id` INT,
+ `name` INT,
+ `age` STRING)
+ WITH (
+ 'dirty.side-output.connector' = 'log',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user',
+ 'inlong.metric.labels' = 'groupId=1&streamId=1&nodeId=1',
+ 'topic' = 'user_input',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'scan.startup.mode' = 'earliest-offset',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'false',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'false',
+ 'properties.group.id' = 'test_group');
+
+ CREATE TABLE `table_user_output`(
+ `id` INT,
+ `name` STRING,
+ `age` INT)
+ WITH (
+ 'topic' = 'user_output',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'sink.ignore.changelog' = 'true',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'true',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'true',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.connector' = 'log',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.log.enable' = 'true',
+ 'dirty.side-output.log-tag' = 'DirtyData',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user');
+
+ INSERT INTO `table_user_output`
+ SELECT
+ `id`,
+ `name`,
+ `age`
+ FROM `table_user_input`;
+-- 在这个例子中, 我们故意输入一条非json格式的数据，比如: 1,zhangsan,18，那么依据配置将在日志中打印如下脏数据：
+ [DirtyData] 2023-01-30 13:01:01 ValueDeserializeError,inlong,user,1,zhangsan,18
+```
+
+* 归档到 S3 的使用
+```sql
+ create table `table_user_input`(
+ `id` INT,
+ `name` INT,
+ `age` STRING)
+ WITH (
+ 'dirty.side-output.connector' = 's3',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user',
+ 'dirty.side-output.s3.bucket' = 's3-test-bucket',
+ 'dirty.side-output.s3.endpoint' = 's3.test.endpoint',
+ 'dirty.side-output.s3.key' = 'dirty/test',
+ 'dirty.side-output.s3.region' = 'region',
+ 'dirty.side-output.s3.access-key-id' = 'access_key_id',
+ 'dirty.side-output.s3.secret-key-id' = 'secret_key_id',
+ 'dirty.identifier' = 'inlong-user-${SYSTEM_TIME}',
+ 'inlong.metric.labels' = 'groupId=1&streamId=1&nodeId=1',
+ 'topic' = 'user_input',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'scan.startup.mode' = 'earliest-offset',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'false',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'false',
+ 'properties.group.id' = 'test_group');
+
+ CREATE TABLE `table_user_output`(
+ `id` INT,
+ `name` STRING,
+ `age` INT)
+ WITH (
+ 'topic' = 'user_output',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'sink.ignore.changelog' = 'true',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'true',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'true',
+ 'dirty.side-output.connector' = 's3',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user',
+ 'dirty.side-output.s3.bucket' = 's3-test-bucket',
+ 'dirty.side-output.s3.endpoint' = 's3.test.endpoint',
+ 'dirty.side-output.s3.key' = 'dirty/test',
+ 'dirty.side-output.s3.region' = 'region',
+ 'dirty.side-output.s3.access-key-id' = 'access_key_id',
+ 'dirty.side-output.s3.secret-key-id' = 'secret_key_id',
+ 'dirty.identifier' = 'inlong-user-${SYSTEM_TIME}');
+
+ INSERT INTO `table_user_output`
+ SELECT
+ `id`,
+ `name`,
+ `age`
+ FROM `table_user_input`;
+ -- 在这个例子中, 我们故意输入一条非json格式的数据，比如: 1,zhangsan,18，那么依据配置将向s3中写入如下脏数据（文件路径为: dirty/test/inlong-user-2023-01-01130101xxxx.txt, xxxx为4位随机序列）：
+ [DirtyData] 2023-01-30 13:01:01 ValueDeserializeError,inlong,user,1,zhangsan,18
+```
+
+
diff --git a/i18n/zh-CN/docusaurus-plugin-content-docs/version-1.5.0/modules/sort/dirty_data_archive.md b/i18n/zh-CN/docusaurus-plugin-content-docs/version-1.5.0/modules/sort/dirty_data_archive.md
new file mode 100644
index 0000000000..f040701d63
--- /dev/null
+++ b/i18n/zh-CN/docusaurus-plugin-content-docs/version-1.5.0/modules/sort/dirty_data_archive.md
@@ -0,0 +1,216 @@
+---
+title: 脏数据归档
+sidebar_position: 5
+---
+
+## 概览
+
+脏数据为在数据抽取、转换、加载过程中，因数据本身的质量问题导致数据无法正确抽取、转换、加载的数据，
+常见的脏数据可能有字段类型、长度、个数不匹配，数据序列化、反序列化异常，目标端库、表不存在等。
+脏数据归档可以将 InLong 数据流运行中的脏数据转存到第三方存储，便于业务查找问题。
+目前 Sort 支持了 S3、Log 两种目标端的脏数据归档，同时支持 Kafka、Doris、Iceberg、Hbase、Hive、Elasticsearch、JDBC 等常见数据源的脏数据归档。
+
+## 支持的节点
+
+| 类型 | 名称 | 归档目标端 |
+|--------------|--------------------------------------------|-----------|
+| Extract 节点 | Kafka | Log, S3 |
+| Load 节点 | Hive | Log, S3 |
+| | Kafka | Log, S3 |
+| | HBase | Log, S3 |
+| | ClickHouse | Log, S3 |
+| | Iceberg | Log, S3 |
+| | Elasticsearch | Log, S3 |
+| | PostgreSQL | Log, S3 |
+| | HDFS | Log, S3 |
+| | TDSQL Postgres | Log, S3 |
+| | Doris | Log, S3 |
+| | SQL Server | Log, S3 |
+| | Greenplum | Log, S3 |
+
+## 脏数据格式化
+
+在脏数据归档处理过程中，我们定义了下面的系统变量，用作脏数据的格式化：
+* SYSTEM_TIME: 系统时间，格式为 'yyyy-MM-dd HH:mm:ss'
+* DIRTY_TYPE：脏数据类型，常见的有 SerializeError、DeserializeError、DataTypeMappingError 等
+* DIRTY_MESSAGE: 脏数据信息，即脏数据产生的原因、异常信息等
+
+归档到 Log 的格式化为：
+* [${dirty.side-output.log-tag}] ${value}，其中 ${value} 为 ${dirty.side-output.labels} 和 ${脏数据} 合并，并按照 'csv' 或者 'json' 进行格式化后的值
+
+归档到 S3 的格式化为：
+* S3 文件名生成格式为: ${dirty.side-output.s3.key}/${dirty.identifier}-${随机序列}.${文件后缀名}
+* S3 文件内数据格式为：会将 ${dirty.side-output.labels} 和 ${脏数据} 合并，并按照 'csv' 或者 'json' 进行格式化
+
+**注**：${dirty.side-output.log-tag}、${dirty.side-output.labels}、${dirty.identifier}、${dirty.side-output.s3.key} 等见下面配置详解。
+
+## 配置
+
+### 公共配置
+
+| 参数 | 是否必选 | 默认值 | 数据类型 | 描述 |
+|---------|----------|---------|------|------------|
+| dirty.ignore | 可选 | false | Boolean | 是否忽略脏数据，只有为 'true' 才能进行脏数据归档，默认为 'false' |
+| dirty.side-output.enable | 可选 | false | Boolean | 是否支持脏数据归档，默认为 'false' |
+| dirty.side-output.connector | 必选 | (none) | String | 归档目标端名称，当 'dirty.side-output.enable' 为 'true' 时必须设置该值，目前仅支持 'log' 和 's3'。 |
+| dirty.side-output.format | 可选 | csv | String | 脏数据格式化，目前仅支持 'csv' 和 'json'，默认为 'csv'。 |
+| dirty.side-output.log.enable | 可选 | true | Boolean |脏数据归档时是否支持日志打印, 默认为 'true'。 |
+| dirty.side-output.ignore-errors | 可选 | true | Boolean | 是否忽略脏数据归档中的错误，默认为 'true'。 |
+| dirty.identifier | 必选 | (none) | String | 脏数据标识，用作 File 类型脏数据归档的文件名称生成或者 MQ 类型脏数据归档的 Topic 生成或者 数据库类型脏数据归档的 Tablename生成等。 它支持形如 ${variable} 变量替换，这里支持 SYSTEM_TIME、DIRTY_TYPE、DIRTY_MESSAGE 几种系统变量，其他的变量支持取决于具体的节点实现。 |
+| dirty.side-output.log-tag | 可选 | (none) | String | 脏数据 Tag，用作日志打印时标识等，比如 [${logTag}] ${logLabels},${dirtydata}。 它支持形如 ${variable} 变量替换，这里支持 SYSTEM_TIME、DIRTY_TYPE、DIRTY_MESSAGE 几种系统变量，其他的变量支持取决于具体的节点实现。 |
+| dirty.side-output.labels | 可选 | (none) | String | 脏数据 labels，用作日志打印标识，并将和脏数据一起归档等，比如 [${logTag}] ${logLabels},${dirtydata}。 它支持形如 ${variable} 变量替换，这里支持 SYSTEM_TIME、DIRTY_TYPE、DIRTY_MESSAGE 几种系统变量，其他的变量支持取决于具体的节点实现。 |
+| dirty.side-output.field-delimiter | 可选 | , | String | 脏数据归档列分割符，用作 'csv' 等格式化场景，默认为 ','。 |
+| dirty.side-output.line-delimiter | 可选 | \n | String | 脏数据归档行分割符，用作 'csv'、'json' 等格式化场景，默认为 '\n'。 |
+| dirty.side-output.batch.size | 可选 | 100 | Integer | 脏数据归档缓存 Batch 条数，默认为 '100'。|
+| dirty.side-output.batch.bytes | 可选 | 10240 | Integer | 脏数据归档缓存 Batch 大小，单位为 Byte, 默认为 '10240' 即 10KB。|
+| dirty.side-output.retries | 可选 | 3 | Integer | 脏数据归档发生错误时重试次数，默认为 '3'。 |
+| dirty.side-output.batch.interval | 可选 | 60000 | Integer | 脏数据归档间隔时间，单位为毫秒，默认为 '60000' 即 60s。|
+
+### S3 归档配置
+
+| 参数 | 是否必选 | 默认值 | 数据类型 | 描述 |
+|---------|----------|---------|------|------------|
+| dirty.side-output.s3.endpoint | 必选 | (none) | String | S3 归档的 Endpoint。 |
+| dirty.side-output.s3.region | 必选 | (none) | String | S3 归档的 Region。 |
+| dirty.side-output.s3.bucket | 必选 | (none) | String | S3 归档的 Bucket。 |
+| dirty.side-output.s3.key | 必选 | (none) | String | S3 归档的 Key。 |
+| dirty.side-output.s3.access-key-id | 可选 | (none) | String | S3 归档的 Access Key Id，若不配置该项，则需要在环境中配置好。 |
+| dirty.side-output.s3.secret-key-id | 可选 | (none) | String | S3 归档的 Secret Key Id，若不配置该项，则需要在环境中配置好。 |
+
+## 用法
+
+这里将介绍一个同步 Kafka 数据到 Kafka 的例子，同时介绍脏数据归档的使用，其他节点类似。
+
+* 归档到 Log 的使用
+```sql
+ create table `table_user_input`(
+ `id` INT,
+ `name` INT,
+ `age` STRING)
+ WITH (
+ 'dirty.side-output.connector' = 'log',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user',
+ 'inlong.metric.labels' = 'groupId=1&streamId=1&nodeId=1',
+ 'topic' = 'user_input',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'scan.startup.mode' = 'earliest-offset',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'false',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'false',
+ 'properties.group.id' = 'test_group');
+
+ CREATE TABLE `table_user_output`(
+ `id` INT,
+ `name` STRING,
+ `age` INT)
+ WITH (
+ 'topic' = 'user_output',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'sink.ignore.changelog' = 'true',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'true',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'true',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.connector' = 'log',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.log.enable' = 'true',
+ 'dirty.side-output.log-tag' = 'DirtyData',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user');
+
+ INSERT INTO `table_user_output`
+ SELECT
+ `id`,
+ `name`,
+ `age`
+ FROM `table_user_input`;
+-- 在这个例子中, 我们故意输入一条非json格式的数据，比如: 1,zhangsan,18，那么依据配置将在日志中打印如下脏数据：
+ [DirtyData] 2023-01-30 13:01:01 ValueDeserializeError,inlong,user,1,zhangsan,18
+```
+
+* 归档到 S3 的使用
+```sql
+ create table `table_user_input`(
+ `id` INT,
+ `name` INT,
+ `age` STRING)
+ WITH (
+ 'dirty.side-output.connector' = 's3',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user',
+ 'dirty.side-output.s3.bucket' = 's3-test-bucket',
+ 'dirty.side-output.s3.endpoint' = 's3.test.endpoint',
+ 'dirty.side-output.s3.key' = 'dirty/test',
+ 'dirty.side-output.s3.region' = 'region',
+ 'dirty.side-output.s3.access-key-id' = 'access_key_id',
+ 'dirty.side-output.s3.secret-key-id' = 'secret_key_id',
+ 'dirty.identifier' = 'inlong-user-${SYSTEM_TIME}',
+ 'inlong.metric.labels' = 'groupId=1&streamId=1&nodeId=1',
+ 'topic' = 'user_input',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'scan.startup.mode' = 'earliest-offset',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'false',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'false',
+ 'properties.group.id' = 'test_group');
+
+ CREATE TABLE `table_user_output`(
+ `id` INT,
+ `name` STRING,
+ `age` INT)
+ WITH (
+ 'topic' = 'user_output',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'sink.ignore.changelog' = 'true',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'true',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'true',
+ 'dirty.side-output.connector' = 's3',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user',
+ 'dirty.side-output.s3.bucket' = 's3-test-bucket',
+ 'dirty.side-output.s3.endpoint' = 's3.test.endpoint',
+ 'dirty.side-output.s3.key' = 'dirty/test',
+ 'dirty.side-output.s3.region' = 'region',
+ 'dirty.side-output.s3.access-key-id' = 'access_key_id',
+ 'dirty.side-output.s3.secret-key-id' = 'secret_key_id',
+ 'dirty.identifier' = 'inlong-user-${SYSTEM_TIME}');
+
+ INSERT INTO `table_user_output`
+ SELECT
+ `id`,
+ `name`,
+ `age`
+ FROM `table_user_input`;
+ -- 在这个例子中, 我们故意输入一条非json格式的数据，比如: 1,zhangsan,18，那么依据配置将向s3中写入如下脏数据（文件路径为: dirty/test/inlong-user-2023-01-01130101xxxx.txt, xxxx为4位随机序列）：
+ [DirtyData] 2023-01-30 13:01:01 ValueDeserializeError,inlong,user,1,zhangsan,18
+```
+
+
diff --git a/versioned_docs/version-1.5.0/modules/sort/dirty_data_archive.md b/versioned_docs/version-1.5.0/modules/sort/dirty_data_archive.md
new file mode 100644
index 0000000000..4b17cdfcb0
--- /dev/null
+++ b/versioned_docs/version-1.5.0/modules/sort/dirty_data_archive.md
@@ -0,0 +1,213 @@
+---
+title: Dirty Data Archive
+sidebar_position: 5
+---
+
+## Overview
+
+Dirty data refers to data that cannot be extracted, transformed, and loaded correctly due to the quality of the data itself during the process of data extraction, transform, and loading.
+Common dirty data may have field types, lengths, and numbers that do not match. Data serialization, deserialization exception, target library, the table does not exist, etc.
+Dirty data archives can dump the dirty data during InLong data flow operation to third-party storage, which is convenient for businesses to find problems.
+Sort currently supports dirty data archiving for S3 and Log targets, as well as dirty data archiving for common data sources such as Kafka, Doris, Iceberg, HBase, Hive, Elasticsearch, and JDBC.
+## Supported Nodes
+
+| Type | Name | Archive Target System |
+|--------------|--------------------------------------------|-----------|
+| Extract Node | Kafka | Log, S3 |
+| Load Node | Hive | Log, S3 |
+| | Kafka | Log, S3 |
+| | HBase | Log, S3 |
+| | ClickHouse | Log, S3 |
+| | Iceberg | Log, S3 |
+| | Elasticsearch | Log, S3 |
+| | PostgreSQL | Log, S3 |
+| | HDFS | Log, S3 |
+| | TDSQL Postgres | Log, S3 |
+| | Doris | Log, S3 |
+| | SQL Server | Log, S3 |
+| | Greenplum | Log, S3 |
+
+## Dirty Data Format
+
+During the processing of dirty data archive, we define the following system variables for formatting dirty data:
+* SYSTEM_TIME: System time, the format is 'yyyy-MM-dd HH:mm:ss'
+* DIRTY_TYPE：Dirty type, common dirty types are SerializeError, DeserializeError, DataTypeMappingError, etc
+* DIRTY_MESSAGE: Dirty data message, that is the cause of dirty data, abnormal information, etc
+
+The format for archiving to Log：
+* [${dirty.side-output.log-tag}] ${value}, Among them, ${value} is the merged value of ${dirty.side-output.labels} and ${dirty data}, and formatted by 'csv' or 'json'
+
+The format for archiving to S3：
+* The filename generation format of S3: ${dirty.side-output.s3.key}/${dirty.identifier}-${randome-sequence}.${suffix}
+* The format of the data in the file of S3: it merge the ${dirty.side-output.labels} and ${dirty data} and formatted by 'csv' or 'json'
+
+**Notes**：For ${dirty.side-output.log-tag}, ${dirty.side-output.labels}, ${dirty.identifier}, ${dirty.side-output.s3.key} see the configuration details below.
+
+## Configuration
+
+### Common Configuration
+
+| Option | Required | Default | Type | Description |
+|---------|----------|---------|------|------------|
+| dirty.ignore | optional | false | Boolean | Whether to ignore dirty data, the default is 'false' |
+| dirty.side-output.enable | optional | false | Boolean | Whether to support dirty data archive, the default is 'false'|
+| dirty.side-output.connector | required | (none) | String | The connector of dirty data archive, it must be set when 'dirty.side-output.enable' is 'true', currently only 'log' and 's3' are supported. |
+| dirty.side-output.format | optional | csv | String | The format of dirty data archive, currently only supports 'csv' and 'json', defaults is 'csv'. |
+| dirty.side-output.log.enable | optional | true | Boolean |Whether to support log printing when dirty data is archived, the default is 'true'. |
+| dirty.side-output.ignore-errors | optional | true | Boolean | Whether to ignore errors in dirty data archives, defaults is 'true'. |
+| dirty.identifier | required | (none) | String | The identifier of dirty data, it will be used for filename generation of file dirty archive, topic generation of mq dirty archive, tablename generation of database, etc, and it supports variable replace like '${variable}'. There are several system variables(SYSTEM_TIME,DIRTY_TYPE,DIRTY_MESSAGE) are currently supported, and the support of other variables is determined by the connector. |
+| dirty.side-output.log-tag | optional | (none) | String | The log tag of dirty data, it will be used for log printing, and it supports variable replace like '${variable}'. There are several system variables(SYSTEM_TIME,DIRTY_TYPE,DIRTY_MESSAGE) are currently supported, and the support of other variables is determined by the connector. |
+| dirty.side-output.labels | optional | (none) | String | The labels of dirty data, it will be used for log printing and will be archived with dirty data, and it supports variable replace like '${variable}'. There are several system variables(SYSTEM_TIME,DIRTY_TYPE,DIRTY_MESSAGE) are currently supported, and the support of other variables is determined by the connector. |
+| dirty.side-output.field-delimiter | optional | , | String | The column separator of dirty data archive, it is used for 'csv' format, the default is ','. |
+| dirty.side-output.line-delimiter | optional | \n | String | The row separator of dirty data archive, it is used for 'csv' and 'json' format, the default is '\n'. |
+| dirty.side-output.batch.size | optional | 100 | Integer | The cache batch size of dirty data archive, the default is '100'.|
+| dirty.side-output.batch.bytes | optional | 10240 | Integer | The cache batch byte of dirty data archive, the unit is byte and the default is '10240'(10KB).|
+| dirty.side-output.retries | optional | 3 | Integer | The retris of dirty data archive，the default is '3'.|
+| dirty.side-output.batch.interval | optional | 60000 | Integer | The interval time of irty data archive, the unit is millisecond, the default is '60000'(60s).|
+
+### S3 Archive Configuration
+
+| Option | Required | Default | Type | Description |
+|---------|----------|---------|------|------------|
+| dirty.side-output.s3.endpoint | required | (none) | String | The endpoint of s3 archive |
+| dirty.side-output.s3.region | required | (none) | String | The region of s3 archive|
+| dirty.side-output.s3.bucket | required | (none) | String | The bucket of s3 archive|
+| dirty.side-output.s3.key | required | (none) | String | The key of s3 archive|
+| dirty.side-output.s3.access-key-id | optional | (none) | String | The access key id of s3 archive, it needs to be configured in the environment if this item is not configured. |
+| dirty.side-output.s3.secret-key-id | optional | (none) | String | The secret key id of s3 archive, it needs to be configured in the environment if this item is not configured. |
+
+## Usage
+
+One example about sync Kafka data to Kafka data and we will introduce usage of dirty data archive(it is similar to other nodes).
+
+* The useage for archive to log
+```sql
+create table `table_user_input`(
+ `id` INT,
+ `name` INT,
+ `age` STRING)
+ WITH (
+ 'dirty.side-output.connector' = 'log',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user',
+ 'inlong.metric.labels' = 'groupId=1&streamId=1&nodeId=1',
+ 'topic' = 'user_input',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'scan.startup.mode' = 'earliest-offset',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'false',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'false',
+ 'properties.group.id' = 'test_group');
+
+CREATE TABLE `table_user_output`(
+ `id` INT,
+ `name` STRING,
+ `age` INT)
+ WITH (
+ 'topic' = 'user_output',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'sink.ignore.changelog' = 'true',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'true',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'true',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.connector' = 'log',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.log.enable' = 'true',
+ 'dirty.side-output.log-tag' = 'DirtyData',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user');
+
+INSERT INTO `table_user_output`
+SELECT
+ `id`,
+ `name`,
+ `age`
+FROM `table_user_input`;
+-- In this example, we deliberately input a piece of data in non-json format, such as: 1,zhangsan,18, then the following dirty data will be printed in the log according to the configuration：
+[DirtyData] 2023-01-30 13:01:01 ValueDeserializeError,inlong,user,1,zhangsan,18
+```
+
+* The useage for archive to s3
+```sql
+create table `table_user_input`(
+ `id` INT,
+ `name` INT,
+ `age` STRING)
+ WITH (
+ 'dirty.side-output.connector' = 's3',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user',
+ 'dirty.side-output.s3.bucket' = 's3-test-bucket',
+ 'dirty.side-output.s3.endpoint' = 's3.test.endpoint',
+ 'dirty.side-output.s3.key' = 'dirty/test',
+ 'dirty.side-output.s3.region' = 'region',
+ 'dirty.side-output.s3.access-key-id' = 'access_key_id',
+ 'dirty.side-output.s3.secret-key-id' = 'secret_key_id',
+ 'dirty.identifier' = 'inlong-user-${SYSTEM_TIME}',
+ 'inlong.metric.labels' = 'groupId=1&streamId=1&nodeId=1',
+ 'topic' = 'user_input',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'scan.startup.mode' = 'earliest-offset',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'false',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'false',
+ 'properties.group.id' = 'test_group');
+
+CREATE TABLE `table_user_output`(
+ `id` INT,
+ `name` STRING,
+ `age` INT)
+ WITH (
+ 'topic' = 'user_output',
+ 'properties.bootstrap.servers' = 'localhost:9092',
+ 'connector' = 'kafka-inlong',
+ 'sink.ignore.changelog' = 'true',
+ 'json.timestamp-format.standard' = 'SQL',
+ 'json.encode.decimal-as-plain-number' = 'true',
+ 'json.map-null-key.literal' = 'null',
+ 'json.ignore-parse-errors' = 'true',
+ 'json.map-null-key.mode' = 'DROP',
+ 'format' = 'json',
+ 'json.fail-on-missing-field' = 'true',
+ 'dirty.side-output.connector' = 's3',
+ 'dirty.ignore' = 'true',
+ 'dirty.side-output.enable' = 'true',
+ 'dirty.side-output.format' = 'csv',
+ 'dirty.side-output.labels' = 'SYSTEM_TIME=${SYSTEM_TIME}&DIRTY_TYPE=${DIRTY_TYPE}&database=inlong&table=user',
+ 'dirty.side-output.s3.bucket' = 's3-test-bucket',
+ 'dirty.side-output.s3.endpoint' = 's3.test.endpoint',
+ 'dirty.side-output.s3.key' = 'dirty/test',
+ 'dirty.side-output.s3.region' = 'region',
+ 'dirty.side-output.s3.access-key-id' = 'access_key_id',
+ 'dirty.side-output.s3.secret-key-id' = 'secret_key_id',
+ 'dirty.identifier' = 'inlong-user-${SYSTEM_TIME}');
+
+INSERT INTO `table_user_output`
+SELECT
+ `id`,
+ `name`,
+ `age`
+FROM `table_user_input`;
+-- In this example, we deliberately input a piece of data in non-json format, such as: 1,zhangsan,18, then the following dirty data will be written to s3 according to the configuration(the file path is: dirty/test/inlong-user-2023-01-01130101xxxx.txt, where xxxx is a 4-digit random sequence):
+[DirtyData] 2023-01-30 13:01:01 ValueDeserializeError,inlong,user,1,zhangsan,18
+```
\ No newline at end of file