Apache InLong ManagerにおけるSPI拡張の実践

Apache InLong (应龙) は、データ収集、集約、ストレージ、およびデータ処理の全プロセスを統合する、オールインワンの大量データ統合フレームワークです。自動化、セキュリティ、信頼性、高性能なデータ転送機能を提供し、ストリームベースのデータ分析、モデリング、およびアプリケーションの構築を容易にします。

InLong Manager の役割

InLong は、データの収集、集約、キャッシュ、および選別機能をサポートしています。ユーザーは簡単な設定を行うだけで、ソースからリアルタイム計算エンジンにデータをインポートしたり、オフラインストレージシステムに書き込んだりできます。InLong システムは、収集、集約、キャッシュ、選別の 4 つの主要なプロセスで構成されています。これらのプロセスを連携させる役割を担っているのが、InLong Manager です。Manager は、システムとタスクのメタデータを管理し、タスクの全プロセスを連携させます。

メタデータには、InLong システム内のユーザー情報、承認情報、クラスター構成情報、ユーザーが作成したデータソース、データターゲット、およびデータスキーマなどの情報が含まれます。

ユーザーは、InLong Dashboard で提供される Web UI または Manager Client を通じてデータストリームタスクを作成できます。タスクが承認されると、MQ の Topic とコンシューマーの作成、ターゲットデータベース/テーブル構造の作成、Flink タスクの起動、MQ からのデータ消費、およびターゲットへの書き込みといった一連のプロセスが連携されます。

InLong Manager の SPI 実践

課題

InLong は元々 Tencent の社内システムから発展したため、当初は限られたデータソースとストレージシステムのみをサポートしていました。ストレージシステムの種類が増え、クラウドへの移行が進むにつれて、データソースとターゲットのタイプが急増しました。この拡張プロセスで、従来の設計におけるいくつかの問題点が浮き彫りになりました。

テーブル数の増加と重複フィールド: 各ストレージタイプごとにテーブルが作成され、ステータス、作成者、更新者、削除フラグ、タイムスタンプ、バージョン番号などの共通フィールドが多数存在し、メンテナンスコストが増大しました。
類似コードの多用: 保存操作（MQ リソースの作成、ストレージ側テーブル構造の作成、タスクの配信など）において、大量の if-else や switch-case 文による条件分岐が発生し、コードの可読性と保守性が低下しました。
拡張性の低さ (開発生産原則違反): 新しいストレージタイプを拡張する場合、新しいテーブルの追加 (DDL の頻繁な変更は避けるべき) に加え、既存のコードに侵入して else/case 文を追加し、特殊なロジックを実装する必要がありました。これは開発生産原則に反します。

SPI (Service Provider Interface) とは

SPI は、Java が提供する API の拡張ポイントであり、サードパーティによる実装や拡張を可能にします。これにより、フレームワークの拡張やコンポーネントの置き換えが可能になります。JDBC ドライバーのロード、SLF4J によるログ実装の選択、Spring の型変換など、多くの Java フレームワークで利用されています。

Java SPI は、基本的に「インターフェースベースのプログラミング + Strategy パターン + 設定ファイル」を組み合わせた動的なロードメカニズムです。

SPI の注意点:

ServiceLoader は遅延ロードですが、必要な実装を取得するためにすべての実装クラスを走査し、すべてをロードおよびインスタンス化する可能性があります。
マルチスレッド環境で ServiceLoader クラスに同時にアクセスすると、スレッドセーフの問題が発生する可能性があります。

Manager の SPI 改造実践

SPI の思想に基づき、InLong Manager のコードを改造しました。

サービス層コードの簡略化:

冗長な if-else / switch-case 文を削除しました。同一ドメインモデルのリクエストは、単一の Service インターフェースで処理されます。例えば、保存操作では、シンクタイプに基づいてオペレーターファクトリーから具体的な設定オペレーターを取得し、そのオペレーターが実際の保存処理を実行します。

SpringBoot 環境では、Spring フレームワーク自体の Beanロードメカニズムが SPI と同様の機能を提供するため、明示的な SPI インターフェースを使用する必要はありません。

// org.apache.inlong.manager.service.sink.StreamSinkServiceImpl
public Integer save(SinkRequest request, String operator) {
    // ...
    // 根据 sink 类型保存 sink 信息
    StreamSinkOperator sinkOperator = operatorFactory.getInstance(request.getSinkType());
    return sinkOperator.saveOpt(request, operator);
}

// org.apache.inlong.manager.service.sink.SinkOperatorFactory
@Service
public class SinkOperatorFactory {

