
![4aea1ddbb5a97ca5e7ab7a20b5bea8f1](https://developer.qcloudimg.com/http-save/yehe-11851421/a4f31bd3cb4662fbb28e742af65708dc.png)

RowKindExtractor 是 Apache SeaTunnel 的一个转换插件，它能将 CDC 数据流转为 Append-Only 模式，并提取原始 RowKind 信息为新字段。本文将介绍 RowKindExtractor 的核心功能，其在 CDC 数据同步场景下的使用方法，以及配置选项、注意事项及多种应用示例。 

## RowKindExtractor
RowKindExtractor 转换插件用于将 CDC（Change Data Capture）数据流转换为 Append-Only（仅追加）模式，同时将原始的 RowKind 信息提取为一个新的字段。

**核心功能：**
* 将所有数据行的 RowKind 统一改为 +I（INSERT），实现 Append-Only 模式
* 将原始的 RowKind 信息（INSERT、UPDATE_BEFORE、UPDATE_AFTER、DELETE）保存到新增的字段中
* 支持短格式和完整格式两种输出方式

**为什么需要这个插件？**

在 CDC 数据同步场景中，数据行带有 RowKind 标记（+I、-U、+U、-D），表示不同的变更类型。但某些下游系统（如数据湖、分析系统）只支持 Append-Only 模式，不支持 UPDATE 和 DELETE 操作。此时需要：
1. 将所有数据转换为 INSERT 类型（Append-Only）
2. 将原始的变更类型保存为普通字段，供后续分析使用

**转换示例：**


```
输入（CDC 数据）：
  RowKind: -D (DELETE)
  数据: id=1, name="test1", age=20

输出（Append-Only 数据）：
  RowKind: +I (INSERT)
  数据: id=1, name="test1", age=20, row_kind="DELETE"
```
  
  
**典型应用场景**：
* 将 CDC 数据写入只支持 Append 的数据湖
* 需要在数据仓库中保留完整的变更历史记录
* 需要对不同类型的变更进行统计分析

## 配置选项

![](https://developer.qcloudimg.com/http-save/yehe-11851421/c7cf0868d3f34052ccb107353b8fe4e5.jpg)


**custom_field_name [string]**

指定新增字段的名称，该字段用于存储原始的 RowKind 信息。

默认值：row_kind

注意事项：
- 字段名不能与原有字段重名，否则会报错
- 建议使用有意义的名称，如 operation_type、change_type、cdc_op 等

示例：
```
custom_field_name = "operation_type"  # 使用自定义字段名
```
**transform_type [enum]**

指定 RowKind 字段值的输出格式。

可选值：
![](https://developer.qcloudimg.com/http-save/yehe-11851421/7076c95b21d4daa272872c3f358640e8.jpg)


默认值：SHORT

各值含义：

![](https://developer.qcloudimg.com/http-save/yehe-11851421/e4892b0ed988b2e6e3cf882c503ea93a.jpg)


选择建议：
- SHORT 格式：节省存储空间，适合对存储敏感的场景
- FULL 格式：可读性更好，适合需要人工查看或分析的场景

示例：

```
transform_type = FULL  # 使用完整格式
```

## 完整示例

- 示例 1：使用默认配置（SHORT 格式）

使用默认配置，将 CDC 数据转换为 Append-Only 模式，RowKind 以短格式保存。
```
env {
  parallelism = 1
  job.mode = "STREAMING"
}

source {
  MySQL-CDC {
    plugin_output = "cdc_source"
    server-id = 5652
    username = "root"
    password = "your_password"
    table-names = ["mydb.users"]
    url = "jdbc:mysql://localhost:3306/mydb"
  }
}

transform {
  RowKindExtractor {
    plugin_input = "cdc_source"
    plugin_output = "append_only_data"
    # 使用默认配置：
    # custom_field_name = "row_kind"
    # transform_type = SHORT
  }
}

sink {
  Console {
    plugin_input = "append_only_data"
  }
}
```
数据转换过程：
```
输入数据（CDC 格式）：
  1. RowKind=+I, id=1, name="张三", age=25
  2. RowKind=-U, id=1, name="张三", age=25
  3. RowKind=+U, id=1, name="张三", age=26
  4. RowKind=-D, id=1, name="张三", age=26

输出数据（Append-Only 格式）：
  1. RowKind=+I, id=1, name="张三", age=25, row_kind="+I"
  2. RowKind=+I, id=1, name="张三", age=25, row_kind="-U"
  3. RowKind=+I, id=1, name="张三", age=26, row_kind="+U"
  4. RowKind=+I, id=1, name="张三", age=26, row_kind="-D"
```
- 示例 2：使用 FULL 格式和自定义字段名

使用完整格式输出 RowKind，并自定义字段名称。

```
env {
  parallelism = 1
  job.mode = "STREAMING"
}

source {
  MySQL-CDC {
    plugin_output = "cdc_source"
    server-id = 5652
    username = "root"
    password = "your_password"
    table-names = ["mydb.orders"]
    url = "jdbc:mysql://localhost:3306/mydb"
  }
}

transform {
  RowKindExtractor {
    plugin_input = "cdc_source"
    plugin_output = "append_only_data"
    custom_field_name = "operation_type"  # 自定义字段名
    transform_type = FULL                 # 使用完整格式
  }
}

sink {
  Iceberg {
    plugin_input = "append_only_data"
    catalog_name = "iceberg_catalog"
    database = "mydb"
    table = "orders_history"
    # Iceberg 表会包含 operation_type 字段，记录每条数据的变更类型
  }
}
```

```
数据转换过程：

输入数据（CDC 格式）：
  1. RowKind=+I, order_id=1001, amount=100.00
  2. RowKind=-U, order_id=1001, amount=100.00
  3. RowKind=+U, order_id=1001, amount=150.00
  4. RowKind=-D, order_id=1001, amount=150.00

输出数据（Append-Only 格式，FULL 格式）：
  1. RowKind=+I, order_id=1001, amount=100.00, operation_type="INSERT"
  2. RowKind=+I, order_id=1001, amount=100.00, operation_type="UPDATE_BEFORE"
  3. RowKind=+I, order_id=1001, amount=150.00, operation_type="UPDATE_AFTER"
  4. RowKind=+I, order_id=1001, amount=150.00, operation_type="DELETE"
```


- 示例 3：完整的测试示例（使用 FakeSource）

使用 FakeSource 生成测试数据，演示各种 RowKind 的转换效果。
```
env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    plugin_output = "fake_cdc_data"
    schema = {
      fields {
        pk_id = bigint
        name = string
        score = int
      }
      primaryKey {
        name = "pk_id"
        columnNames = [pk_id]
      }
    }
    rows = [
      {
        kind = INSERT
        fields = [1, "A", 100]
      },
      {
        kind = INSERT
        fields = [2, "B", 100]
      },
      {
        kind = UPDATE_BEFORE
        fields = [1, "A", 100]
      },
      {
        kind = UPDATE_AFTER
        fields = [1, "A_updated", 95]
      },
      {
        kind = UPDATE_BEFORE
        fields = [2, "B", 100]
      },
      {
        kind = UPDATE_AFTER
        fields = [2, "B_updated", 98]
      },
      {
        kind = DELETE
        fields = [1, "A_updated", 95]
      }
    ]
  }
}

transform {
  RowKindExtractor {
    plugin_input = "fake_cdc_data"
    plugin_output = "transformed_data"
    custom_field_name = "change_type"
    transform_type = FULL
  }
}

sink {
  Console {
    plugin_input = "transformed_data"
  }
}
```


预期输出：
```
+I, pk_id=1, name="A", score=100, change_type="INSERT"
+I, pk_id=2, name="B", score=100, change_type="INSERT"
+I, pk_id=1, name="A", score=100, change_type="UPDATE_BEFORE"
+I, pk_id=1, name="A_updated", score=95, change_type="UPDATE_AFTER"
+I, pk_id=2, name="B", score=100, change_type="UPDATE_BEFORE"
+I, pk_id=2, name="B_updated", score=98, change_type="UPDATE_AFTER"
+I, pk_id=1, name="A_updated", score=95, change_type="DELETE"
```    

RowKindExtractor 是 Apache SeaTunnel 的一个转换插件，它能将 CDC 数据流转为 Append-Only 模式，并提取原始 RowKind 信息为新字段。本文将介绍 RowKindExtractor 的核心功能，其在 CDC 数据同步场景下的使用方法，以及配置选项、注意事项及多种应用示例。

Apache SeaTunnel 如何将 CDC 数据流转换为 Append-Only 模式？

数据库

大数据

后端

RowKindExtractor是Apache SeaTunnel插件，可将CDC数据流转换为Append-Only模式并保留原始变更类型。支持自定义字段名和输出格式，适用于数据湖同步、变更历史记录等场景，帮助下游系统处理CDC数据。

数据仓库

数据同步

数据转换

数据湖

数据流

MySQL

11.11畅享AI算力 云启新增长

bigdata-class

云数据库采购特惠，高性能、高可靠、高安全，超值优惠等你享！


datainlong

tchouse

文章

问答

视频

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

Apache SeaTunnel 如何将 CDC 数据流转换为 Append-Only 模式？-腾讯云开发者社区-腾讯云

Apache SeaTunnel 如何将 CDC 数据流转换为 Append-Only 模式？

Apache SeaTunnel 如何将 CDC 数据流转换为 Append-Only 模式？

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