Skip to main content

ETL Components

Qiantong Tech6/11/26About 6 minDevelopment Docs

Tip

To meet ETL data processing needs in different business scenarios, the system provides three types of components: Reader (input), Transition (transformation), and Writer (output).
This article introduces the overall structure, source code directories, and extension methods of ETL components, helping developers quickly connect new components and encapsulate parameters.

I. Definition (ETL)

This module is used to configure and execute data integration (ETL) tasks. It supports access, transformation, and output for multiple data sources, and is the core component for building data flows and data processing pipelines. Through node-based graphical configuration, users can complete the full process from data reading, cleansing, and transformation to final storage.

II. Overall Structure

Runtime Mechanism

The current ETL components rely on DolphinScheduler for scheduling and execution. The overall runtime mechanism is as follows:

  1. Users configure input, transformation, and output nodes on the frontend page by dragging and dropping.
  2. The system encapsulates and packages the configuration parameters.
  3. The system interacts with DolphinScheduler to create and store tasks.
  4. Task information is stored in both the local table and the DolphinScheduler scheduler.
  5. During execution, DolphinScheduler triggers the task and calls the integrated ETL JAR package to execute it.

Extending Transformation Components

To add a transformation component, three layers need to be modified:

  1. Frontend page: Add the node configuration UI.
  2. Backend code encapsulation: Support parameter serialization and passing.
  3. ETL JAR package source code: Add the concrete transformation component implementation.

Extending Input and Output Components

(1) Extending Input Data Sources

  1. Refer to Data Source Extension Documentation to complete new data source extension.
  2. Modify the page, parameter encapsulation logic, and ETL JAR package source code, keeping them consistent with transformation extension.

(2) Extending Non-Data-Source Input Components (Example: Excel File)

  1. File acquisition: Complete Excel file upload and storage first.
  2. Parameter encapsulation: Record file metadata and pass it to the component.
  3. Create a new Reader: Write Excel reading logic and convert the data into a DataSet.
  4. Enter the ETL process: Keep the flow consistent with regular Reader components.

III. Source Code Directory Structure

Module Location

The ETL component source code is located under the qdata-etl module:

tech.qiantong.qdata.spark.etl

Directory Structure

  • reader: Input components (data source reading logic)
  • transition: Transformation components (data cleansing and transformation logic)
  • writer: Output components (data writing logic)
  • utils: Utility classes (common methods and helper functions)
  • EtlApplication: ETL task execution entry point

Location Notes

  • Extend input → modify reader
  • Extend transformation → modify transition
  • Extend output → modify writer
  • Common utility methods → use utils
  • Execution entry logic → see EtlApplication

IV. Input Components (Reader)

Location

Input components are located in:

tech.qiantong.qdata.spark.etl.reader

Core classes include:

  • Reader: Input component interface, defining a unified read method.
  • ReaderFactory: Input component factory, obtaining the concrete implementation based on the component code.
  • ReaderRegistry: Input component registry, maintaining registered Reader mappings.

Reader Interface

public interface Reader {
    Dataset<Row> read(SparkSession spark, JSONObject reader, List<String> readerColumns, String logPath);
    String code();
}
•	read: Core method responsible for reading data from the input source and returning `Dataset<Row>`
•	code: Identifies the input component type

ReaderFactory Factory Class

Obtain a concrete Reader instance by component code:

public static Reader getReader(String code) {
    return Optional.ofNullable(COMPONENT_ITEM_REGISTRY.getReader(code))
        .orElseThrow(() -> new ServiceException(String.format("%s not supported.", code)));
}

ReaderRegistry Registry

Readers currently registered in the system:

  • DBReader (database input)
  • ExcelReader (Excel file input)
  • CsvReader (CSV file input)
  • KafkaReader (Kafka message stream input)
public class ReaderRegistry {
    private final Map<String, Reader> readerMap = new HashMap<>();

    public ReaderRegistry() {
        this.readerMap.put(TaskComponentTypeEnum.DB_READER.getCode(), new DBReader());
        this.readerMap.put(TaskComponentTypeEnum.EXCEL_READER.getCode(), new ExcelReader());
        this.readerMap.put(TaskComponentTypeEnum.CSV_READER.getCode(), new CsvReader());
        this.readerMap.put(TaskComponentTypeEnum.KAFKA_READER.getCode(), new KafkaReader());
    }

    public Reader getReader(String code) {
        return this.readerMap.get(code);
    }
}

Extension Methods