    // Spring 自动加载 StreamSinkOperator 接口的所有实现者，与 SPI 要达到的效果相同
    @Autowired
    private List<StreamSinkOperator> sinkOperatorList;

    /**
     * 通过给定的 sinkType 获取 sink 操作实例
     */
    public StreamSinkOperator getInstance(String sinkType) {
        return sinkOperatorList.stream()
                .filter(inst -> inst.accept(sinkType))
                .findFirst()
                .orElseThrow(() -> new BusinessException(ErrorCodeEnum.SINK_TYPE_NOT_SUPPORT,
                        String.format(ErrorCodeEnum.SINK_TYPE_NOT_SUPPORT.getMessage(), sinkType)));
    }
}

データベースエンティティモデルのリファクタリング:

1 つのテーブルで任意のタイプの Sink 設定をサポートするようにしました。テーブルには共通フィールドを記録し、各 Sink タイプ固有のフィールドは JSON 形式の拡張フィールド（ext_params）に格納します。

CREATE TABLE IF NOT EXISTS `stream_sink`
(
    `id`                     int(11)      NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
    `inlong_group_id`        varchar(256) NOT NULL COMMENT 'Owning inlong group id',
    `inlong_stream_id`       varchar(256) NOT NULL COMMENT 'Owning inlong stream id',
    `sink_type`              varchar(15)           DEFAULT 'HIVE' COMMENT 'Sink type, including: HIVE, ES, etc',
    `sink_name`              varchar(128) NOT NULL DEFAULT '' COMMENT 'Sink name',
    `description`            varchar(500) NULL COMMENT 'Sink description',
    `enable_create_resource` tinyint(1)            DEFAULT '1' COMMENT 'Whether to enable create sink resource? 0-disable, 1-enable',
    `data_node_name`         varchar(128)          DEFAULT NULL COMMENT 'Node name, which links to data_node table',
    `ext_params`             text         NULL COMMENT 'Another fields, will be saved as JSON type',
    `operate_log`            text                  DEFAULT NULL COMMENT 'Background operate log',
    `status`                 int(11)               DEFAULT '100' COMMENT 'Status',
    `previous_status`        int(11)               DEFAULT '100' COMMENT 'Previous status',
    `is_deleted`             int(11)               DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
    `creator`                varchar(64)  NOT NULL COMMENT 'Creator name',
    `modifier`               varchar(64)           DEFAULT NULL COMMENT 'Modifier name',
    `create_time`            timestamp    NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Create time',
    `modify_time`            timestamp    NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 'Modify time',
    `version`                int(11)      NOT NULL DEFAULT '1' COMMENT 'Version number, which will be incremented by 1 after modification',
    PRIMARY KEY (`id`),
    UNIQUE KEY `unique_sink_name` (`inlong_group_id`, `inlong_stream_id`, `sink_name`, `is_deleted`)
);

各 Sink タイプ固有のフィールドは JSON 形式に変換され、ext_params フィールドに格納されます。クエリ時には、JSON を解析して固有の DTO に変換します。関連コードは以下の通りです。

org.apache.inlong.manager.service.sink.AbstractSinkOperator の saveOpt メソッドから呼び出される setTargetEntity メソッド
org.apache.inlong.manager.service.sink.StreamSinkOperator インターフェースの getFromEntity メソッドの各実装

その他の箇所での利用:
このアプローチは、データソース (StreamSource)、データターゲット (StreamSink) の設定管理、InlongGroup による異なるタイプの MQ への拡張、MQ トピックとコンシューマーの作成、Sink テーブル構造の作成など、様々な箇所で利用されています。
- org.apache.inlong.manager.service.group.InlongGroupOperator
- org.apache.inlong.manager.service.sink.StreamSinkOperator
- org.apache.inlong.manager.service.source.StreamSourceOperator
- org.apache.inlong.manager.service.resource.sink.SinkResourceOperator
- org.apache.inlong.manager.service.resource.queue.QueueResourceOperator

改造後のメリット

コードの再利用性向上: 大量の重複・類似コードが削減され、メンテナンスコストが低下しました。
拡張性の劇的な向上: 新しいタイプの構成を拡張する場合、既存のインターフェースを変更することなく、独自のロジックを実装するだけで対応できるようになりました。
DDL 変更の頻度低減: テーブルの DDL 変更頻度が減り、メンテナンスコストが低減され、DDL 変更に起因するオンライン問題が回避されました。
OSS への貢献: OSS コードに侵入することなく、Tencent 内部の構成タイプを拡張し、社内固有のビジネスロジックを追加できるようになりました。

タグ: Java SPI Apache InLong Microservices Design Patterns

7月14日 17:29 投稿

異端開発室