ETL コンポーネント

千桐テクノロジー2026/06/11約5分デベロップメントドキュメント

ヒント

さまざまな業務シナリオにおける ETL データ処理の要件を満たすため、システムは Reader（入力）、Transition（変換）、Writer（出力）の 3 種類のコンポーネントを提供します。
本記事では、ETL コンポーネントの全体構造、ソースコードディレクトリ、拡張方法を紹介し、デベロップメント者が新しいコンポーネントの接続とパラメータのカプセル化をすばやく完了できるようにします。

定義（ETL）

本モジュールは、データ統合（ETL）タスクを設定および実行するために使用します。複数データソースの接続、変換、書き出しをサポートし、データフローやデータ処理チェーンを構築するための中核コンポーネントです。ノードベースのグラフィカル設定により、ユーザーはデータの読み取り、クレンジング、変換から保存までの一連のプロセスを完了できます。

全体構造

実行メカニズム

現在の ETL コンポーネントは DolphinScheduler によるスケジューリングと実行に依存しています。全体の実行メカニズムは次のとおりです。

ユーザーはフロントエンド画面でドラッグ操作により、入力、変換、出力を設定します。
システムは設定パラメータをカプセル化してパッケージ化します。
DolphinScheduler と連携し、タスクを作るして保存します。
タスク情報はローカルテーブルと DolphinScheduler スケジューラの両方に保存されます。
実行段階では DolphinScheduler がトリガーし、統合済みの ETL JAR パッケージ を呼び出してタスクを実行します。

変換コンポーネントの拡張

変換コンポーネントを加えるするには、次の 3 つの層を変更する必要があります。

フロントエンド画面：ノード設定 UI を加えるします。
バックエンドコードのカプセル化：パラメータのシリアライズと受け渡しをサポートします。
ETL JAR パッケージのソースコード：具体的な変換コンポーネントの実装を加えるします。

入力コンポーネントと出力コンポーネントの拡張

（一）入力データソースの拡張

データソース拡張ドキュメント を参照して、新しいデータソースの拡張を完了します。
画面、パラメータのカプセル化ロジック、ETL JAR パッケージのソースコード を変更し、変換拡張と整合させます。

（二）非データソース入力コンポーネントの拡張（例：Excel ファイル）

ファイル取得：まず Excel ファイルのアップロードと保存を完了します。
パラメータのカプセル化：ファイルのメタ情報を記録し、コンポーネントへ渡します。
Reader の新規作る：Excel 読み取りロジックを作るし、データを DataSet に変換します。
ETL プロセスへの参加：通常の Reader コンポーネントのフローと一致させます。

ソースコードディレクトリ構造

モジュール位置

ETL コンポーネントのソースコードは qdata-etl モジュール配下にあります。

tech.qiantong.qdata.spark.etl

ディレクトリ構造

reader：入力コンポーネント（データソース読み取りロジック）
transition：変換コンポーネント（データクレンジングと変換ロジック）
writer：出力コンポーネント（データ書き込みロジック）
utils：ユーティリティクラス（共通メソッド、補助関数）
EtlApplication：ETL タスク実行エントリ

設置説明

入力を拡張 → reader を変更
変換を拡張 → transition を変更
出力を拡張 → writer を変更
共通ユーティリティメソッド → utils を使用
実行エントリロジック → EtlApplication を参照

入力コンポーネント（Reader）

位置

入力コンポーネントは次にあります。

tech.qiantong.qdata.spark.etl.reader

主なクラスは次のとおりです。

Reader：入力コンポーネントのインターフェースで、統一された読み取りメソッドを定義します。
ReaderFactory：入力コンポーネントのファクトリで、コンポーネントの code に基づいて具体的な実装を取得します。
ReaderRegistry：入力コンポーネントのレジストリで、登録済み Reader のマッピングを管理します。

Reader インターフェース

public interface Reader {
    Dataset<Row> read(SparkSession spark, JSONObject reader, List<String> readerColumns, String logPath);
    String code();
}

•	read：入力元からデータを読み取り、`Dataset<Row>` を返す中核メソッド
•	code：入力コンポーネントのタイプを識別します

ReaderFactory ファクトリクラス

