Ejemplo: transformación DateTime de valores

Inicie sesión en la consola Kinesis AWS Management Console y ábrala en https://console.aws.amazon.com/kinesis.

Elija Flujos de datos en el panel de navegación.

Elija Crear secuencia de Kinesis y a continuación cree una secuencia con una partición.

Ejecute el siguiente código de Python para rellenar la secuencia con datos de muestra. Este código de ejemplo escribe continuamente un registro con un símbolo de cotización aleatorio y la marca temporal actual en la secuencia.

 
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 la consola de Managed Service para Apache Flink en https://console.aws.amazon.com/kinesisanalytics.

Elija Create application (Crear aplicación), escriba el nombre de la aplicación y elija Create application (Crear aplicación).

En la página de detalles de la aplicación, elija Connect streaming data (Conectar datos de streaming) para conectarse al origen.

En la página Connect to source (Conectarse al origen), haga lo siguiente:

1. Elija la secuencia que ha creado en la sección anterior.

2. Elija crear un rol de IAM.

3. Seleccione Detectar esquema. Espere a que la consola muestre el esquema inferido y los registros de muestra utilizados para inferir en el esquema de la secuencia en la aplicación que ha creado. El esquema inferido cuenta con dos columnas.

4. Elija Edit Schema (Editar esquema). Cambie el Column type (Tipo de columna) de la columna EVENT_TIME a TIMESTAMP.

5. Elija Save schema and update stream samples. Después de que la consola guarde el esquema, elija Exit (Salir).

6. Elija Guardar y continuar.

En la página de detalles de la aplicación, elija Go to SQL editor (Ir al editor de SQL). Para iniciar la aplicación, elija Yes, start application (Sí, iniciar la aplicación) en el cuadro de diálogo que aparece.

En el editor de SQL, escriba el código de la aplicación y verifique los resultados como se indica a continuación:

1. Copie el siguiente código de la aplicación y péguelo en el 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. Elija Save and run SQL. En la pestaña Real-time analytics (Análisis en tiempo real), puede ver todas las secuencias en la aplicación creadas por esta y comprobar los datos.