RowKindExtractor is a transformation plugin in Apache SeaTunnel that can convert CDC data streams to Append-Only mode and extract the original RowKind information as a new field. This article introduces the core functionalities of RowKindExtractor, its usage in CDC data synchronization scenarios, configuration options, precautions, and multiple application examples. RowKindExtractor Apache SeaTunnel Append-Only mode RowKindExtractor RowKindExtractor RowKindExtractor The RowKindExtractor transformation plugin is used to convert CDC (Change Data Capture) data streams into Append-Only mode while extracting the original RowKind information into a new field. RowKindExtractor Append-Only Core Functions: Core Functions: Convert all RowKind values of data rows to +I(INSERT), achieving Append-Only mode.Preserve the original RowKind information (INSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE) in a newly added field.Support both SHORT and FULL output formats. Convert all RowKind values of data rows to +I(INSERT), achieving Append-Only mode. +I Preserve the original RowKind information (INSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE) in a newly added field. Support both SHORT and FULL output formats. SHORT FULL Why is this plugin needed? Why is this plugin needed? In CDC data synchronization scenarios, each data row carries a RowKind flag (+I, -U, +U, -D), representing different types of changes. Some downstream systems (such as data lakes or analytical platforms) only support Append-Only mode and do not support UPDATE or DELETE operations. In such cases, it is necessary to: +I -U +U -D Append-Only Convert all rows to INSERT type (Append-Only).Store the original change type in a regular field for downstream analysis. Convert all rows to INSERT type (Append-Only). Store the original change type in a regular field for downstream analysis. Transformation Example: Transformation Example: Input (CDC data):
 RowKind: -D (DELETE)
 Data: id=1, name="test1", age=20

Output (Append-Only data):
 RowKind: +I (INSERT)
 Data: id=1, name="test1", age=20, row_kind="DELETE" Input (CDC data):
 RowKind: -D (DELETE)
 Data: id=1, name="test1", age=20

Output (Append-Only data):
 RowKind: +I (INSERT)
 Data: id=1, name="test1", age=20, row_kind="DELETE" Typical Use Cases: Typical Use Cases: Writing CDC data into data lakes that only support Append-Only operations.Retaining a complete change history in data warehouses.Analyzing and aggregating different types of data changes. Writing CDC data into data lakes that only support Append-Only operations. Retaining a complete change history in data warehouses. Analyzing and aggregating different types of data changes. Configuration Options Configuration Options Parameter NameTypeRequiredDefault ValueDescriptioncustom_field_namestringNorow_kindName of the new field used to store the original RowKind informationtransform_typeenumNoSHORTOutput format of RowKind; possible values: SHORT (short format) or FULL (full format) Parameter NameTypeRequiredDefault ValueDescriptioncustom_field_namestringNorow_kindName of the new field used to store the original RowKind informationtransform_typeenumNoSHORTOutput format of RowKind; possible values: SHORT (short format) or FULL (full format) Parameter NameTypeRequiredDefault ValueDescription Parameter NameTypeRequiredDefault ValueDescription Parameter Name Type Required Default Value Description custom_field_namestringNorow_kindName of the new field used to store the original RowKind informationtransform_typeenumNoSHORTOutput format of RowKind; possible values: SHORT (short format) or FULL (full format) custom_field_namestringNorow_kindName of the new field used to store the original RowKind information custom_field_name string No row_kind Name of the new field used to store the original RowKind information transform_typeenumNoSHORTOutput format of RowKind; possible values: SHORT (short format) or FULL (full format) transform_type enum No SHORT Output format of RowKind; possible values: SHORT (short format) or FULL (full format) custom_field_name [string] custom_field_name [string] Specifies the name of the new field used to store the original RowKind information. Default: row_kind Default: row_kind Notes: Notes: The field name cannot duplicate an existing field, otherwise an error will occur.It is recommended to use meaningful names such as operation_type, change_type, or cdc_op. The field name cannot duplicate an existing field, otherwise an error will occur. It is recommended to use meaningful names such as operation_type, change_type, or cdc_op. operation_type change_type cdc_op Example: Example: custom_field_name = "operation_type"  # Use a custom field name custom_field_name = "operation_type"  # Use a custom field name transform_type [enum] transform_type [enum] Specifies the output format of the RowKind field value. Options: Options: FormatDescriptionOutput ValuesSHORTShort format (symbolic)+I, -U, +U, -DFULLFull format (English name)INSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE FormatDescriptionOutput ValuesSHORTShort format (symbolic)+I, -U, +U, -DFULLFull format (English name)INSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE FormatDescriptionOutput Values FormatDescriptionOutput Values Format Description Output Values SHORTShort format (symbolic)+I, -U, +U, -DFULLFull format (English name)INSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE SHORTShort format (symbolic)+I, -U, +U, -D SHORT Short format (symbolic) +I, -U, +U, -D FULLFull format (English name)INSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE FULL Full format (English name) INSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE Default: SHORT Default: Meaning of each value： Meaning of each value RowKind TypeSHORT FormatFULL FormatDescriptionINSERT+IINSERTInsert operationUPDATE_BEFORE-UUPDATE_BEFOREValue before updateUPDATE_AFTER+UUPDATE_AFTERValue after updateDELETE-DDELETEDelete operation RowKind TypeSHORT FormatFULL FormatDescriptionINSERT+IINSERTInsert operationUPDATE_BEFORE-UUPDATE_BEFOREValue before updateUPDATE_AFTER+UUPDATE_AFTERValue after updateDELETE-DDELETEDelete operation RowKind TypeSHORT FormatFULL FormatDescription RowKind TypeSHORT FormatFULL FormatDescription RowKind Type SHORT Format FULL Format Description INSERT+IINSERTInsert operationUPDATE_BEFORE-UUPDATE_BEFOREValue before updateUPDATE_AFTER+UUPDATE_AFTERValue after updateDELETE-DDELETEDelete operation INSERT+IINSERTInsert operation INSERT +I INSERT Insert operation UPDATE_BEFORE-UUPDATE_BEFOREValue before update UPDATE_BEFORE -U UPDATE_BEFORE Value before update UPDATE_AFTER+UUPDATE_AFTERValue after update UPDATE_AFTER +U UPDATE_AFTER Value after update DELETE-DDELETEDelete operation DELETE -D DELETE Delete operation Usage Recommendation: Usage Recommendation: SHORT format: Saves storage, suitable for space-sensitive scenarios.FULL format: More readable, suitable for human inspection or analysis. SHORT format: Saves storage, suitable for space-sensitive scenarios. SHORT format: FULL format: More readable, suitable for human inspection or analysis. FULL format: Example: Example: transform_type = FULL  # Use full format transform_type = FULL  # Use full format Full Examples Full Examples Example 1: Using Default Configuration (SHORT Format) Example 1: Using Default Configuration (SHORT Format) Using default configuration, CDC data is converted to Append-Only mode, with RowKind saved in short format. 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"
    # Default configuration:
    # custom_field_name = "row_kind"
    # transform_type = SHORT
  }
}

