Java で KCL を使用してコンシューマーを開発する

翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

前提条件

KCL 3.x の使用を開始する前に、以下があることを確認してください。

KCL は、ワーカーが実行しているコンピューティングホストから CPU 使用率などの CPU 使用率メトリクスを収集して負荷のバランスを取り、ワーカー間で均等なリソース使用率レベルを実現します。KCL がワーカーから CPU 使用率メトリクスを収集できるようにするには、次の前提条件を満たす必要があります。

Amazon Elastic Compute Cloud(Amazon EC2)

Amazon EC2 での Amazon Elastic Container Service (Amazon ECS)

での Amazon ECS AWS Fargate

Amazon EC2 での Amazon Elastic Kubernetes Service (Amazon EKS)

での Amazon EKS AWS Fargate

依存関係をインストールして追加する

Maven を使用している場合は、次の依存関係を pom.xml ファイルに追加します。3.x.x を最新の KCL バージョンに置き換えてください。

<dependency>
    <groupId>software.amazon.kinesis</groupId>
    <artifactId>amazon-kinesis-client</artifactId>
    <version>3.x.x</version> <!-- Use the latest version -->
</dependency>

Gradle を使用している場合は、build.gradleファイルに以下を追加します。3.x.x を最新の KCL バージョンに置き換えてください。

implementation 'software.amazon.kinesis:amazon-kinesis-client:3.x.x'

Maven Central Repository で KCL の最新バージョンを確認できます。

コンシューマーを実装する

KCL コンシューマーアプリケーションは、次の主要なコンポーネントで構成されます。

RecordProcessor

RecordProcessor は、Kinesis データストリームレコードを処理するためのビジネスロジックが存在するコアコンポーネントです。アプリケーションが Kinesis ストリームから受信したデータを処理する方法を定義します。

主な責任：

実装例を次に示します。

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import software.amazon.kinesis.exceptions.InvalidStateException;
import software.amazon.kinesis.exceptions.ShutdownException;
import software.amazon.kinesis.lifecycle.events.*;
import software.amazon.kinesis.processor.ShardRecordProcessor;

public class SampleRecordProcessor implements ShardRecordProcessor {
    private static final String SHARD_ID_MDC_KEY = "ShardId";
    private static final Logger log = LoggerFactory.getLogger(SampleRecordProcessor.class);
    private String shardId;

    @Override
    public void initialize(InitializationInput initializationInput) {
        shardId = initializationInput.shardId();
        MDC.put(SHARD_ID_MDC_KEY, shardId);
        try {
            log.info("Initializing @ Sequence: {}", initializationInput.extendedSequenceNumber());
        } finally {
            MDC.remove(SHARD_ID_MDC_KEY);
        }
    }

    @Override
    public void processRecords(ProcessRecordsInput processRecordsInput) {
        MDC.put(SHARD_ID_MDC_KEY, shardId);
        try {
            log.info("Processing {} record(s)", processRecordsInput.records().size());
            processRecordsInput.records().forEach(r -> 
                log.info("Processing record pk: {} -- Seq: {}", r.partitionKey(), r.sequenceNumber())
            );
            
            // Checkpoint periodically
            processRecordsInput.checkpointer().checkpoint();
        } catch (Throwable t) {
            log.error("Caught throwable while processing records. Aborting.", t);
        } finally {
            MDC.remove(SHARD_ID_MDC_KEY);
        }
    }

    @Override
    public void leaseLost(LeaseLostInput leaseLostInput) {
        MDC.put(SHARD_ID_MDC_KEY, shardId);
        try {
            log.info("Lost lease, so terminating.");
        } finally {
            MDC.remove(SHARD_ID_MDC_KEY);
        }
    }

    @Override
    public void shardEnded(ShardEndedInput shardEndedInput) {
        MDC.put(SHARD_ID_MDC_KEY, shardId);
        try {
            log.info("Reached shard end checkpointing.");
            shardEndedInput.checkpointer().checkpoint();
        } catch (ShutdownException | InvalidStateException e) {
            log.error("Exception while checkpointing at shard end. Giving up.", e);
        } finally {
            MDC.remove(SHARD_ID_MDC_KEY);
        }
    }

    @Override
    public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) {
        MDC.put(SHARD_ID_MDC_KEY, shardId);
        try {
            log.info("Scheduler is shutting down, checkpointing.");
            shutdownRequestedInput.checkpointer().checkpoint();
        } catch (ShutdownException | InvalidStateException e) {
            log.error("Exception while checkpointing at requested shutdown. Giving up.", e);
        } finally {
            MDC.remove(SHARD_ID_MDC_KEY);
        }
    }
}

以下は、例で使用されている各メソッドの詳細な説明です。

initialize(InitializationInput initializationInput)

processRecords(ProcessRecordsInput processRecordsInput)

leaseLost(LeaseLostInput leaseLostInput)