コンポーネントコードにより具体的な Reader インスタンスを取得します。

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

ReaderRegistry レジストリ

システムに登録済みの Reader は次のとおりです。

DBReader（データベース入力）
ExcelReader（Excel ファイル入力）
CsvReader（CSV ファイル入力）
KafkaReader（Kafka メッセージストリーム入力）

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);
    }
}

拡張方法

Reader の拡張には、主に次の 2 つの方法があります。

データソースの拡張（DBReader）
- DBReader 内に、対象データベースに対応する JDBC 互換処理を加えるします。
- MySQL、Oracle、SQL Server、Dameng などのリレーショナルデータベースに適しています。
非データソースの拡張（Excel ファイルなど）
- Reader インターフェースを実装する XXXReader クラスを新規作るします。
- ReaderRegistry に新しい Reader を登録します。
- read メソッドを実装し、データ読み込みを完了します。

例：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;
}

変換コンポーネント（Transition）

位置

変換コンポーネントは次にあります。

tech.qiantong.qdata.spark.etl.transition

現在のシステム組み込み例は次のとおりです。

CleanTransition（クレンジングコンポーネント）。拡張方法は クレンジングデベロップメントドキュメント を参照できます。

現在の実装の制限

EtlApplication では、変換ロジックが次のように ハードコード されています。

最初の変換コンポーネント のみを取得します：transitionArr.get(0)
デフォルトで CleanTransition を呼び出して実行します

サンプルコード：

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);
}

拡張方法

実際の業務では、すべての変換コンポーネントを CleanTransition に積み上げるべきではありません。拡張メカニズムにより複数コンポーネントをサポートする必要があります。
具体的な手順は次のとおりです。

Transition インターフェースの抽象化
拡張しやすいように、統一された変換仕様を定義します。

   public interface Transition {
       Dataset<Row> transform(Dataset<Row> dataset, JSONObject transition, String logPath);
       String code();
   }

Transition 実装クラスの加える

例：MappingTransition（フィールドマッピング）、SplitTransition（フィールド分割）、DerivationTransition（派生フィールド）など。
各実装クラスは単一の変換ロジックを担当し、責務を明確に保ちます。

ファクトリ + レジストリパターンの導入

Reader / Writer と同様に、TransitionFactory と TransitionRegistry を作るします。
設定内の code に基づいて、対応する変換コンポーネントインスタンスを動的に取得します。

EtlApplication の改造

元のハードコードされた CleanTransition をファクトリパターンに置き換えます。
transitionArr を走査し、異なる変換コンポーネントを順番に呼び出します。
タスクログ、例外キャッチ、タスク状態アップデートを既存ロジックと一致させます。

出力コンポーネント（Writer）

位置

出力コンポーネントは次にあります。

    tech.qiantong.qdata.spark.etl.writer

主なクラス説明

Writer インターフェース（Writer.java）
出力コンポーネントの統一仕様を定義し、すべての Writer 実装クラスにデータ書き出しメソッドとコンポーネントタイプ識別子の提供を求めます。
WriterFactory ファクトリクラス（WriterFactory.java）
コンポーネントの code に基づいて具体的な Writer 実装を取得し、外部呼び出し側が具体的なクラス名を意識しなくてよいようにします。
WriterRegistry レジストリ（WriterRegistry.java）
登録済み Writer コンポーネントインスタンスのマッピングを管理します。現在は DBWriter（データベース書き込み）のみ登録されています。

拡張方法

Writer インターフェースを実装するクラスを加えるし、Hive、HDFS、Kafka などへの書き込みといった対応する出力ロジックを完了します。
WriterRegistry に新しいコンポーネントを登録します。
WriterFactory を通じて統一的に取得し、使用します。

まとめ

Writer は Reader と同じ構造を保ち、インターフェース + ファクトリ + レジストリ パターンに従うため、拡張と管理が容易です。
現在のバージョンではデータベース書き込みのみが公開されています。デベロップメント者は業務要件に応じて新しい出力コンポーネントを拡張できます。

業務コードとパラメータカプセル化の拡張

位置

業務コードのパラメータカプセル化ロジックは次にあります。

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