sink {
  Console {
    plugin_input = "append_only_data"
  }
} 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"
    # Default configuration:
    # custom_field_name = "row_kind"
    # transform_type = SHORT
  }
}

sink {
  Console {
    plugin_input = "append_only_data"
  }
} Data Transformation Process: Data Transformation Process: Input Data (CDC Format):
 1. RowKind=+I, id=1, name="Zhang San", age=25
 2. RowKind=-U, id=1, name="Zhang San", age=25
 3. RowKind=+U, id=1, name="Zhang San", age=26
 4. RowKind=-D, id=1, name="Zhang San", age=26

Output Data (Append-Only Format):
 1. RowKind=+I, id=1, name="Zhang San", age=25, row_kind="+I"
 2. RowKind=+I, id=1, name="Zhang San", age=25, row_kind="-U"
 3. RowKind=+I, id=1, name="Zhang San", age=26, row_kind="+U"
 4. RowKind=+I, id=1, name="Zhang San", age=26, row_kind="-D" Input Data (CDC Format):
 1. RowKind=+I, id=1, name="Zhang San", age=25
 2. RowKind=-U, id=1, name="Zhang San", age=25
 3. RowKind=+U, id=1, name="Zhang San", age=26
 4. RowKind=-D, id=1, name="Zhang San", age=26

Output Data (Append-Only Format):
 1. RowKind=+I, id=1, name="Zhang San", age=25, row_kind="+I"
 2. RowKind=+I, id=1, name="Zhang San", age=25, row_kind="-U"
 3. RowKind=+I, id=1, name="Zhang San", age=26, row_kind="+U"
 4. RowKind=+I, id=1, name="Zhang San", age=26, row_kind="-D" Example 2: Using FULL Format with Custom Field Name Example 2: Using FULL Format with Custom Field Name Output RowKind in full format and use a custom field name. 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"  # Custom field name
    transform_type = FULL                 # Use full format
  }
}

sink {
  Iceberg {
    plugin_input = "append_only_data"
    catalog_name = "iceberg_catalog"
    database = "mydb"
    table = "orders_history"
    # The Iceberg table will include the operation_type field to record the change type of each row
  }
} 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"  # Custom field name
    transform_type = FULL                 # Use full format
  }
}

sink {
  Iceberg {
    plugin_input = "append_only_data"
    catalog_name = "iceberg_catalog"
    database = "mydb"
    table = "orders_history"
    # The Iceberg table will include the operation_type field to record the change type of each row
  }
} Data Transformation Process: Data Transformation Process: Input Data (CDC Format):
 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

Output Data (Append-Only Format, 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" Input Data (CDC Format):
 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

Output Data (Append-Only Format, 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" Example 3: Full Test Using FakeSource Example 3: Full Test Using FakeSource Generate test data using FakeSource to demonstrate transformation of various RowKind values. FakeSource 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"
  }
} 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"
  }
} Expected Output: Expected Output: +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" +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"

The code in this story is for educational purposes. The readers are solely responsible for whatever they build with it.

SeaTunnel’s RowKindExtractor Simplifies CDC Pipelines With Append-Only Transformation

About Author

Comments

TOPICS

THIS ARTICLE WAS FEATURED IN

Related Stories

A 10-Minute Deep Dive Into the Core Architecture of Apache SeaTunnel and DataX

What the heck is Apache SeaTunnel?

How to Sync Data From MySQL to Doris Using SeaTunnel

How to Simplify Data Flow By Multi-Table Synchronization With Apache SeaTunnel

Source Code Analysis of Apache SeaTunnel Zeta Engine (Part 1): Server Initialization

The Evolution of Apache SeaTunnel’s Technical Architecture and Its Applications in the AI Field

A 10-Minute Deep Dive Into the Core Architecture of Apache SeaTunnel and DataX

What the heck is Apache SeaTunnel?

How to Sync Data From MySQL to Doris Using SeaTunnel

How to Simplify Data Flow By Multi-Table Synchronization With Apache SeaTunnel

Source Code Analysis of Apache SeaTunnel Zeta Engine (Part 1): Server Initialization

The Evolution of Apache SeaTunnel’s Technical Architecture and Its Applications in the AI Field

Light-Mode

Classic

Newspaper

Minty

Dark-Mode

Neon Noir

Minty

HN StartUps