shardEnded(ShardEndedInput shardEndedInput)

shutdownRequested(ShutdownRequestedInput shutdownRequestedInput)

RecordProcessorFactory

RecordProcessorFactory は、新しい RecordProcessor インスタンスを作成する責任があります。KCL はこのファクトリを使用して、アプリケーションが処理する必要があるシャードごとに新しい RecordProcessor を作成します。

主な責任：

実装例を次に示します。

import software.amazon.kinesis.processor.ShardRecordProcessor;
import software.amazon.kinesis.processor.ShardRecordProcessorFactory;

public class SampleRecordProcessorFactory implements ShardRecordProcessorFactory {
    @Override
    public ShardRecordProcessor shardRecordProcessor() {
        return new SampleRecordProcessor();
    }
}

この例では、ファクトリは shardRecordProcessor() が呼び出されるたびに新しい SampleRecordProcessor を作成します。 shardRecordProcessor これを拡張して、必要な初期化ロジックを含めることができます。

スケジューラー

スケジューラは、KCL アプリケーションのすべてのアクティビティを調整する高レベルのコンポーネントです。データ処理の全体的なオーケストレーションを担当します。

主な責任：

スケジューラは通常、メインアプリケーションで作成および開始されます。スケジューラの実装例は、次のセクションの Main Consumer Application で確認できます。

メインコンシューマーアプリケーション

メインコンシューマーアプリケーションは、すべてのコンポーネントを結び付けます。KCL コンシューマーの設定、必要なクライアントの作成、スケジューラの設定、アプリケーションのライフサイクルの管理を担当します。

主な責任：

実装例を次に示します。

import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.cloudwatch.CloudWatchAsyncClient;
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
import software.amazon.awssdk.services.kinesis.KinesisAsyncClient;
import software.amazon.kinesis.common.ConfigsBuilder;
import software.amazon.kinesis.common.KinesisClientUtil;
import software.amazon.kinesis.coordinator.Scheduler;
import java.util.UUID;

public class SampleConsumer {
    private final String streamName;
    private final Region region;
    private final KinesisAsyncClient kinesisClient;

    public SampleConsumer(String streamName, Region region) {
        this.streamName = streamName;
        this.region = region;
        this.kinesisClient = KinesisClientUtil.createKinesisAsyncClient(KinesisAsyncClient.builder().region(this.region));
    }

    public void run() {
        DynamoDbAsyncClient dynamoDbAsyncClient = DynamoDbAsyncClient.builder().region(region).build();
        CloudWatchAsyncClient cloudWatchClient = CloudWatchAsyncClient.builder().region(region).build();
        
        ConfigsBuilder configsBuilder = new ConfigsBuilder(
            streamName, 
            streamName, 
            kinesisClient, 
            dynamoDbAsyncClient,
            cloudWatchClient, 
            UUID.randomUUID().toString(), 
            new SampleRecordProcessorFactory()
        );

        Scheduler scheduler = new Scheduler(
            configsBuilder.checkpointConfig(),
            configsBuilder.coordinatorConfig(),
            configsBuilder.leaseManagementConfig(),
            configsBuilder.lifecycleConfig(),
            configsBuilder.metricsConfig(),
            configsBuilder.processorConfig(),
            configsBuilder.retrievalConfig()
        );

        Thread schedulerThread = new Thread(scheduler);
        schedulerThread.setDaemon(true);
        schedulerThread.start();
    }

    public static void main(String[] args) {
        String streamName = "your-stream-name"; // replace with your stream name
        Region region = Region.US_EAST_1; // replace with your region
        new SampleConsumer(streamName, region).run();
    }
}

KCL は、デフォルトで専用スループットを持つ拡張ファンアウト (EFO) コンシューマーを作成します。拡張ファンアウトの詳細については、「」を参照してください専用スループットで拡張ファンアウトコンシューマーを開発する。コンシューマーが 2 つ未満の場合、または 200 ミリ秒未満の読み込み伝達遅延を必要としない場合は、共有スループットコンシューマーを使用するようにスケジューラオブジェクトで次の設定を行う必要があります。

configsBuilder.retrievalConfig().retrievalSpecificConfig(new PollingConfig(streamName, kinesisClient))

次のコードは、共有スループットコンシューマーを使用するスケジューラオブジェクトを作成する例です。

インポート：

import software.amazon.kinesis.retrieval.polling.PollingConfig;

コード：

Scheduler scheduler = new Scheduler(
            configsBuilder.checkpointConfig(),
            configsBuilder.coordinatorConfig(),
            configsBuilder.leaseManagementConfig(),
            configsBuilder.lifecycleConfig(),
            configsBuilder.metricsConfig(),
            configsBuilder.processorConfig(),
            configsBuilder.retrievalConfig().retrievalSpecificConfig(new PollingConfig(streamName, kinesisClient))
        );/