Apache Camel 経由で Elasticsearch にデータを取り込む方法

実際の例を使用して、Apache Camel を介して Elasticsearch にデータを取り込む方法を学びます。

2024年9月09日

ジャンプ

Apache Camel を使用した Elasticsearch へのデータ取り込みは、検索エンジンの堅牢性と統合フレームワークの柔軟性を組み合わせたプロセスです。この記事では、Apache Camel が Elasticsearch へのデータ取り込みを簡素化し、最適化する方法を説明します。この機能を説明するために、Apache Camel を設定して使用し、Elasticsearch にデータを送信する方法を段階的に説明する入門アプリケーションを実装します。

Apache Camel とは何ですか?

Apache Camel は、さまざまなシステムの接続を簡素化するオープンソースの統合フレームワークであり、開発者はシステム通信の複雑さを気にすることなくビジネスロジックに集中できます。Camel の中心的な概念は「ルート」です。これは、メッセージが送信元から送信先までたどるパスを定義し、変換、検証、フィルタリングなどの中間ステップが含まれる可能性があります。

Apache Camel アーキテクチャ

Camel は、データベースやメッセージングサービスなどのさまざまなシステムやプロトコルに接続するために「コンポーネント」を使用し、メッセージの入口と出口を表すために「エンドポイント」を使用します。これらのコンセプトは、モジュール式で柔軟な設計を提供し、複雑な統合を効率的かつスケーラブルに構成および管理することを容易にします。

ElasticsearchとApache Camelの使用

Apache Camel を使用して Elasticsearch クラスターにデータを取り込むシンプルな Java アプリケーションを構成する方法を説明します。Apache Camel で定義されたルートを使用して Elasticsearch でデータを作成、更新、削除するプロセスについても説明します。

1. 依存関係の追加

この統合を構成する最初のステップは、プロジェクトのpom.xmlファイルに必要な依存関係を追加することです。これには、Apache Camel および Elasticsearch ライブラリが含まれます。新しい Java API クライアントライブラリを使用するため、 camel-elasticsearchコンポーネントをインポートし、バージョンをcamel-coreライブラリと同じにする必要があります。

Java 低レベル Rest クライアントを使用する場合は、Elasticsearch 低レベル Rest クライアントコンポーネントを使用する必要があります。

<dependency>
   <groupId>org.apache.camel</groupId>
   <artifactId>camel-core</artifactId>
   <version>4.7.0</version>
</dependency>

<dependency>
   <groupId>org.apache.camel</groupId>
   <artifactId>camel-elasticsearch</artifactId>
   <version>4.7.0</version>
</dependency>

<dependency>
   <groupId>org.apache.camel</groupId>
   <artifactId>camel-jackson</artifactId>
   <version>4.7.0</version>
</dependency>

<dependency>
   <groupId>co.elastic.clients</groupId>
   <artifactId>elasticsearch-java</artifactId>
   <version>8.14.3</version>
</dependency>

2. Camel Context の設定と実行

設定は、ルートの定義と実行のベースとして機能するDefaultCamelContextクラスを使用して新しい Camel コンテキストを作成することから始まります。次に、Elasticsearch コンポーネントを構成して、Apache Camel が Elasticsearch クラスターと対話できるようにします。ESlasticsearchComponentインスタンスは、ローカル Elasticsearch クラスターのデフォルトアドレスであるアドレスlocalhost:9200に接続するように構成されています。認証が必要な環境設定の場合は、 「コンポーネントを構成して基本認証を有効にする」と呼ばれる、コンポーネントを構成して基本認証を有効にする方法に関するドキュメントを読む必要があります。

public class ESComponent {

    public static ElasticsearchComponent getInstance() {
        var elasticsearch = new ElasticsearchComponent();
        elasticsearch.setHostAddresses("localhost:9200");
        return elasticsearch;
    }

    public static String getName() {
        return "elasticsearch";
    }
}

このコンポーネントは Camel コンテキストに追加され、定義されたルートがこのコンポーネントを使用して Elasticsearch で操作を実行できるようになります。

try (var context = new DefaultCamelContext()) {
   context.addComponent(ESComponent.getName(), ESComponent.getInstance());
   context.addRoutes(new OperationBulkRoute());
   context.start();
}