主なメソッド：

createEtlTask：ETL タスク作るエントリ
TaskConverter.buildEtlTaskParams：ETL プログラムに必要なパラメータを構築します

パラメータフロー

フロントエンド画面：ユーザーが入力、変換、出力ノードを設定し、パラメータを生成します。
バックエンド業務サービス：パラメータを受け取り、TaskConverter を呼び出してカプセル化します。
DolphinScheduler：スケジューラがタスクを保存し、スケジューリングします。
ETL コンポーネント：カプセル化されたパラメータに基づいて、具体的な Reader / Transition / Writer を実行します。

TaskConverter.buildEtlTaskParams

このメソッドは、コンポーネントタイプに基づいてパラメータを振り分け、カプセル化します。

Reader ノード：DB_READER、EXCEL_READER、CSV_READER などを result.reader に入れます。
Transition ノード：SPARK_CLEAN、SORT_RECORD、FIELD_DERIVATION などを transitionList に加えるします。
Writer ノード：DB_WRITER を result.writer に入れます。
Config ノード：タスクの基本情報、リソースパス、メッセージキュー設定などを保存します。

最終的に返される結果構造は次のとおりです。

reader：入力コンポーネントパラメータ
transition：変換コンポーネントパラメータリスト
writer：出力コンポーネントパラメータ
config：共通設定

拡張方法

新しいコンポーネントを加えるする場合は、次が必要です。

フロントエンドで：パラメータ設定と受け渡しをサポートします。
バックエンドで：
- TaskComponentTypeEnum に新しいコンポーネントタイプを定義します。
- TaskConverter.buildEtlTaskParams の switch-case に分岐ロジックを加えるします。
- 対応するファクトリメソッドを呼び出してパラメータをカプセル化します。
スケジューリングと実行層で：DolphinScheduler がタスクを正しくスケジューリングでき、ETL コンポーネントがパラメータを解析して実行できることを保証します。

まとめ

業務コード部分は、コンポーネントとスケジューリングの間の橋渡し です。
Reader / Transition / Writer の 3 端点と TaskConverter のパラメータカプセル化ロジックが一致している場合にのみ、コンポーネントを正しく呼び出せます。

注意事項

パラメータの一貫性
- フロントエンドで設定するパラメータフィールドは、バックエンドの TaskConverter カプセル化ロジックと一致している必要があります。
- コンポーネントを加えるする際は、実行時のパラメータ欠落や解析エラーを避けるため、パラメータ名と解析方法が統一されていることを必ず確認してください。
コンポーネント登録
- Reader、Transition、Writer はいずれも インターフェース + ファクトリ + レジストリ パターンを採用しています。
- 新しいコンポーネントは対応する Registry に登録する必要があります。登録しないと、ファクトリから取得できません。
タスクチェーンの完全性
- ETL タスクには入力（Reader）と出力（Writer）が必須で、変換（Transition）は任意です。
- パラメータをカプセル化する際は、3 つの部分のデータ構造がそろっていることを確認し、タスク実行チェーンを円滑に保ちます。
例外処理
- 変換と書き込みの段階ではエラーが発生しやすいため、コンポーネント拡張時にはログ記録と例外キャッチを加えるする必要があります。
- 例外発生時にタスク状態をアップデートできるようにし、スケジューラが停止状態になることを避けます。
性能とリソース制御
- 読み取り、変換、書き込みの過程では Spark Dataset 操作が関係するため、メモリとパーティション処理に注意する必要があります。
- 大きなファイルや高並行タスクでは、OOM を避けるため Spark パラメータを適切に設定する必要があります。
拡張ドキュメント
- データソースを加えるする場合は、まず データ接続デベロップメントドキュメント を参照してください。
- クレンジングまたは変換コンポーネントを加えるする場合は、クレンジングルールデベロップメントドキュメント を参照し、拡張仕様の一貫性を保ってください。

総括

ETL コンポーネントの拡張は、単にコード実装を加えるするだけではありません。より重要なのは、フロントエンドパラメータ → バックエンドカプセル化 → スケジューラのスケジューリング → コンポーネント実行 という全体チェーンの一貫性と健全性を保証することです。