Managed Service for Apache Flink へのストリーミングデータソースの追加

Apache Flink には、ファイル、ソケット、コレクション、カスタムソースから読み取るためのコネクタが用意されています。アプリケーションコードでは、「Apache Flink ソース」を使用してストリームからデータを受信します。このセクションでは、Amazon のサービスで使用できるソースについて説明します。

Kinesis Data Streams

FlinkKinesisConsumer ソースは Amazon Kinesis データストリームからアプリケーションにストリーミングデータを提供します。

「FlinkKinesisConsumer」の作成

次のコード例は、FlinkKinesisConsumer の作成を示しています。

Properties inputProperties = new Properties();
inputProperties.setProperty(ConsumerConfigConstants.AWS_REGION, region);
inputProperties.setProperty(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "LATEST");

DataStream<string> input = env.addSource(new FlinkKinesisConsumer<>(inputStreamName, new SimpleStringSchema(), inputProperties));

FlinkKinesisConsumer の使用方法の詳細については、「Apache Flink ストリーミング Java コードをダウンロードして調べる」を参照してください。

EFO コンシューマーを使用する FlinkKinesisConsumer の作成

で拡張ファンアウト (EFO) がサポートされ FlinkKinesisConsumer るようになりました。

Kinesis コンシューマーが EFO を使用する場合、Kinesis Data Streams サービスは、コンシューマーがストリームの固定帯域幅を、ストリームから読み取る他のコンシューマーと共有するのではなく、独自の専用帯域幅を提供します。

Kinesis コンシューマーで EFO を使用する方法の詳細については、FLIP-128: Enhanced Fan Out for AWS Kinesis Consumers」を参照してください。

EFO コンシューマーを有効にするには、Kinesis コンシューマーで次のパラメータを設定します。

• RECORD_PUBLISHER_TYPE: アプリケーションが EFO Consumerを使用して Kinesis Data Streamsデータにアクセスできるようにするには、このパラメータを EFO に設定します。

• EFO_CONSUMER_NAME: このパラメータを、このストリームのコンシューマー間で一意の文字列値に設定します。同じ Kinesis Data Stream でコンシューマー名を再利用すると、その名前を使用していた以前のコンシューマーは終了します。

EFO を使用するように FlinkKinesisConsumer を設定するには、コンシューマーに以下のパラメータを追加します。

consumerConfig.putIfAbsent(RECORD_PUBLISHER_TYPE, "EFO");
consumerConfig.putIfAbsent(EFO_CONSUMER_NAME, "basic-efo-flink-app");

EFO コンシューマーを使用する Apache Flink 用 Managed Serviceの例については、 EFO Consumer を参照してください。

Amazon MSK

KafkaSource ソースは Amazon MSK トピックからアプリケーションにストリーミングデータを提供します。

「KafkaSource」の作成

次のコード例は、KafkaSource の作成を示しています。

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers(brokers)
    .setTopics("input-topic")
    .setGroupId("my-group")
    .setStartingOffsets(OffsetsInitializer.earliest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

KafkaSource の使用方法の詳細については、「MSK レプリケーション」を参照してください。