Añada fuentes de datos de streaming a Managed Service for Apache Flink - Managed Service para Apache Flink
Utilice los flujos de datos de KinesisUsa Amazon MSK

Amazon Managed Service para Apache Flink Amazon se denominaba anteriormente Amazon Kinesis Data Analytics para Apache Flink.

Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Añada fuentes de datos de streaming a Managed Service for Apache Flink

Apache Flink proporciona conectores para leer archivos, sockets, colecciones y fuentes personalizadas. En el código de su aplicación, debe utilizar una fuente de Apache Flink para recibir datos de un flujo. En esta sección se describen las fuentes disponibles para los servicios de Amazon.

Utilice los flujos de datos de Kinesis

KinesisStreamsSourceProporciona datos de streaming a su aplicación desde una transmisión de datos de Amazon Kinesis.

Creación de un KinesisStreamsSource

En el siguiente código de ejemplo se muestra la creación de 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();

Para obtener más información sobre el uso de unKinesisStreamsSource, consulte Amazon Kinesis Data Streams Connector en la documentación de Apache Flink y nuestro ejemplo KinesisConnectors público en Github.

Cree uno KinesisStreamsSource que utilice un consumidor EFO

KinesisStreamsSourceAhora es compatible con Enhanced Fan-Out (EFO).

Si un consumidor usa KinesisEFO, el servicio Kinesis Data Streams le proporciona su propio ancho de banda dedicado, en lugar de que el consumidor comparta el ancho de banda fijo de la transmisión con los demás consumidores que leen la transmisión.

Para obtener más información sobre su uso EFO con el Kinesis Consumer, consulte FLIP-128: Salida de ventilador mejorada para consumidores de AWS Kinesis.

Para habilitar al EFO consumidor, configure los siguientes parámetros en el consumidor de Kinesis:

  • READER_TYPE: Defina este parámetro EFOpara que su aplicación utilice un EFO consumidor para acceder a los datos de Kinesis Data Stream.

  • EFO_ CONSUMER _NAME: Establezca este parámetro en un valor de cadena que sea único entre los consumidores de esta transmisión. La reutilización de un nombre de consumidor en el mismo flujo de datos de Kinesis provocará la cancelación del consumidor anterior que utilizó ese nombre.

Para configurar un KinesisStreamsSource objeto para su usoEFO, añada los siguientes parámetros al consumidor:

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

Para ver un ejemplo de una aplicación de servicio gestionado para Apache Flink que utiliza un EFO consumidor, consulte nuestro ejemplo de conectores Kinesis públicos en Github.

Usa Amazon MSK

La KafkaSource fuente proporciona datos de streaming a tu aplicación desde un MSK tema de Amazon.

Creación de un KafkaSource

En el siguiente código de ejemplo se muestra la creación de 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");

Para obtener más información sobre cómo usar un KafkaSource, consulte Replicación MSK.