This document describes how to understand, develop and contribute a transform.

We also provide the transform e2e test to verify the data input and output by the transform.

Concepts

Using SeaTunnel you can read or write data through the connector, but if you need to process your data after reading or before writing, then need to use transform.

Use transform to make simple edits to your data rows or fields, such as split field, change field values, add or remove field.

DataType transform

Transform receives datatype input from upstream(source or transform) and outputs new datatype to downstream(sink or transform), this process is datatype transform.

Example 1:Remove fields

  1. | A | B | C |
  2. |-----------|-----------|-----------|
  3. | STRING | INT | BOOLEAN |
  4. | A | B |
  5. |-----------|-----------|
  6. | STRING | INT |

Example 2:Sort fields

  1. | B | C | A |
  2. |-----------|-----------|-----------|
  3. | INT | BOOLEAN | STRING |
  4. | A | B | C |
  5. |-----------|-----------|-----------|
  6. | STRING | INT | BOOLEAN |

Example 3:Update fields datatype

  1. | A | B | C |
  2. |-----------|-----------|-----------|
  3. | STRING | INT | BOOLEAN |
  4. | A | B | C |
  5. |-----------|-----------|-----------|
  6. | STRING | STRING | STRING |

Example 4:Add new fields

  1. | A | B | C |
  2. |-----------|-----------|-----------|
  3. | STRING | INT | BOOLEAN |
  4. | A | B | C | D |
  5. |-----------|-----------|-----------|-----------|
  6. | STRING | INT | BOOLEAN | DOUBLE |

Data transform

After datatype transformed, Transform will receives data-row input from upstream(source or transform), edit into data-row with new datatype and outputs to downstream (sink or transform). This process is data transform.

Translation

Transform is decoupled from the execution engine, any transform implement can run into all engines without change the code & config, which requires the translation layer to adapt transform and execution engine.

Example:Translation datatype & data

  1. Original:
  2. | A | B | C |
  3. |-----------|-----------|-----------|
  4. | STRING | INT | BOOLEAN |
  5. Datatype translation:
  6. | A | B | C |
  7. |-------------------|-------------------|-------------------|
  8. | ENGINE<STRING> | ENGINE<INT> | ENGINE<BOOLEAN> |
  9. Data translation:
  10. | A | B | C |
  11. |-------------------|-------------------|-------------------|
  12. | ENGINE<"test"> | ENGINE<1> | ENGINE<false> |

Core APIs

SeaTunnelTransform

SeaTunnelTransform provides all major and primary APIs, you can subclass it to do whatever transform.

  1. Receive datatype input from upstream.
  1. /**
  2. * Set the data type info of input data.
  3. *
  4. * @param inputDataType The data type info of upstream input.
  5. */
  6. void setTypeInfo(SeaTunnelDataType<T> inputDataType);
  1. Outputs new datatype to downstream.
  1. /**
  2. * Get the data type of the records produced by this transform.
  3. *
  4. * @return Produced data type.
  5. */
  6. SeaTunnelDataType<T> getProducedType();
  1. Edit input data and outputs new data to downstream.
  1. /**
  2. * Transform input data to {@link this#getProducedType()} types data.
  3. *
  4. * @param row the data need be transform.
  5. * @return transformed data.
  6. */
  7. T map(T row);

SingleFieldOutputTransform

SingleFieldOutputTransform abstract single field change operator

  1. Define output field
  1. /**
  2. * Outputs new field
  3. *
  4. * @return
  5. */
  6. protected abstract String getOutputFieldName();
  1. Define output field datatype
  1. /**
  2. * Outputs new field datatype
  3. *
  4. * @return
  5. */
  6. protected abstract SeaTunnelDataType getOutputFieldDataType();
  1. Define output field value
  1. /**
  2. * Outputs new field value
  3. *
  4. * @param inputRow The inputRow of upstream input.
  5. * @return
  6. */
  7. protected abstract Object getOutputFieldValue(SeaTunnelRowAccessor inputRow);

MultipleFieldOutputTransform

