Ajouter des sources de données de streaming au service géré pour Apache Flink - Service géré pour Apache Flink
Utiliser les flux de données KinesisUtilisez Amazon MSK

Le service géré Amazon pour Apache Flink était auparavant connu sous le nom d’Amazon Kinesis Data Analytics pour Apache Flink.

Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Ajouter des sources de données de streaming au service géré pour Apache Flink

Apache Flink fournit des connecteurs pour lire à partir de fichiers, de sockets, de collections et de sources personnalisées. Dans le code de votre application, vous utilisez une source Apache Flink pour recevoir les données d’un flux. Cette section décrit les sources disponibles pour les services Amazon.

Utiliser les flux de données Kinesis

La source KinesisStreamsSource fournit des données de streaming à votre application à partir d’un flux de données Amazon Kinesis.

Créer une KinesisStreamsSource

L’exemple de code suivant illustre la création d’un KinesisStreamsSource :

// Configure the KinesisStreamsSource
Configuration sourceConfig = new Configuration();
sourceConfig.set(KinesisSourceConfigOptions.STREAM_INITIAL_POSITION, KinesisSourceConfigOptions.InitialPosition.TRIM_HORIZON); // This is optional, by default connector will read from LATEST

// Create a new KinesisStreamsSource to read from specified Kinesis Stream.
KinesisStreamsSource<String> kdsSource =
        KinesisStreamsSource.<String>builder()
                .setStreamArn("arn:aws:kinesis:us-east-1:123456789012:stream/test-stream")
                .setSourceConfig(sourceConfig)
                .setDeserializationSchema(new SimpleStringSchema())
                .setKinesisShardAssigner(ShardAssignerFactory.uniformShardAssigner()) // This is optional, by default uniformShardAssigner will be used.
                .build();

Pour plus d'informations sur l'utilisation d'unKinesisStreamsSource, consultez le connecteur Amazon Kinesis Data Streams dans la documentation d'Apache Flink et notre exemple KinesisConnectors public sur Github.

Créez un KinesisStreamsSource qui utilise un EFO consommateur

Le supporte KinesisStreamsSource désormais Enhanced Fan-Out () EFO.

Si un client Kinesis l'utiliseEFO, le service Kinesis Data Streams lui fournit sa propre bande passante dédiée, au lieu que le consommateur partage la bande passante fixe du flux avec les autres consommateurs lisant le flux.

Pour plus d'informations sur l'utilisation EFO avec les consommateurs Kinesis, voir FLIP-128 : Enhanced Fan Out for AWS Kinesis Consumers.

Vous activez le EFO consommateur en définissant les paramètres suivants sur le consommateur Kinesis :

  • READER_ TYPE : Définissez ce paramètre sur EFOpour que votre application utilise un EFO consommateur pour accéder aux données Kinesis Data Stream.

  • EFO_ CONSUMER _ NAME : Définissez ce paramètre sur une valeur de chaîne unique parmi les consommateurs de ce flux. La réutilisation d’un nom de consommateur dans le même flux de données Kinesis entraînera la résiliation du client qui utilisait ce nom précédemment.

Pour configurer un KinesisStreamsSource à utiliserEFO, ajoutez les paramètres suivants au consommateur :

sourceConfig.set(KinesisSourceConfigOptions.READER_TYPE, KinesisSourceConfigOptions.ReaderType.EFO);
sourceConfig.set(KinesisSourceConfigOptions.EFO_CONSUMER_NAME, "my-flink-efo-consumer");

Pour un exemple de service géré pour une application Apache Flink utilisant un EFO consommateur, consultez notre exemple public de Kinesis Connectors sur Github.

Utilisez Amazon MSK

La KafkaSource source fournit des données de streaming à votre application à partir d'un MSK sujet Amazon.

Créer une KafkaSource

L’exemple de code suivant illustre la création d’un 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");

Pour plus d’informations sur l’utilisation d’un KafkaSource, consultez MSKRéplication.