その後、ルートがコンテキストに追加されます。ドキュメントの一括インデックス作成、更新、削除のためのルートを作成します。

3. Camelルートの設定

データのインデックス作成

最初に設定するルートは、データのインデックス作成用です。映画カタログを含む JSON ファイルを使用します。ルートは、 src/main/resources/movies.jsonにあるファイルを読み取り、JSON コンテンツを Java オブジェクトに逆シリアル化し、集約戦略を適用して複数のメッセージを 1 つに結合し、Elasticsearch でのバッチ操作を可能にするように構成されます。メッセージあたり 500 項目のサイズが設定されました。つまり、一括処理により一度に 500 本の映画がインデックス化されます。

ルートElasticsearch操作一括

String URI_BULK_OPERATION = String
       .format("elasticsearch://elasticsearch?operation=%s&indexName=%s",
               IndexOperationConfig.BULK_OPERATION,
               INDEX_NAME);

public class OperationBulkRoute extends RouteBuilder {
   private static final Log log = LogFactory.getLog(OperationBulkRoute.class);
   private static final int BULK_SIZE = 500;

   @Override
   public void configure() {
       from("file:src/main/resources?fileName=movies.json&noop=true")
               .routeId("route-bulk-ingest")
               .unmarshal().json()
               .split(body())
               .aggregate(constant(true), new BulkAggregationStrategy())
               .completionSize(BULK_SIZE)
               .to(URI_BULK_OPERATION)
               .process(exchange -> {
                   var body = exchange.getIn().getBody(String.class);
                   log.info(String.format("Response: %s", body));
               })
               .end();
   }
}

ドキュメントのバッチは、Elasticsearch の一括操作エンドポイントに送信されます。このアプローチにより、大量のデータを処理する際の効率と速度が保証されます。

データ更新

次のルートはドキュメントを更新することです。前の手順でいくつかの映画のインデックスを作成しました。ここでは、参照コードでドキュメントを検索し、評価フィールドを更新するための新しいルートを作成します。

Camel コンテキスト(DefaultCamelContext)を設定し、Elasticsearch コンポーネントを登録し、カスタムルート IngestionRoute を追加します。操作は、ProducerTemplate を介してドキュメントコードを送信することから開始され、direct:update-ingestion エンドポイントからルートが開始されます。

try (var context = new DefaultCamelContext()) {
    context.addComponent(ESComponent.getName(), ESComponent.getInstance());
    context.addRoutes(new IngestionRoute());
    context.start();
    ProducerTemplate producerTemplate = context.createProducerTemplate();
    producerTemplate.sendBody("direct:update-ingestion", documentCode);
    Thread.sleep(5000);
}

次に、このフローの入力エンドポイントである IngestionRoute があります。ルートはいくつかのパイプライン操作を実行します。まず、Elasticsearch で検索が行われ、コード(direct:search-by-id)でドキュメントが検索されます。ここで、SearchByCodeProcessor がコードに基づいてクエリを組み立てます。次に、取得されたドキュメントは UpdateRatingProcessor によって処理され、その結果が Movie オブジェクトに変換され、映画の評価が特定の値に更新され、更新されたドキュメントが Elasticsearch に送り返されて更新されるように準備されます。

public class IngestionRoute extends RouteBuilder {
    private static final Log log = LogFactory.getLog(IngestionRoute.class);

    @Override
    public void configure() throws Exception {

        from("direct:update-ingestion")
                .pipeline()
                .to("direct:search-by-id")
                .to(URI_SEARCH_OPERATION)
                .to("direct:update-rating")
                .to(URI_UPDATE_OPERATION)
                .process(exchange -> {
                    var body = exchange.getIn().getBody(String.class);
                    log.info(String.format("Response: %s", body));
                })
                .end();

        from("direct:search-by-id")
                .process(new SearchByCodeProcessor());

        from("direct:update-rating")
                .process(new UpdateRatingProcessor());
    }
}

SearchByCodeProcessorプロセッサは検索クエリを実行するためだけに構成されました:

