Etapa 1: preparar

Antes de criar um aplicativo Amazon Kinesis Data Analytics para este exercício, crie dois fluxos de dados do Kinesis. Configure um dos streamings como a origem de streaming do aplicativo e outro como o destino em que o Kinesis Data Analytics mantém a saída do aplicativo.

Etapa 1.1: criar os fluxos de dados de entrada e saída

Nesta seção, você cria dois streams do Kinesis: ExampleInputStream e ExampleOutputStream. Crie esses streams usando o AWS Management Console ou a AWS CLI.

• Para usar o console do

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

  2. Selecione Criar fluxo de dados. Crie um fluxo com um estilhaço chamado ExampleInputStream. Para obter mais informações, consulte Criar um fluxo no Guia do desenvolvedor do Amazon Kinesis Data Streams.

  3. Repita a etapa anterior, criando um streaming com um estilhaço denominado ExampleOutputStream.

• Para usar o AWS CLI

  1. Use o seguinte create-stream AWS CLI comando do Kinesis para criar o primeiro stream ()ExampleInputStream.

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

  2. Execute o mesmo comando, alterando o nome do fluxo para ExampleOutputStream. Esse comando cria o segundo fluxo usado pelo aplicativo para gravar a saída.

Etapa 1.2: Gravação de registros de amostra no stream de entrada

Nesta etapa, você executa o código Python para gerar continuamente registros de amostra e gravar esses registros no fluxo ExampleInputStream.

{"heartRate": 60, "rateType":"NORMAL"} 
...
{"heartRate": 180, "rateType":"HIGH"}

1. Instale o Python e o pip.

   Para obter informações sobre como instalar o Python, consulte o site do Python.

   Você pode instalar dependências usando o pip. Para obter informações sobre como instalar o pip, consulte Installation no site do pip.

2. Execute o código do Python a seguir. O comando put-record no código grava os registros JSON no fluxo.

    
   from enum import Enum
   import json
   import random
   import boto3
   
   STREAM_NAME = "ExampleInputStream"
   
   
   class RateType(Enum):
       normal = "NORMAL"
       high = "HIGH"
   
   
   def get_heart_rate(rate_type):
       if rate_type == RateType.normal:
           rate = random.randint(60, 100)
       elif rate_type == RateType.high:
           rate = random.randint(150, 200)
       else:
           raise TypeError
       return {"heartRate": rate, "rateType": rate_type.value}
   
   
   def generate(stream_name, kinesis_client, output=True):
       while True:
           rnd = random.random()
           rate_type = RateType.high if rnd < 0.01 else RateType.normal
           heart_rate = get_heart_rate(rate_type)
           if output:
               print(heart_rate)
           kinesis_client.put_record(
               StreamName=stream_name,
               Data=json.dumps(heart_rate),
               PartitionKey="partitionkey",
           )
   
   
   if __name__ == "__main__":
       generate(STREAM_NAME, boto3.client("kinesis"))
   

Próxima etapa

Etapa 2: criar um aplicativo