Paso 1: Crear las secuencias de entrada y de salida

Antes de crear una aplicación de análisis de datos de Amazon Kinesis Data Analytics para el ejemplo de puntos de acceso, deberá crear dos secuencias de datos de Kinesis. Configure una de las secuencias como el origen de streaming de su aplicación y otra secuencia como el destino, donde Kinesis Data Analytics sigue siendo la salida de su aplicación.

Paso 1.1: Crear las secuencias de datos de Kinesis

En esta sección, creará dos secuencias de Kinesis: ExampleInputStream y ExampleOutputStream.

Cree estas secuencias de datos con la consola o la AWS CLI.

• Para crear estas secuencias de datos con la consola:

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

  2. Elija Data Streams (Secuencias de datos) en el panel de navegación.

  3. Elija Create Kinesis Stream (Crear secuencia de Kinesis) y, a continuación, cree una secuencia con un fragmento denominado ExampleInputStream.

  4. Repita el paso anterior y cree una secuencia con un fragmento denominada ExampleOutputStream.

• Para crear secuencias de datos con la AWS CLI:

  • Cree las secuencias (ExampleInputStream y ExampleOutputStream) con el siguiente comando create-stream de la AWS CLI de Kinesis. Para crear la segunda secuencia, que la aplicación utilizará para escribir la salida, ejecute el mismo comando, cambiando el nombre de la secuencia a ExampleOutputStream.

    $ aws kinesis create-stream \
    --stream-name ExampleInputStream \
    --shard-count 1 \
    --region us-west-2 \
    --profile adminuser
                             
    $ aws kinesis create-stream \
    --stream-name ExampleOutputStream \
    --shard-count 1 \
    --region us-west-2 \
    --profile adminuser
                                        

Paso 1.2: Escribir registros de muestra en la secuencia de entrada

En este paso, ejecute el código de Python para generar registros de muestra y escribir en la secuencia ExampleInputStream.

{"x": 7.921782426109737, "y": 8.746265312709893, "is_hot": "N"}
{"x": 0.722248626580026, "y": 4.648868803193405, "is_hot": "Y"}

1. Instale Python y pip.

   Para obtener más información sobre la instalación de Python, consulte la página web de Python.

   Puede instalar dependencias con pip. Para obtener más información sobre la instalación de pip, consulte la sección Installation en la página web de pip.

2. Ejecute el siguiente código de Python. Este código hace lo siguiente:

   • Genera un punto caliente potencial en algún lugar del plano (X, Y).

   • Genera un conjunto de 1000 puntos para cada punto caliente. De estos puntos, el 20 % se agrupa en torno a un punto caliente. El resto se genera de forma aleatoria en todo el espacio.

   • El comando put-record escribe los registros JSON en la secuencia.

   importante

   No cargue este archivo en un servidor web, ya que contiene sus credenciales de AWS.

    
   import json
   from pprint import pprint
   import random
   import time
   import boto3
   
   STREAM_NAME = "ExampleInputStream"
   
   
   def get_hotspot(field, spot_size):
       hotspot = {
           "left": field["left"] + random.random() * (field["width"] - spot_size),
           "width": spot_size,
           "top": field["top"] + random.random() * (field["height"] - spot_size),
           "height": spot_size,
       }
       return hotspot
   
   
   def get_record(field, hotspot, hotspot_weight):
       rectangle = hotspot if random.random() < hotspot_weight else field
       point = {
           "x": rectangle["left"] + random.random() * rectangle["width"],
           "y": rectangle["top"] + random.random() * rectangle["height"],
           "is_hot": "Y" if rectangle is hotspot else "N",
       }
       return {"Data": json.dumps(point), "PartitionKey": "partition_key"}
   
   
   def generate(
       stream_name, field, hotspot_size, hotspot_weight, batch_size, kinesis_client
   ):
       """
       Generates points used as input to a hotspot detection algorithm.
       With probability hotspot_weight (20%), a point is drawn from the hotspot;
       otherwise, it is drawn from the base field. The location of the hotspot
       changes for every 1000 points generated.
       """
       points_generated = 0
       hotspot = None
       while True:
           if points_generated % 1000 == 0:
               hotspot = get_hotspot(field, hotspot_size)
           records = [
               get_record(field, hotspot, hotspot_weight) for _ in range(batch_size)
           ]
           points_generated += len(records)
           pprint(records)
           kinesis_client.put_records(StreamName=stream_name, Records=records)
   
           time.sleep(0.1)
   
   
   if __name__ == "__main__":
       generate(
           stream_name=STREAM_NAME,
           field={"left": 0, "width": 10, "top": 0, "height": 10},
           hotspot_size=1,
           hotspot_weight=0.2,
           batch_size=10,
           kinesis_client=boto3.client("kinesis"),
       )
   

Paso siguiente

Paso 2: Creación de una aplicación de Kinesis Data Analytics