public class SearchByCodeProcessor implements Processor {
    @Override
    public void process(Exchange exchange) throws Exception {
        var code = exchange.getIn().getBody();

        String query = "{\n" +
                "  \"query\": {\n" +
                "   \"term\": {\n" +
                "     \"code\": {\n" +
                "       \"value\":" + code + "\n" +
                "     }\n" +
                "   }\n" +
                "  }\n" +
                "}";
        exchange.setProperty("document_code", code);
        exchange.getIn().setBody(query);
    }
}

UpdateRatingProcessorプロセッサは評価フィールドの更新を担当します。

public class UpdateRatingProcessor implements Processor {

    private final ObjectMapper objectMapper;

    public UpdateRatingProcessor() {
        this.objectMapper = new ObjectMapper();
        this.objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
    }

    @Override
    public void process(Exchange exchange) throws Exception {

        HitsMetadata response = exchange.getIn().getBody(HitsMetadata.class);
        var code = Long.parseLong(exchange.getProperty("document_code").toString());

        if (response != null && response.hits() != null) {

            var documents = parseToMovies(response);

            var optionalMovie = documents.stream()
                    .filter(document -> code == (document.getSource().getCode())).findAny();

            optionalMovie.ifPresent(document -> {
                document.getSource().setRating(13.0);
                Map<String, Object> updateMap = new HashMap<>();
                updateMap.put("doc", document.getSource());
                exchange.getIn().setHeader("indexId", document.getId());
                exchange.getIn().setBody(updateMap);
            });
        }
    }

データの削除

最後に、ドキュメントを削除するためのルートが設定されます。ここでは、ID を使用してドキュメントを削除します。Elasticsearch でドキュメントを削除するには、ドキュメント ID、ドキュメントが保存されているインデックスを知って、削除リクエストを実行する必要があります。Apache Camel では、以下に示すように新しいルートを作成してこの操作を実行します。

ルートは、エントリポイントとして機能する direct:op-delete エンドポイントから始まります。ドキュメントを削除する必要がある場合、その識別子(_id)メッセージの本文で受信されます。次に、ルートは単純な("${body}")を使用してこの識別子の値で indexId ヘッダーを設定し、メッセージの本文から _id を抽出します。

public class OperationDeleteRoute extends RouteBuilder {
   private static final Log log = LogFactory.getLog(OperationDeleteRoute.class);

   @Override
   public void configure() {
       from("direct:op-delete")
               .routeId("route-delete")
               .setHeader("indexId", simple("${body}"))
               .to(URI_DELETE_OPERATION)
               .process(exchange -> {
                   var body = exchange.getIn().getBody(String.class);
                   log.info(String.format("Response: %s", body));
               })
               .end();
       ;
   }
}

String URI_DELETE_OPERATION = String
       .format("elasticsearch://elasticsearch?operation=%s&indexName=%s",
               IndexOperationConfig.DELETE_OPERATION,
               INDEX_NAME);

最後に、メッセージは URI_DELETE_OPERATION で指定されたエンドポイントに送信され、Elasticsearch に接続して対応するインデックスでドキュメント削除操作を実行します。ルートを作成したので、Elasticsearch コンポーネントを含めるように構成された Camel コンテキスト(DefaultCamelContext)を作成できます。

try (var context = new DefaultCamelContext()) {
   context.addComponent(ESComponent.getName(), ESComponent.getInstance());
   context.addRoutes(new OperationDeleteRoute());
   context.start();
   ProducerTemplate producerTemplate = context.createProducerTemplate();
   producerTemplate.sendBody("direct:op-delete", documentId);
}

次に、 OperationDeleteRouteクラスによって定義された削除ルートがコンテキストに追加されます。コンテキストが初期化されると、 ProducerTemplateを使用して、削除するドキュメントの識別子がdirect:op-deleteエンドポイントに渡され、削除ルートがトリガーされます。

まとめ

Apache Camel と Elasticsearch の統合により、Camel の柔軟性を活用して、インデックス作成、更新、削除などのさまざまなデータ操作シナリオを処理できるルートを定義できるため、堅牢で効率的なデータ取り込みが可能になります。この設定により、複雑なプロセスをスケーラブルにオーケストレーションおよび自動化できるため、Elasticsearch でデータが効率的に管理されるようになります。この例では、これらのツールを組み合わせて使用して、データ取り込みのための効率的で適応性の高いソリューションを作成する方法を示しました。

参照資料