Conectores de API de tabela

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. AWS

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

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

• Coletores de API de tabela: 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 de tabela

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 Conectores de tabela e SQL na documentação do Apache Flink.

Coletores de API de tabela

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 compatíveis 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.