Conectores da API Table - Managed Service for Apache Flink
Fontes da API TableColetores da API TableFontes e coletores definidos pelo usuário

Anteriormente, o Amazon Managed Service for Apache Flink (Amazon MSF) era conhecido como Amazon Kinesis Data Analytics for Apache Flink.

Conectores da API Table

No modelo de programação do Apache Flink, os conectores são componentes que seu aplicativo usa para ler ou gravar dados de fontes externas, como outros serviços de AWS.

Com a API de tabela do Apache Flink, você pode usar os seguintes tipos de conectores:

  • Fontes da API Table: use conectores de origem da API de tabela para criar tabelas dentro da sua TableEnvironment usando chamadas de API ou consultas SQL.

  • Coletores da API Table: use comandos SQL para gravar dados de tabela em fontes externas, como um tópico do Amazon MSK ou um bucket do Amazon S3.

Fontes da API Table

Você cria uma fonte de tabela a partir de um fluxo de dados. O código a seguir cria uma tabela a partir de um tópico do Amazon MSK:

//create the table
    final FlinkKafkaConsumer<StockRecord> consumer = new FlinkKafkaConsumer<StockRecord>(kafkaTopic, new KafkaEventDeserializationSchema(), kafkaProperties);
    consumer.setStartFromEarliest();
    //Obtain stream
    DataStream<StockRecord> events = env.addSource(consumer);

    Table table = streamTableEnvironment.fromDataStream(events);

Para obter mais informações sobre fontes de tabelas, consulte Tabela e conectores SQL na Documentação do Apache Flink.

Coletores da API Table

Para gravar dados da tabela em um coletor, você cria o coletor em SQL e, em seguida, executa o coletor baseado em SQL no objeto StreamTableEnvironment.

O exemplo de código a seguir demonstra como gravar dados de tabela em um coletor do Amazon S3:

final String s3Sink = "CREATE TABLE sink_table (" +
    "event_time TIMESTAMP," +
    "ticker STRING," +
    "price DOUBLE," +
    "dt STRING," +
    "hr STRING" +
    ")" +
    " PARTITIONED BY (ticker,dt,hr)" +
    " WITH" +
    "(" +
    " 'connector' = 'filesystem'," +
    " 'path' = '" + s3Path + "'," +
    " 'format' = 'json'" +
    ") ";

    //send to s3
    streamTableEnvironment.executeSql(s3Sink);
    filteredTable.executeInsert("sink_table");

Você pode usar o parâmetro format para controlar qual formato o Managed Service for Apache Flink usa para gravar a saída no coletor. Para obter informações sobre formatos, consulte Conectores suportados na Documentação do Apache Flink.

Fontes e coletores definidos pelo usuário

Você pode usar os conectores Apache Kafka existentes para enviar dados de e para outros serviços de AWS, como Amazon MSK e Amazon S3. Para interagir com outras fontes de dados e destinos, você pode definir suas próprias fontes e coletores. Para obter mais informações, consulte Fontes e coletores definidos pelo usuário na Documentação do Apache Flink.