Após uma análise cuidadosa, decidimos descontinuar o Amazon Kinesis Data Analytics para aplicativos SQL: 1. A partir de **1º de setembro de 2025,** não forneceremos nenhuma correção de bug para aplicativos do Amazon Kinesis Data Analytics para SQL porque teremos suporte limitado para ele, devido à próxima descontinuação. 2. A partir **de 15 de outubro de 2025,** você não poderá criar novos aplicativos Kinesis Data Analytics para SQL. 3. Excluiremos as aplicações a partir de **27 de janeiro de 2026**. Você não poderá mais iniciar nem operar as aplicações do Amazon Kinesis Data Analytics para SQL. A partir dessa data, não haverá mais suporte ao Amazon Kinesis Data Analytics para SQL. Para obter mais informações, consulte [Descontinuação de aplicações do Amazon Kinesis Data Analytics para SQL](discontinuation.md). As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá. # Exemplos: janelas e agregação Esta seção fornece exemplos de aplicativos do Amazon Kinesis Data Analytics que usam consultas em janela e de agregação. (Para ter mais informações, consulte [Consultas em janelas](windowed-sql.md).) Cada exemplo fornece step-by-step instruções e exemplos de código para configurar o aplicativo Kinesis Data Analytics. **Topics** + [Exemplo: janela de escalonamento](examples-window-stagger.md) + [Exemplo: janela em cascata usando ROWTIME](examples-window-tumbling-rowtime.md) + [Exemplo: janela em cascata usando um time stamp do evento](examples-window-tumbling-event.md) + [Exemplo: recuperação dos valores que ocorrem com mais frequência (TOP\$1K\$1ITEMS\$1TUMBLING)](examples-window-topkitems.md) + [Exemplo: agregação dos resultados parciais de uma consulta](examples-window-partialresults.md) # Exemplo: janela de escalonamento Quando uma consulta em janela processa janelas separadas para cada chave de partição exclusiva, a partir da chegada dos dados com a chave correspondente, a janela é chamada de *janela de escalonamento*. Para obter detalhes, consulte [Janelas de escalonamento](stagger-window-concepts.md). Esse exemplo do Amazon Kinesis Data Analytics usa as colunas EVENT\$1TIME e TICKER para criar janelas de escalonamento. O stream de origem contém grupos de seis registros com valores EVENT\$1TIME e TICKER idênticos, que chegam em um período de um minuto, mas não necessariamente com o mesmo valor de minutos (por exemplo, `18:41:xx`). Neste exemplo, você grava os registros a seguir em um fluxo de dados do Kinesis nos seguintes tempos. O script não grava os horários no fluxo, mas o horário em que o registro é consumido pelo aplicativo é gravado no campo `ROWTIME`: ``` {"EVENT_TIME": "2018-08-01T20:17:20.797945", "TICKER": "AMZN"} 20:17:30 {"EVENT_TIME": "2018-08-01T20:17:20.797945", "TICKER": "AMZN"} 20:17:40 {"EVENT_TIME": "2018-08-01T20:17:20.797945", "TICKER": "AMZN"} 20:17:50 {"EVENT_TIME": "2018-08-01T20:17:20.797945", "TICKER": "AMZN"} 20:18:00 {"EVENT_TIME": "2018-08-01T20:17:20.797945", "TICKER": "AMZN"} 20:18:10 {"EVENT_TIME": "2018-08-01T20:17:20.797945", "TICKER": "AMZN"} 20:18:21 {"EVENT_TIME": "2018-08-01T20:18:21.043084", "TICKER": "INTC"} 20:18:31 {"EVENT_TIME": "2018-08-01T20:18:21.043084", "TICKER": "INTC"} 20:18:41 {"EVENT_TIME": "2018-08-01T20:18:21.043084", "TICKER": "INTC"} 20:18:51 {"EVENT_TIME": "2018-08-01T20:18:21.043084", "TICKER": "INTC"} 20:19:01 {"EVENT_TIME": "2018-08-01T20:18:21.043084", "TICKER": "INTC"} 20:19:11 {"EVENT_TIME": "2018-08-01T20:18:21.043084", "TICKER": "INTC"} 20:19:21 ... ``` Em seguida, você cria um aplicativo Kinesis Data Analytics no, com Console de gerenciamento da AWS o stream de dados do Kinesis como fonte de streaming. O processo de descoberta lê os registros de exemplo na origem de streaming e infere um esquema no aplicativo com duas colunas (`EVENT_TIME` e `TICKER`) conforme mostrado a seguir. ![\[Captura de tela do console mostrando o esquema no aplicativo com as colunas de preço e de marcador.\]](http://docs.aws.amazon.com/pt_br/kinesisanalytics/latest/dev/images/ex_stagger_schema.png) Você usa o código do aplicativo com a função `COUNT` para criar uma agregação em janelas dos dados. Em seguida, insira os dados resultantes em outro stream no aplicativo, conforme mostrado na captura de tela a seguir: ![\[Captura de tela do console mostrando os dados resultantes em um stream no aplicativo.\]](http://docs.aws.amazon.com/pt_br/kinesisanalytics/latest/dev/images/ex_stagger.png) No procedimento a seguir, você cria um aplicativo do Kinesis Data Analytics que agrega valores no streaming de entrada em uma janela de escalonamento com base em EVENT\$1TIME e TICKER. **Topics** + [Etapa 1: Criar um fluxo de dados Kinesis](#examples-stagger-window-1) + [Etapa 2: Criar o aplicativo Kinesis Data Analytics](#examples-stagger-window-2) ## Etapa 1: Criar um fluxo de dados Kinesis Crie um fluxo de dados do Amazon Kinesis e preencha registros da seguinte forma: 1. [Faça login no Console de gerenciamento da AWS e abra o console do Kinesis em https://console.aws.amazon.com /kinesis.](https://console.aws.amazon.com/kinesis) 1. Selecione **Fluxos de dados** no painel de navegação. 1. Selecione **Create Kinesis stream (Criar stream do Kinesis)** e crie um stream com um estilhaço. Para obter mais informações, consulte [Criar um fluxo](https://docs.aws.amazon.com/streams/latest/dev/learning-kinesis-module-one-create-stream.html) no *Guia do desenvolvedor do Amazon Kinesis Data Streams*. 1. Para gravar registros em um streaming de dados do Kinesis em um ambiente de produção, recomendamos usar o [Kinesis Producer Library](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-kpl.html) ou a [API do Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-sdk.html). Para simplificar, este exemplo usa o script Python a seguir para gerar registros. Execute o código para preencher os registros de marcador de exemplo. Esse código simples grava continuamente um grupo de seis registros com o mesmo símbolo de índice e `EVENT_TIME` aleatórios no stream durante o período de um minuto. Mantenha o script em execução para que você possa gerar o esquema do aplicativo em uma etapa posterior. ``` import datetime import json import random import time import boto3 STREAM_NAME = "ExampleInputStream" def get_data(): event_time = datetime.datetime.utcnow() - datetime.timedelta(seconds=10) return { "EVENT_TIME": event_time.isoformat(), "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]), } def generate(stream_name, kinesis_client): while True: data = get_data() # Send six records, ten seconds apart, with the same event time and ticker for _ in range(6): print(data) kinesis_client.put_record( StreamName=stream_name, Data=json.dumps(data), PartitionKey="partitionkey", ) time.sleep(10) if __name__ == "__main__": generate(STREAM_NAME, boto3.client("kinesis")) ``` ## Etapa 2: Criar o aplicativo Kinesis Data Analytics Crie um aplicativo Kinesis Data Analytics, da seguinte maneira: 1. [Abra o console do Managed Service for Apache Flink em https://console.aws.amazon.com /kinesisanalytics.](https://console.aws.amazon.com/kinesisanalytics) 1. Escolha **Create application (Criar aplicativo)**, digite um nome para o aplicativo e selecione **Create application (Criar aplicativo)**. 1. Na página de detalhes do aplicativo, escolha **Connect streaming data (Conectar dados de streaming)** para se conectar com a fonte. 1. Na página **Connect to source (Conectar com a fonte)**, faça o seguinte: 1. Escolha o stream criado na seção anterior. 1. Selecione **Discover schema (Descobrir esquema)**. Aguarde o console mostrar o esquema inferido e os registros de exemplos usados para inferir o esquema do stream do aplicativo criado. O esquema inferido tem duas colunas. 1. Escolha **Edit Schema (Editar esquema)**. Mude **Column type (Tipo de coluna)** da coluna **EVENT\$1TIME** para `TIMESTAMP`. 1. Escolha **Save schema and update stream samples**. Depois que o console salvar o esquema, escolha **Exit (Sair)**. 1. Escolha **Save and continue**. 1. 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. 1. 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" ( event_time TIMESTAMP, ticker_symbol VARCHAR(4), ticker_count INTEGER); CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM EVENT_TIME, TICKER, COUNT(TICKER) AS ticker_count FROM "SOURCE_SQL_STREAM_001" WINDOWED BY STAGGER ( PARTITION BY TICKER, EVENT_TIME RANGE INTERVAL '1' MINUTE); ``` 1. 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. # Exemplo: janela em cascata usando ROWTIME Quando uma consulta em janela processa cada janela de forma não sobreposta, é chamada de *janela em cascata*. Para obter detalhes, consulte [Janelas em cascata (Agregações usando GROUP BY)](tumbling-window-concepts.md). Este exemplo do Amazon Kinesis Data Analytics usa a coluna `ROWTIME` para criar janelas em cascata. A coluna `ROWTIME` representa o momento em que o registro foi lido pelo aplicativo. Neste exemplo, você grava os registros a seguir em um streaming de dados do Kinesis. ``` {"TICKER": "TBV", "PRICE": 33.11} {"TICKER": "INTC", "PRICE": 62.04} {"TICKER": "MSFT", "PRICE": 40.97} {"TICKER": "AMZN", "PRICE": 27.9} ... ``` Em seguida, você cria um aplicativo Kinesis Data Analytics no, com Console de gerenciamento da AWS o stream de dados do Kinesis como fonte de streaming. O processo de descoberta lê os registros de exemplo na origem de streaming e infere um esquema no aplicativo com duas colunas (`TICKER` e `PRICE`) conforme mostrado a seguir. ![\[Captura de tela do console mostrando o esquema no aplicativo com as colunas de preço e de marcador.\]](http://docs.aws.amazon.com/pt_br/kinesisanalytics/latest/dev/images/ex_tumbling_rowtime_schema.png) Você usa o código do aplicativo com as funções `MIN` e `MAX` para criar uma agregação em janela dos dados. Em seguida, insira os dados resultantes em outro stream no aplicativo, conforme mostrado na captura de tela a seguir: ![\[Captura de tela do console mostrando os dados resultantes em um stream no aplicativo.\]](http://docs.aws.amazon.com/pt_br/kinesisanalytics/latest/dev/images/ex_tumbling_rowtime.png) No procedimento a seguir, você cria um aplicativo do Kinesis Data Analytics que agrega valores no stream de entrada em uma janela em cascata com base em ROWTIME. **Topics** + [Etapa 1: Criar um fluxo de dados Kinesis](#examples-tumbling-window-1) + [Etapa 2: Criar o aplicativo Kinesis Data Analytics](#examples-tumbling-window-2) ## Etapa 1: Criar um fluxo de dados Kinesis Crie um fluxo de dados do Amazon Kinesis e preencha registros da seguinte forma: 1. [Faça login no Console de gerenciamento da AWS e abra o console do Kinesis em https://console.aws.amazon.com /kinesis.](https://console.aws.amazon.com/kinesis) 1. Selecione **Fluxos de dados** no painel de navegação. 1. Selecione **Create Kinesis stream (Criar stream do Kinesis)** e crie um stream com um estilhaço. Para obter mais informações, consulte [Criar um fluxo](https://docs.aws.amazon.com/streams/latest/dev/learning-kinesis-module-one-create-stream.html) no *Guia do desenvolvedor do Amazon Kinesis Data Streams*. 1. Para gravar registros em um streaming de dados do Kinesis em um ambiente de produção, recomendamos usar o [Kinesis Client Library](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-kpl.html) ou a [API do Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-sdk.html). Para simplificar, este exemplo usa o script Python a seguir para gerar registros. Execute o código para preencher os registros de marcador de exemplo. Esse código simples grava continuamente um registro de marcador aleatório no stream. Mantenha o script em execução para que você possa gerar o esquema do aplicativo em uma etapa posterior. ``` 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")) ``` ## Etapa 2: Criar o aplicativo Kinesis Data Analytics Crie um aplicativo Kinesis Data Analytics, da seguinte maneira: 1. [Abra o console do Managed Service for Apache Flink em https://console.aws.amazon.com /kinesisanalytics.](https://console.aws.amazon.com/kinesisanalytics) 1. Escolha **Create application (Criar aplicativo)**, insira um nome para o aplicativo e selecione **Create application (Criar aplicativo)**. 1. Na página de detalhes do aplicativo, escolha **Connect streaming data (Conectar dados de streaming)** para se conectar com a fonte. 1. Na página **Connect to source (Conectar com a fonte)**, faça o seguinte: 1. Escolha o stream criado na seção anterior. 1. Selecione **Discover schema (Descobrir esquema)**. Aguarde o console mostrar o esquema inferido e os registros de exemplos usados para inferir o esquema do stream do aplicativo criado. O esquema inferido tem duas colunas. 1. Escolha **Save schema and update stream samples**. Depois que o console salvar o esquema, escolha **Exit (Sair)**. 1. Escolha **Save and continue**. 1. 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. 1. 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), MIN_PRICE REAL, MAX_PRICE REAL); CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM TICKER, MIN(PRICE), MAX(PRICE) FROM "SOURCE_SQL_STREAM_001" GROUP BY TICKER, STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND); ``` 1. 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. # Exemplo: janela em cascata usando um time stamp do evento Quando uma consulta em janela processa cada janela de forma não sobreposta, é chamada de *janela em cascata*. Para obter detalhes, consulte [Janelas em cascata (Agregações usando GROUP BY)](tumbling-window-concepts.md). Este exemplo do Amazon Kinesis Data Analytics demonstra uma janela em cascata que usa um time stamp do evento, que é um time stamp criado pelo usuário que está incluído no streaming de dados. Ele usa essa abordagem, em vez de apenas usar ROWTIME, que é um time stamp criado pelo Kinesis Data Analytics quando o aplicativo recebe o registro. Use um time stamp do evento no streaming de dados se quiser criar uma agregação com base em quando um evento ocorreu, em vez de quando ele foi recebido pelo aplicativo. Neste exemplo, o valor de `ROWTIME` aciona a agregação a cada minuto, e os registros são agregados pelo `ROWTIME` e pelo horário do evento incluído. Neste exemplo, você grava os registros a seguir em um stream do Amazon Kinesis. O valor `EVENT_TIME` é definido como 5 segundos no passado, para simular o atraso no processamento e na transmissão que pode criar um atraso de quando o evento ocorreu para quando o registro foi recebido no Kinesis Data Analytics. ``` {"EVENT_TIME": "2018-06-13T14:11:05.766191", "TICKER": "TBV", "PRICE": 43.65} {"EVENT_TIME": "2018-06-13T14:11:05.848967", "TICKER": "AMZN", "PRICE": 35.61} {"EVENT_TIME": "2018-06-13T14:11:05.931871", "TICKER": "MSFT", "PRICE": 73.48} {"EVENT_TIME": "2018-06-13T14:11:06.014845", "TICKER": "AMZN", "PRICE": 18.64} ... ``` Em seguida, você cria um aplicativo Kinesis Data Analytics no, com Console de gerenciamento da AWS o stream de dados do Kinesis como fonte de streaming. O processo de descoberta lê os registros de exemplo na origem de streaming e infere um esquema no aplicativo com três colunas (`EVENT_TIME`, `TICKER` e `PRICE`) conforme mostrado a seguir. ![\[Captura de tela do console mostrando o esquema no aplicativo com as colunas de horário do evento, de preço e de marcador.\]](http://docs.aws.amazon.com/pt_br/kinesisanalytics/latest/dev/images/ex_tumbling_event_schema.png) Você usa o código do aplicativo com as funções `MIN` e `MAX` para criar uma agregação em janela dos dados. Em seguida, insira os dados resultantes em outro stream no aplicativo, conforme mostrado na captura de tela a seguir: ![\[Captura de tela do console mostrando os dados resultantes em um stream no aplicativo.\]](http://docs.aws.amazon.com/pt_br/kinesisanalytics/latest/dev/images/ex_tumbling_event.png) No procedimento a seguir, você cria um aplicativo do Kinesis Data Analytics que agrega valores no stream de entrada em uma janela em cascata com base em um horário de evento. **Topics** + [Etapa 1: Criar um fluxo de dados Kinesis](#examples-window-tumbling-event-1) + [Etapa 2: Criar o aplicativo Kinesis Data Analytics](#examples-window-tumbling-event-2) ## Etapa 1: Criar um fluxo de dados Kinesis Crie um fluxo de dados do Amazon Kinesis e preencha registros da seguinte forma: 1. [Faça login no Console de gerenciamento da AWS e abra o console do Kinesis em https://console.aws.amazon.com /kinesis.](https://console.aws.amazon.com/kinesis) 1. Selecione **Fluxos de dados** no painel de navegação. 1. Selecione **Create Kinesis stream (Criar stream do Kinesis)** e crie um stream com um estilhaço. Para obter mais informações, consulte [Criar um fluxo](https://docs.aws.amazon.com/streams/latest/dev/learning-kinesis-module-one-create-stream.html) no *Guia do desenvolvedor do Amazon Kinesis Data Streams*. 1. Para gravar registros em um streaming de dados do Kinesis em um ambiente de produção, recomendamos usar o [Kinesis Client Library](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-kpl.html) ou a [API do Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-sdk.html). Para simplificar, este exemplo usa o script Python a seguir para gerar registros. Execute o código para preencher os registros de marcador de exemplo. Esse código simples grava continuamente um registro de marcador aleatório no stream. Mantenha o script em execução para que você possa gerar o esquema do aplicativo em uma etapa posterior. ``` 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")) ``` ## Etapa 2: Criar o aplicativo Kinesis Data Analytics Crie um aplicativo Kinesis Data Analytics, da seguinte maneira: 1. [Abra o console do Managed Service for Apache Flink em https://console.aws.amazon.com /kinesisanalytics.](https://console.aws.amazon.com/kinesisanalytics) 1. Escolha **Create application (Criar aplicativo)**, insira um nome para o aplicativo e selecione **Create application (Criar aplicativo)**. 1. Na página de detalhes do aplicativo, escolha **Connect streaming data (Conectar dados de streaming)** para se conectar com a fonte. 1. Na página **Connect to source (Conectar com a fonte)**, faça o seguinte: 1. Escolha o stream criado na seção anterior. 1. Selecione **Discover schema (Descobrir esquema)**. Aguarde o console mostrar o esquema inferido e os registros de exemplos usados para inferir o esquema do stream do aplicativo criado. O esquema inferido tem três colunas. 1. Escolha **Edit Schema (Editar esquema)**. Mude **Column type (Tipo de coluna)** da coluna **EVENT\$1TIME** para `TIMESTAMP`. 1. Escolha **Save schema and update stream samples**. Depois que o console salvar o esquema, escolha **Exit (Sair)**. 1. Escolha **Save and continue**. 1. 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. 1. 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" (EVENT_TIME timestamp, TICKER VARCHAR(4), min_price REAL, max_price REAL); CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM STEP("SOURCE_SQL_STREAM_001".EVENT_TIME BY INTERVAL '60' SECOND), TICKER, MIN(PRICE) AS MIN_PRICE, MAX(PRICE) AS MAX_PRICE FROM "SOURCE_SQL_STREAM_001" GROUP BY TICKER, STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND), STEP("SOURCE_SQL_STREAM_001".EVENT_TIME BY INTERVAL '60' SECOND); ``` 1. 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. # Exemplo: recuperação dos valores que ocorrem com mais frequência (TOP\$1K\$1ITEMS\$1TUMBLING) Este exemplo do Amazon Kinesis Data Analytics demonstra como usar a função `TOP_K_ITEMS_TUMBLING` para recuperar os valores que ocorrem com mais frequência em uma janela em cascata. Para obter mais informações, consulte a [`TOP_K_ITEMS_TUMBLING`função](https://docs.aws.amazon.com/kinesisanalytics/latest/sqlref/top-k.html) em *Amazon Managed Service for Apache Flink SQL Reference*. A função `TOP_K_ITEMS_TUMBLING` é útil quando agrega dezenas ou centenas de milhares de chaves e você deseja reduzir o uso de recursos. A função gera o mesmo resultado que a agregação com as cláusulas `GROUP BY` e `ORDER BY`. Neste exemplo, você grava os registros a seguir em um fluxo de dados do Amazon Kinesis. ``` {"TICKER": "TBV"} {"TICKER": "INTC"} {"TICKER": "MSFT"} {"TICKER": "AMZN"} ... ``` Em seguida, você cria um aplicativo Kinesis Data Analytics no, com Console de gerenciamento da AWS o stream de dados do Kinesis como fonte de streaming. O processo de descoberta lê os registros de exemplo na origem de streaming e infere um esquema no aplicativo com uma coluna (`TICKER`) conforme mostrado a seguir. ![\[Captura de tela do console mostrando o esquema no aplicativo com uma coluna de marcador.\]](http://docs.aws.amazon.com/pt_br/kinesisanalytics/latest/dev/images/ex_topk_schema.png) Você usa o código do aplicativo com a função `TOP_K_VALUES_TUMBLING` para criar uma agregação em janelas dos dados. Em seguida, insira os dados resultantes em outro stream no aplicativo, conforme mostrado na captura de tela a seguir: ![\[Captura de tela do console mostrando os dados resultantes em um stream no aplicativo.\]](http://docs.aws.amazon.com/pt_br/kinesisanalytics/latest/dev/images/ex_topk.png) No procedimento a seguir, você criará um aplicativo do Kinesis Data Analytics que recupera os valores que ocorrem com mais frequência no stream de entrada. **Topics** + [Etapa 1: Criar um fluxo de dados Kinesis](#examples-window-topkitems-1) + [Etapa 2: Criar o aplicativo Kinesis Data Analytics](#examples-window-topkitems-2) ## Etapa 1: Criar um fluxo de dados Kinesis Crie um fluxo de dados do Amazon Kinesis e preencha registros da seguinte forma: 1. [Faça login no Console de gerenciamento da AWS e abra o console do Kinesis em https://console.aws.amazon.com /kinesis.](https://console.aws.amazon.com/kinesis) 1. Selecione **Fluxos de dados** no painel de navegação. 1. Selecione **Create Kinesis stream (Criar stream do Kinesis)** e crie um stream com um estilhaço. Para obter mais informações, consulte [Criar um fluxo](https://docs.aws.amazon.com/streams/latest/dev/learning-kinesis-module-one-create-stream.html) no *Guia do desenvolvedor do Amazon Kinesis Data Streams*. 1. Para gravar registros em um streaming de dados do Kinesis em um ambiente de produção, recomendamos usar o [Kinesis Client Library](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-kpl.html) ou a [API do Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-sdk.html). Para simplificar, este exemplo usa o script Python a seguir para gerar registros. Execute o código para preencher os registros de marcador de exemplo. Esse código simples grava continuamente um registro de marcador aleatório no stream. Deixe o script em execução para que você possa gerar o esquema do aplicativo em uma etapa posterior. ``` 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")) ``` ## Etapa 2: Criar o aplicativo Kinesis Data Analytics Crie um aplicativo Kinesis Data Analytics, da seguinte maneira: 1. [Abra o console do Managed Service for Apache Flink em https://console.aws.amazon.com /kinesisanalytics.](https://console.aws.amazon.com/kinesisanalytics) 1. Escolha **Create application (Criar aplicativo)**, digite um nome para o aplicativo e selecione **Create application (Criar aplicativo)**. 1. Na página de detalhes do aplicativo, escolha **Connect streaming data (Conectar dados de streaming)** para se conectar com a fonte. 1. Na página **Connect to source (Conectar com a fonte)**, faça o seguinte: 1. Escolha o stream criado na seção anterior. 1. Selecione **Discover schema (Descobrir esquema)**. Aguarde o console mostrar o esquema inferido e os registros de exemplos usados para inferir o esquema do stream do aplicativo criado. O esquema inferido tem uma coluna. 1. Escolha **Save schema and update stream samples**. Depois que o console salvar o esquema, escolha **Exit (Sair)**. 1. Escolha **Save and continue**. 1. 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. 1. 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), "MOST_FREQUENT_VALUES" BIGINT ); CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM * FROM TABLE (TOP_K_ITEMS_TUMBLING( CURSOR(SELECT STREAM * FROM "SOURCE_SQL_STREAM_001"), 'TICKER', -- name of column in single quotes 5, -- number of the most frequently occurring values 60 -- tumbling window size in seconds ) ); ``` 1. 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. # Exemplo: agregação dos resultados parciais de uma consulta Se um streaming de dados do Amazon Kinesis contiver registros que tenham um horário do evento que não corresponda exatamente ao horário da ingestão, uma seleção de resultados em uma janela em cascata contém os registros que chegaram, mas não necessariamente ocorreram, dentro da janela. Nesse caso, a janela em cascata conterá somente um conjunto parcial dos resultados que você deseja. Há várias abordagens que você pode usar para corrigir o problema: + Use somente uma janela em cascata e agregue resultados parciais no pós-processamento por meio de um banco de dados ou data warehouse usando upserts. Essa abordagem é eficiente no processamento de um aplicativo. Ela manipula os dados atrasados indefinidamente para operadores agregados (`sum`, `min`, `max`, e assim por diante). A desvantagem dessa abordagem é que você deve desenvolver e manter lógica adicional do aplicativo na camada de banco de dados. + Use uma janela deslizante e em cascata, que produz resultados parciais logo, mas também continua a produzir resultados completos ao longo do período da janela deslizante. Essa abordagem trata os dados tardios com uma substituição em vez de um upsert para que nenhuma lógica adicional do aplicativo precise ser adicionada à camada de banco de dados. A desvantagem dessa abordagem é que ela usa mais unidades de processamento do Kinesis KPUs () e ainda produz dois resultados, o que pode não funcionar em alguns casos de uso. Para obter mais informações sobre janelas deslizantes e em cascata, consulte [Consultas em janelas](windowed-sql.md). No procedimento a seguir, a agregação de janela em cascata produz dois resultados parciais (enviados ao stream no aplicativo `CALC_COUNT_SQL_STREAM`) que devem ser combinados para produzir um resultado final. O aplicativo, então, produz uma segunda agregação (enviada ao stream no aplicativo `DESTINATION_SQL_STREAM`) que combina os dois resultados parciais. **Para criar um aplicativo que agrega resultados parciais usando um horário do evento** 1. [Faça login no Console de gerenciamento da AWS e abra o console do Kinesis em https://console.aws.amazon.com /kinesis.](https://console.aws.amazon.com/kinesis) 1. Selecione **Data Analytics (Análise de dados)** no painel de navegação. Crie um aplicativo do Kinesis Data Analytics conforme descrito no tutorial [Conceitos básicos do Amazon Kinesis Data Analytics para aplicativos SQL](getting-started.md). 1. No editor SQL, substitua o código de aplicativo pelo seguinte: ``` CREATE OR REPLACE STREAM "CALC_COUNT_SQL_STREAM" (TICKER VARCHAR(4), TRADETIME TIMESTAMP, TICKERCOUNT DOUBLE); CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (TICKER VARCHAR(4), TRADETIME TIMESTAMP, TICKERCOUNT DOUBLE); CREATE PUMP "CALC_COUNT_SQL_PUMP_001" AS INSERT INTO "CALC_COUNT_SQL_STREAM" ("TICKER","TRADETIME", "TICKERCOUNT") SELECT STREAM "TICKER_SYMBOL", STEP("SOURCE_SQL_STREAM_001"."ROWTIME" BY INTERVAL '1' MINUTE) as "TradeTime", COUNT(*) AS "TickerCount" FROM "SOURCE_SQL_STREAM_001" GROUP BY STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '1' MINUTE), STEP("SOURCE_SQL_STREAM_001"."APPROXIMATE_ARRIVAL_TIME" BY INTERVAL '1' MINUTE), TICKER_SYMBOL; CREATE PUMP "AGGREGATED_SQL_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" ("TICKER","TRADETIME", "TICKERCOUNT") SELECT STREAM "TICKER", "TRADETIME", SUM("TICKERCOUNT") OVER W1 AS "TICKERCOUNT" FROM "CALC_COUNT_SQL_STREAM" WINDOW W1 AS (PARTITION BY "TRADETIME" RANGE INTERVAL '10' MINUTE PRECEDING); ``` A instrução `SELECT` no código de aplicativo filtra as linhas no `SOURCE_SQL_STREAM_001` por alterações de preços de ações maiores de 1% e insere essas linhas em outro stream no aplicativo `CHANGE_STREAM` usando uma bomba. 1. Escolha **Save and run SQL**. A primeira bomba produz um stream para `CALC_COUNT_SQL_STREAM` que é semelhante ao seguinte (observe que o conjunto de resultados está incompleto): ![\[Captura de tela do console mostrando resultados parciais.\]](http://docs.aws.amazon.com/pt_br/kinesisanalytics/latest/dev/images/ex_partial_0.png) A segunda bomba produz um stream para `DESTINATION_SQL_STREAM` que contém o conjunto de resultados completo: ![\[Captura de tela do console mostrando resultados completos.\]](http://docs.aws.amazon.com/pt_br/kinesisanalytics/latest/dev/images/ex_partial_1.png)