Exemplo: Transformação de valores DateTime

Faça login AWS Management Console e abra o console do Kinesis em https://console.aws.amazon.com/kinesis.

Selecione Data Streams (Fluxos de dados) no painel de navegação.

Escolha Create Kinesis stream (Criar fluxo do Kinesis) e crie um fluxo com um estilhaço.

Execute o seguinte código Python para preencher o fluxo com dados de amostra. Esse código simples grava continuamente um registro com um símbolo de marcador aleatório e o timestamp atual no fluxo.

 
import datetime
import json
import random
import boto3

STREAM_NAME = "ExampleInputStream"


def get_data():
    return {
        "EVENT_TIME": datetime.datetime.now().isoformat(),
        "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]),
        "PRICE": round(random.random() * 100, 2),
    }


def generate(stream_name, kinesis_client):
    while True:
        data = get_data()
        print(data)
        kinesis_client.put_record(
            StreamName=stream_name, Data=json.dumps(data), PartitionKey="partitionkey"
        )


if __name__ == "__main__":
    generate(STREAM_NAME, boto3.client("kinesis"))

Abra o Managed Service for Apache Flink console em https://console.aws.amazon.com/kinesisanalytics.

Escolha Create application (Criar aplicativo), digite um nome para o aplicativo e selecione Create application (Criar aplicativo).

Na página de detalhes do aplicativo, escolha Connect streaming data (Conectar dados de streaming) para se conectar com a fonte.

Na página Connect to source (Conectar com a fonte), faça o seguinte:

1. Escolha o stream criado na seção anterior.

2. Escolha para criar uma função do IAM.

3. Escolha Discover schema (Descobrir esquema). Aguarde o console mostrar o esquema inferido e os registros de exemplo usados para inferir o esquema do fluxo do aplicativo criado. O esquema inferido tem duas colunas.

4. Escolha Edit Schema (Editar esquema). Mude Column type (Tipo de coluna) da coluna EVENT_TIME para TIMESTAMP.

5. Escolha Save schema and update stream samples. Depois que o console salvar o esquema, escolha Exit (Sair).

6. Escolha Save and continue.

Na página de detalhes de aplicativo, escolha Go to SQL editor (Ir para o editor de SQL). Para iniciar o aplicativo, escolha Yes, start application (Sim, iniciar o aplicativo) na caixa de diálogo exibida.

No editor SQL, escreva o código do aplicativo e verifique os resultados da seguinte forma:

1. Copie o código de aplicativo a seguir e cole-o no editor.

   CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
       TICKER VARCHAR(4), 
       event_time TIMESTAMP, 
       five_minutes_before TIMESTAMP, 
       event_unix_timestamp BIGINT,
       event_timestamp_as_char VARCHAR(50),
       event_second INTEGER);
   
   CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM"
   
   SELECT STREAM 
       TICKER, 
       EVENT_TIME,
       EVENT_TIME - INTERVAL '5' MINUTE,
       UNIX_TIMESTAMP(EVENT_TIME),
       TIMESTAMP_TO_CHAR('yyyy-MM-dd hh:mm:ss', EVENT_TIME),
       EXTRACT(SECOND FROM EVENT_TIME) 
   FROM "SOURCE_SQL_STREAM_001"
   

2. Escolha Save and run SQL. Na guia Real-time analytics (Análise em tempo real), você pode ver todos os fluxos de aplicativo criados pelo aplicativo e verificar os dados.