MultipleFieldOutputTransform abstract multiple fields change operator

  1. Define output fields
  1. /**
  2. * Outputs new fields
  3. *
  4. * @return
  5. */
  6. protected abstract String[] getOutputFieldNames();
  1. Define output fields datatype
  1. /**
  2. * Outputs new fields datatype
  3. *
  4. * @return
  5. */
  6. protected abstract SeaTunnelDataType[] getOutputFieldDataTypes();
  1. Define output field values
  1. /**
  2. * Outputs new fields value
  3. *
  4. * @param inputRow The inputRow of upstream input.
  5. * @return
  6. */
  7. protected abstract Object[] getOutputFieldValues(SeaTunnelRowAccessor inputRow);

AbstractSeaTunnelTransform

AbstractSeaTunnelTransform abstract datatype & fields change operator

  1. Transform input row type and outputs new row type
  1. /**
  2. * Outputs transformed row type.
  3. *
  4. * @param inputRowType upstream input row type
  5. * @return
  6. */
  7. protected abstract SeaTunnelRowType transformRowType(SeaTunnelRowType inputRowType);
  1. Transform input row data and outputs new row data
  1. /**
  2. * Outputs transformed row data.
  3. *
  4. * @param inputRow upstream input row data
  5. * @return
  6. */
  7. protected abstract SeaTunnelRow transformRow(SeaTunnelRow inputRow);

Develop a Transform

It must implement one of the following APIs:

  • SeaTunnelTransform
  • AbstractSeaTunnelTransform
  • SingleFieldOutputTransform
  • MultipleFieldOutputTransform

Add implement subclass into module seatunnel-transforms-v2.

Example: copy field to new field

  1. @AutoService(SeaTunnelTransform.class)
  2. public class CopyFieldTransform extends SingleFieldOutputTransform {
  3. private String srcField;
  4. private int srcFieldIndex;
  5. private SeaTunnelDataType srcFieldDataType;
  6. private String destField;
  7. @Override
  8. public String getPluginName() {
  9. return "Copy";
  10. }
  11. @Override
  12. protected void setConfig(Config pluginConfig) {
  13. this.srcField = pluginConfig.getString("src_field");
  14. this.destField = pluginConfig.getString("dest_fields");
  15. }
  16. @Override
  17. protected void setInputRowType(SeaTunnelRowType inputRowType) {
  18. srcFieldIndex = inputRowType.indexOf(srcField);
  19. srcFieldDataType = inputRowType.getFieldType(srcFieldIndex);
  20. }
  21. @Override
  22. protected String getOutputFieldName() {
  23. return destField;
  24. }
  25. @Override
  26. protected SeaTunnelDataType getOutputFieldDataType() {
  27. return srcFieldDataType;
  28. }
  29. @Override
  30. protected Object getOutputFieldValue(SeaTunnelRowAccessor inputRow) {
  31. return inputRow.getField(srcFieldIndex);
  32. }
  33. }
  1. The getPluginName method is used to identify the transform name.
  2. The @AutoService is used to generate the META-INF/services/org.apache.seatunnel.api.transform.SeaTunnelTransform file automatically.
  3. The setConfig method is used to inject user configs.

Transform Test Tool

Once you add a new plugin, it is recommended to add e2e tests for it. We have a seatunnel-e2e/seatunnel-transforms-v2-e2e module to help you to do this.

For example, if you want to add an e2e test for CopyFieldTransform, you can create a new test in seatunnel-e2e/seatunnel-transforms-v2-e2e module and extend the TestSuiteBase class in the test.

  1. public class TestCopyFieldTransformIT extends TestSuiteBase {
  2. @TestTemplate
  3. public void testCopyFieldTransform(TestContainer container) {
  4. Container.ExecResult execResult = container.executeJob("/copy_transform.conf");
  5. Assertions.assertEquals(0, execResult.getExitCode());
  6. }
  7. }

Once your testcase implements the TestSuiteBase interface and use @TestTemplate annotation startup, it will running job to all engines, and you just need to execute the executeJob method with your SeaTunnel configuration file, it will submit the SeaTunnel job.