There are two typical ways to extend Reader:

  1. Extend a data source (DBReader)

    • Add JDBC compatibility for the corresponding database inside DBReader.
    • Suitable for relational databases such as MySQL, Oracle, SQL Server, and Dameng.

  2. Extend a non-data-source input (such as an Excel file)

    • Create a new XXXReader class that implements the Reader interface.
    • Register the new Reader in ReaderRegistry.
    • Implement the read method to complete data loading.

Example: ExcelReader

@Override
public Dataset<Row> read(SparkSession spark, JSONObject reader, List<String> readerColumns, String logPath) {
    LogUtils.writeLog(logPath, "*********************************  Initialize task context  ***********************************");
    LogUtils.writeLog(logPath, "Start Excel input node");
    LogUtils.writeLog(logPath, "Task start time: " + DateUtil.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS"));
    LogUtils.writeLog(logPath, "Task parameters: " + reader.toJSONString(PrettyFormat));

    JSONObject parameter = reader.getJSONObject("parameter");
    List<Object> column = parameter.getJSONArray("column");
    String path = parameter.getString("path");

    Dataset<Row> dataset = spark.read()
            .format("csv")
            .option("header", "true")
            .option("inferSchema", "true")
            .load(path);

    dataset = dataset.select(column.stream()
            .map(c -> new Column(((JSONObject) c).getString("columnName")))
            .toArray(Column[]::new));

    readerColumns.addAll(column.stream()
            .map(c -> ((JSONObject) c).getString("columnName"))
            .collect(Collectors.toList()));

    LogUtils.writeLog(logPath, "Input data count: " + dataset.count());
    dataset.na().fill("Unknown").show(10);
    LogUtils.writeLog(logPath, "Partial data:\n" + dataset.na().fill("Unknown").showString(10, 0, false));
    return dataset;
}

V. Transformation Components (Transition)

Location

Transformation components are located in:

tech.qiantong.qdata.spark.etl.transition

Current built-in system example:

  • CleanTransition (cleansing component); the extension method can refer to Cleansing Development Documentation.

Limitations of the Current Implementation

In EtlApplication, the transformation logic is hard-coded as follows:

  • Only the first transformation component is taken: transitionArr.get(0)
  • CleanTransition is called by default for execution

Example code:

if (taskParams.getJSONArray("transition") != null && taskParams.getJSONArray("transition").size() > 0) {
    JSONArray transitionArr = taskParams.getJSONArray("transition");
    JSONObject transition = (JSONObject) transitionArr.get(0);

    String transitionLogPath = LogUtils.createLogPath(resourceUrl, transition);
    TaskInstance transitionTaskInstance = createTask(processInstance, transitionLogPath, transition, now, rabbitmq);

    try {
        data = CleanTransition.transition(data, transition, transitionLogPath);
        LogUtils.writeLog(transitionLogPath, "Task succeeded");
    } catch (Exception e) {
        updateProcess(processInstance, WorkflowExecutionStatus.FAILURE, rabbitmq);
        updateTask(transitionTaskInstance, TaskExecutionStatus.FAILURE, rabbitmq);
        spark.stop();
        LogUtils.writeLog(transitionLogPath, "Task failed");
        return;
    }
    updateTask(transitionTaskInstance, TaskExecutionStatus.SUCCESS, rabbitmq);
}

Extension Methods

In real business scenarios, transformation components should not all be piled into CleanTransition. They should support multiple components through an extension mechanism.
The specific steps are as follows:

  1. Abstract the Transition interface
    Define a unified transformation specification for easier extension.
   public interface Transition {
       Dataset<Row> transform(Dataset<Row> dataset, JSONObject transition, String logPath);
       String code();
   }
  1. Add Transition implementation classes
  • For example: MappingTransition (field mapping), SplitTransition (field splitting), DerivationTransition (derived fields), and so on.
  • Each implementation class is responsible for a single transformation logic and keeps responsibilities clear.
  1. Introduce the factory + registry pattern
  • Similar to Reader / Writer, create TransitionFactory and TransitionRegistry.
  • Dynamically obtain the corresponding transformation component instance based on the code in the configuration.
  1. Refactor EtlApplication
  • Replace the original hard-coded CleanTransition with the factory pattern.
  • Iterate through transitionArr and call different transformation components in order.
  • Keep task logs, exception handling, task status updates, and the original logic consistent.

VI. Output Components (Writer)

Location

Output components are located in:

    tech.qiantong.qdata.spark.etl.writer

Main Class Descriptions

  • Writer interface (Writer.java)
    Defines the unified specification for output components, requiring all Writer implementation classes to provide methods for writing data and a component type identifier.

  • WriterFactory factory class (WriterFactory.java)
    Obtains the concrete Writer implementation based on the component code, ensuring external callers do not need to care about specific class names.

  • WriterRegistry registry (WriterRegistry.java)
    Maintains mappings of registered Writer component instances. Currently only DBWriter (database writing) is registered.

Extension Methods

  1. Add a class that implements the Writer interface and completes the corresponding output logic, such as writing to Hive, HDFS, Kafka, and so on.
  2. Register the new component in WriterRegistry.
  3. Obtain and use it uniformly through WriterFactory.

Summary

Writer keeps the same structure as Reader and follows the interface + factory + registry pattern, making extension and management easier.
The current version only opens database writing. Developers can extend new output components based on business requirements.

VII. Extending Business Code and Parameter Encapsulation

Location

The parameter encapsulation logic for business code is located in:

        tech.qiantong.qdata.module.dpp.service.etl.impl.DppEtlTaskServiceImpl

Core methods:

  • createEtlTask: ETL task creation entry point
  • TaskConverter.buildEtlTaskParams: Builds the parameters required by the ETL program

Parameter Flow Path

  1. Frontend page: Users configure input, transformation, and output nodes to generate parameters.
  2. Backend business service: Receives parameters and calls TaskConverter for encapsulation.
  3. DolphinScheduler: The scheduler stores and schedules tasks.
  4. ETL components: Execute specific Reader / Transition / Writer logic based on the encapsulated parameters.

TaskConverter.buildEtlTaskParams

This method dispatches and encapsulates parameters based on component type:

  • Reader node: DB_READER, EXCEL_READER, CSV_READER, and so on are placed into result.reader.
  • Transition node: SPARK_CLEAN, SORT_RECORD, FIELD_DERIVATION, and so on are added to transitionList.
  • Writer node: DB_WRITER is placed into result.writer.
  • Config node: Stores basic task information, resource paths, message queue configuration, and so on.

The final returned result structure includes:

  • reader: Input component parameters
  • transition: Transformation component parameter list
  • writer: Output component parameters
  • config: Common configuration

Extension Methods

When adding a new component, you need to:

  1. On the frontend: Support parameter configuration and passing.
  2. On the backend:
    • Define the new component type in TaskComponentTypeEnum.
    • Add branch logic in the switch-case of TaskConverter.buildEtlTaskParams.
    • Call the corresponding factory method for parameter encapsulation.
  3. At the scheduling and execution layer: Ensure DolphinScheduler can correctly schedule the task, and ETL components can parse and execute the parameters.

Summary

The business code section is the bridge between components and scheduling.
Only when the parameter encapsulation logic for Reader / Transition / Writer and TaskConverter stays consistent can components be called correctly.

VIII. Notes

  1. Parameter consistency

    • The parameter fields configured on the frontend must remain consistent with the backend TaskConverter encapsulation logic.
    • When adding a component, be sure to confirm that parameter names and parsing methods are unified to avoid missing parameters or parsing errors during execution.

  2. Component registration

    • Reader, Transition, and Writer all use the interface + factory + registry pattern.
    • New components must be registered in the corresponding Registry; otherwise, the factory cannot obtain them.

  3. Task pipeline completeness

    • An ETL task must include input (Reader) and output (Writer), while transformation (Transition) is optional.
    • When encapsulating parameters, ensure the data structures of all three parts are complete so the task execution pipeline remains smooth.

  4. Exception handling

    • Transformation and writing stages are prone to errors. When extending components, add log recording and exception capture.
    • Ensure task status can be updated when exceptions occur, avoiding a stuck scheduler state.

  5. Performance and resource control

    • Reading, transformation, and writing involve Spark Dataset operations, so pay attention to memory and partition handling.
    • For large files or highly concurrent tasks, configure Spark parameters properly to avoid OOM.

  6. Extension documentation

    • When adding a data source, refer first to Data Connection Development Documentation.
    • When adding cleansing or transformation components, refer to Cleansing Rules Development Documentation to keep extension specifications consistent.

Conclusion

ETL component extension is not just about adding code implementation. More importantly, it ensures consistency and robustness across the full pipeline of frontend parameters → backend encapsulation → scheduler scheduling → component execution.

Last Updated:
qData Open-Source Data Platform
Copyright © 2026 Qiantong Tech