Après mûre réflexion, nous avons décidé de mettre fin à Amazon Kinesis Data Analytics pour les applications SQL :
1. À compter du **1er septembre 2025,** nous ne fournirons aucune correction de bogue pour les applications Amazon Kinesis Data Analytics for SQL, car leur support sera limité, compte tenu de l'arrêt prochain.
2. À compter du **15 octobre 2025,** vous ne pourrez plus créer de nouvelles applications Kinesis Data Analytics for SQL.
3. Nous supprimerons vos candidatures à compter **du 27 janvier 2026**. Vous ne serez pas en mesure de démarrer ou d'utiliser vos applications Amazon Kinesis Data Analytics for SQL. Support ne sera plus disponible pour Amazon Kinesis Data Analytics for SQL à partir de cette date. Pour de plus amples informations, veuillez consulter [Arrêt d'Amazon Kinesis Data Analytics pour les applications SQL](discontinuation.md).
Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.
# Exemples : Fenêtres et Regroupement
Cette section fournit des exemples d’applications Amazon Kinesis Data Analytics utilisant des requêtes à fenêtres et des requêtes de regroupement. (Pour plus d’informations, consultez [Requêtes à fenêtres](windowed-sql.md).) Chaque exemple fournit des step-by-step instructions et un exemple de code pour configurer l'application Kinesis Data Analytics.
**Topics**
+ [Exemple : Stagger Window](examples-window-stagger.md)
+ [Exemple : fenêtre bascule utilisant ROWTIME](examples-window-tumbling-rowtime.md)
+ [Exemple : fenêtre bascule utilisant un horodatage d'événement](examples-window-tumbling-event.md)
+ [Exemple : extraction des valeurs les plus fréquentes (TOP\$1K\$1ITEMS\$1TUMBLING)](examples-window-topkitems.md)
+ [Exemple : regroupement d'une partie des résultats à partir d'une requête](examples-window-partialresults.md)
# Exemple : Stagger Window
Lorsqu'une requête de fenêtre traite des fenêtres distinctes pour chaque clé de partition unique, à compter du moment où les données avec la clé correspondante arrivent, la fenêtre est appelée *fenêtre stagger*. Pour en savoir plus, consultez [Stagger Windows](stagger-window-concepts.md). Cet exemple Amazon Kinesis Data Analytics exemple utilise les colonnes EVENT\$1TIME et TICKER pour créer des fenêtres stagger. Le flux source contient des groupes de six enregistrements avec des valeurs EVENT\$1TIME et TICKER identiques qui se produisent au sein d'une période d'une minute, mais pas nécessairement avec la même valeur de minute (par exemple, `18:41:xx`).
Dans cet exemple, vous écrivez les enregistrements suivants dans un flux de données Kinesis aux heures suivantes. Le script n'écrit pas les temps dans le flux, mais le temps auquel l'enregistrement est intégré par l'application est écrit dans le champ `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
...
```
Vous créez ensuite une application Kinesis Data Analytics dans AWS Management Console le, avec le flux de données Kinesis comme source de diffusion. Le processus de découverte lit les exemples d'enregistrements sur la source de streaming et en déduit un schéma intégré à l'application avec deux colonnes (`EVENT_TIME` et `TICKER`), comme illustré ci-dessous.
![\[Capture d'écran de la console montrant le schéma intégré à l'application avec les colonnes de prix et de symbole boursier.\]](http://docs.aws.amazon.com/fr_fr/kinesisanalytics/latest/dev/images/ex_stagger_schema.png)
Vous utilisez le code de l'application à l'aide de la fonction `COUNT` pour créer un regroupement des données avec fenêtres. Vous insérez ensuite les données obtenues dans un autre flux intégré à l'application, comme indiqué dans la capture d'écran suivante :
![\[Capture d'écran de la console montrant les données obtenues dans un flux intégré à l'application.\]](http://docs.aws.amazon.com/fr_fr/kinesisanalytics/latest/dev/images/ex_stagger.png)
Dans la procédure suivante, vous créez une application Kinesis Data Analytics qui regroupe les valeurs dans le flux d’entrée dans une fenêtre stagger basée sur EVENT\$1TIME et TICKER.
**Topics**
+ [Étape 1 : Création d’un flux de données Kinesis](#examples-stagger-window-1)
+ [Étape 2 : Création d’une application Kinesis Data Analytics](#examples-stagger-window-2)
## Étape 1 : Création d’un flux de données Kinesis
Créez un flux de données Amazon Kinesis et remplissez les enregistrements comme suit :
1. [Connectez-vous à la console Kinesis AWS Management Console et ouvrez-la à https://console.aws.amazon.com l'adresse /kinesis.](https://console.aws.amazon.com/kinesis)
1. Choisissez **Data Streams (Flux de données)** dans le volet de navigation.
1. Choisissez **Create Kinesis stream (Créer un flux Kinesis)**, puis créez un flux avec une seule partition. Pour de plus amples informations, consultez [Créer un flux](https://docs.aws.amazon.com/streams/latest/dev/learning-kinesis-module-one-create-stream.html) dans le *Guide du développeur Amazon Kinesis Data Streams*.
1. Pour écrire des enregistrements sur un flux de données Kinesis dans un environnement de production, nous vous recommandons d'utiliser [Kinesis Producer Library](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-kpl.html) ou les [API de flux de données Kinesis](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-sdk.html). Pour plus de simplicité, cet exemple utilise le script Python ci-dessous pour générer des enregistrements. Exécutez le code pour remplir les exemples d'enregistrements du symbole boursier. Ce code simple écrit en continu un groupe de six enregistrements avec les mêmes valeurs `EVENT_TIME` et symbole boursier aléatoires dans le flux, durant une minute. Laissez le script s'exécuter pour pouvoir générer le schéma d'application lors d'une étape ultérieure.
```
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"))
```
## Étape 2 : Création d’une application Kinesis Data Analytics
Créez une application Kinesis Data Analytics comme suit :
1. Ouvrez le service géré pour la console Apache Flink à l'adresse [ https://console.aws.amazon.com/kinesisanalytics.](https://console.aws.amazon.com/kinesisanalytics)
1. Choisissez **Create application (Créer une application)**, saisissez un nom d'application, puis sélectionnez **Create application (Créer une application)**.
1. Sur la page de détails de l'application, choisissez **Connect streaming data (Connecter des données de diffusion)** pour vous connecter à la source.
1. Sur la page **Connect to source (Se connecter à la source)**, procédez comme suit :
1. Choisissez le flux que vous avez créé dans la section précédente.
1. Choisissez **Discover schema (Découvrir le schéma)**. Attendez que la console affiche le schéma déduit et les exemples d'enregistrements qui sont utilisés pour déduire le schéma pour le flux intégré à l'application créé. Le schéma déduit comporte deux colonnes.
1. Choisissez **Edit schema (Modifier le schéma)**. Remplacez le **Column type (Type de colonne)** de la colonne **EVENT\$1TIME** par `TIMESTAMP`.
1. Choisissez **Save schema and update stream samples (Enregistrer le schéma et mettre à jour les exemples de flux)**. Une fois que la console a enregistré le schéma, choisissez **Exit (Quitter)**.
1. Choisissez **Save and continue (Enregistrer et continuer)**.
1. Sur la page de détails de l'application, choisissez **Go to SQL editor (Accéder à l'éditeur SQL)**. Pour lancer l'application, choisissez **Yes, start application (Oui, démarrer l'application)** dans la boîte de dialogue qui s'affiche.
1. Dans l'éditeur SQL, écrivez le code d'application et vérifiez les résultats comme suit :
1. Copiez le code d'application suivant et collez-le dans l'éditeur.
```
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. Choisissez **Save and run SQL (Enregistrer et exécuter SQL)**.
Dans l'onglet **Real-time analytics (Analyse en temps réel)**, vous pouvez voir tous les flux intégrés à l'application que l'application a créés et vérifier les données.
# Exemple : fenêtre bascule utilisant ROWTIME
Lorsqu'une requête à fenêtres traite chaque fenêtre sans chevauchement, la fenêtre est appelée *fenêtre bascule*. Pour en savoir plus, consultez [Fenêtres bascules (regroupements à l'aide de GROUP BY)](tumbling-window-concepts.md). Cet exemple Amazon Kinesis Data Analytics utilise la colonne `ROWTIME` pour créer des fenêtres bascules. La colonne `ROWTIME` représente le moment où l'enregistrement a été lu par l'application.
Dans cet exemple, vous écrivez les enregistrements suivants dans un flux de données Kinesis.
```
{"TICKER": "TBV", "PRICE": 33.11}
{"TICKER": "INTC", "PRICE": 62.04}
{"TICKER": "MSFT", "PRICE": 40.97}
{"TICKER": "AMZN", "PRICE": 27.9}
...
```
Vous créez ensuite une application Kinesis Data Analytics dans AWS Management Console le, avec le flux de données Kinesis comme source de diffusion. Le processus de découverte lit les exemples d'enregistrements sur la source de streaming et en déduit un schéma intégré à l'application avec deux colonnes (`TICKER` et `PRICE`), comme illustré ci-dessous.
![\[Capture d'écran de la console montrant le schéma intégré à l'application avec les colonnes de prix et de symbole boursier.\]](http://docs.aws.amazon.com/fr_fr/kinesisanalytics/latest/dev/images/ex_tumbling_rowtime_schema.png)
Vous utilisez le code de l'application à l'aide du `MIN` et des fonctions `MAX` pour créer un regroupement des données avec fenêtres. Vous insérez ensuite les données obtenues dans un autre flux intégré à l'application, comme indiqué dans la capture d'écran suivante :
![\[Capture d'écran de la console montrant les données obtenues dans un flux intégré à l'application.\]](http://docs.aws.amazon.com/fr_fr/kinesisanalytics/latest/dev/images/ex_tumbling_rowtime.png)
Dans la procédure suivante, vous créez une application Kinesis Data Analytics qui regroupe les valeurs dans le flux d’entrée dans une fenêtre bascule basée sur ROWTIME.
**Topics**
+ [Étape 1 : Création d’un flux de données Kinesis](#examples-tumbling-window-1)
+ [Étape 2 : Création d’une application Kinesis Data Analytics](#examples-tumbling-window-2)
## Étape 1 : Création d’un flux de données Kinesis
Créez un flux de données Amazon Kinesis et remplissez les enregistrements comme suit :
1. [Connectez-vous à la console Kinesis AWS Management Console et ouvrez-la à https://console.aws.amazon.com l'adresse /kinesis.](https://console.aws.amazon.com/kinesis)
1. Choisissez **Data Streams (Flux de données)** dans le volet de navigation.
1. Choisissez **Create Kinesis stream (Créer un flux Kinesis)**, puis créez un flux avec une seule partition. Pour de plus amples informations, consultez [Créer un flux](https://docs.aws.amazon.com/streams/latest/dev/learning-kinesis-module-one-create-stream.html) dans le *Guide du développeur Amazon Kinesis Data Streams*.
1. Pour écrire des enregistrements sur un flux de données Kinesis dans un environnement de production, nous vous recommandons d'utiliser [Kinesis Client Library](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-kpl.html) ou les [API de flux de données Kinesis](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-sdk.html). Pour plus de simplicité, cet exemple utilise le script Python ci-dessous pour générer des enregistrements. Exécutez le code pour remplir les exemples d'enregistrements du symbole boursier. Ce code simple écrit de façon continue un enregistrement du symbole boursier aléatoire dans le flux. Laissez le script s'exécuter pour pouvoir générer le schéma d'application lors d'une étape ultérieure.
```
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"))
```
## Étape 2 : Création d’une application Kinesis Data Analytics
Créez une application Kinesis Data Analytics comme suit :
1. Ouvrez le service géré pour la console Apache Flink à l'adresse [ https://console.aws.amazon.com/kinesisanalytics.](https://console.aws.amazon.com/kinesisanalytics)
1. Choisissez **Créer une application**, saisissez un nom d'application, puis sélectionnez **Créer une application**.
1. Sur la page de détails de l'application, choisissez **Connect streaming data (Connecter des données de diffusion)** pour vous connecter à la source.
1. Sur la page **Connect to source (Se connecter à la source)**, procédez comme suit :
1. Choisissez le flux que vous avez créé dans la section précédente.
1. Choisissez **Discover schema (Découvrir le schéma)**. Attendez que la console affiche le schéma déduit et les exemples d'enregistrements qui sont utilisés pour déduire le schéma pour le flux intégré à l'application créé. Le schéma déduit comporte deux colonnes.
1. Choisissez **Save schema and update stream samples (Enregistrer le schéma et mettre à jour les exemples de flux)**. Une fois que la console a enregistré le schéma, choisissez **Exit (Quitter)**.
1. Choisissez **Save and continue (Enregistrer et continuer)**.
1. Sur la page de détails de l'application, choisissez **Go to SQL editor (Accéder à l'éditeur SQL)**. Pour lancer l'application, choisissez **Yes, start application (Oui, démarrer l'application)** dans la boîte de dialogue qui s'affiche.
1. Dans l'éditeur SQL, écrivez le code d'application et vérifiez les résultats comme suit :
1. Copiez le code d'application suivant et collez-le dans l'éditeur.
```
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. Choisissez **Save and run SQL (Enregistrer et exécuter SQL)**.
Dans l'onglet **Real-time analytics (Analyse en temps réel)**, vous pouvez voir tous les flux intégrés à l'application que l'application a créés et vérifier les données.
# Exemple : fenêtre bascule utilisant un horodatage d'événement
Lorsqu'une requête à fenêtres traite chaque fenêtre sans chevauchement, la fenêtre est appelée *fenêtre bascule*. Pour en savoir plus, consultez [Fenêtres bascules (regroupements à l'aide de GROUP BY)](tumbling-window-concepts.md). Cet exemple Amazon Kinesis Data Analytics illustre une fenêtre bascule qui utilise un horodatage d’événement, c’est-à-dire un horodatage créé par l’utilisateur et qui est compris dans les données de streaming. Il utilise cette approche plutôt que ROWTIME, qui est un horodatage que Kinesis Data Analytics crée lorsque l’application reçoit l’enregistrement. Vous devrez utiliser un horodatage d'événement dans les données de streaming si vous souhaitez créer un regroupement basé sur le moment où un événement s'est produit, plutôt que sur le moment où il a été reçu par l'application. Dans cet exemple, la valeur `ROWTIME` déclenche le regroupement chaque minute, et les enregistrements sont regroupés à la fois par `ROWTIME` et par heure d'événement.
Dans cet exemple, vous écrivez les enregistrements suivants dans un flux de données Amazon Kinesis. La valeur `EVENT_TIME` est fixée à 5 secondes dans le passé afin de simuler un délai de transmission et de traitement susceptible de créer un retard entre le moment où l’événement s’est produit et le moment où l’enregistrement est transféré vers 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}
...
```
Vous créez ensuite une application Kinesis Data Analytics dans AWS Management Console le, avec le flux de données Kinesis comme source de diffusion. Le processus de découverte lit les exemples d'enregistrements sur la source de streaming et en déduit un schéma intégré à l'application avec trois colonnes (`EVENT_TIME`, `TICKER`, et `PRICE`) comme illustré ci-dessous.
![\[Capture d'écran de la console montrant le schéma intégré à l'application avec les colonnes de l'heure de l'événement, de symbole boursier et de prix.\]](http://docs.aws.amazon.com/fr_fr/kinesisanalytics/latest/dev/images/ex_tumbling_event_schema.png)
Vous utilisez le code de l'application à l'aide du `MIN` et des fonctions `MAX` pour créer un regroupement des données avec fenêtres. Vous insérez ensuite les données obtenues dans un autre flux intégré à l'application, comme indiqué dans la capture d'écran suivante :
![\[Capture d'écran de la console montrant les données obtenues dans un flux intégré à l'application.\]](http://docs.aws.amazon.com/fr_fr/kinesisanalytics/latest/dev/images/ex_tumbling_event.png)
Dans la procédure suivante, vous créez une application Kinesis Data Analytics qui regroupe les valeurs dans le flux d’entrée dans une fenêtre bascule basée sur une heure d’événement.
**Topics**
+ [Étape 1 : Création d’un flux de données Kinesis](#examples-window-tumbling-event-1)
+ [Étape 2 : Création d’une application Kinesis Data Analytics](#examples-window-tumbling-event-2)
## Étape 1 : Création d’un flux de données Kinesis
Créez un flux de données Amazon Kinesis et remplissez les enregistrements comme suit :
1. [Connectez-vous à la console Kinesis AWS Management Console et ouvrez-la à https://console.aws.amazon.com l'adresse /kinesis.](https://console.aws.amazon.com/kinesis)
1. Choisissez **Data Streams (Flux de données)** dans le volet de navigation.
1. Choisissez **Create Kinesis stream (Créer un flux Kinesis)**, puis créez un flux avec une seule partition. Pour de plus amples informations, consultez [Créer un flux](https://docs.aws.amazon.com/streams/latest/dev/learning-kinesis-module-one-create-stream.html) dans le *Guide du développeur Amazon Kinesis Data Streams*.
1. Pour écrire des enregistrements sur un flux de données Kinesis dans un environnement de production, nous vous recommandons d'utiliser [Kinesis Client Library](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-kpl.html) ou les [API de flux de données Kinesis](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-sdk.html). Pour plus de simplicité, cet exemple utilise le script Python ci-dessous pour générer des enregistrements. Exécutez le code pour remplir les exemples d'enregistrements du symbole boursier. Ce code simple écrit de façon continue un enregistrement du symbole boursier aléatoire dans le flux. Laissez le script s'exécuter pour pouvoir générer le schéma d'application lors d'une étape ultérieure.
```
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"))
```
## Étape 2 : Création d’une application Kinesis Data Analytics
Créez une application Kinesis Data Analytics comme suit :
1. Ouvrez le service géré pour la console Apache Flink à l'adresse [ https://console.aws.amazon.com/kinesisanalytics.](https://console.aws.amazon.com/kinesisanalytics)
1. Choisissez **Créer une application**, saisissez un nom d'application, puis sélectionnez **Créer une application**.
1. Sur la page de détails de l'application, choisissez **Connect streaming data (Connecter des données de diffusion)** pour vous connecter à la source.
1. Sur la page **Connect to source (Se connecter à la source)**, procédez comme suit :
1. Choisissez le flux que vous avez créé dans la section précédente.
1. Choisissez **Discover schema (Découvrir le schéma)**. Attendez que la console affiche le schéma déduit et les exemples d'enregistrements qui sont utilisés pour déduire le schéma pour le flux intégré à l'application créé. Le schéma déduit comporte trois colonnes.
1. Choisissez **Edit schema (Modifier le schéma)**. Remplacez le **Column type (Type de colonne)** de la colonne **EVENT\$1TIME** par `TIMESTAMP`.
1. Choisissez **Save schema and update stream samples (Enregistrer le schéma et mettre à jour les exemples de flux)**. Une fois que la console a enregistré le schéma, choisissez **Exit (Quitter)**.
1. Choisissez **Save and continue (Enregistrer et continuer)**.
1. Sur la page de détails de l'application, choisissez **Go to SQL editor (Accéder à l'éditeur SQL)**. Pour lancer l'application, choisissez **Yes, start application (Oui, démarrer l'application)** dans la boîte de dialogue qui s'affiche.
1. Dans l'éditeur SQL, écrivez le code d'application et vérifiez les résultats comme suit :
1. Copiez le code d'application suivant et collez-le dans l'éditeur.
```
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. Choisissez **Save and run SQL (Enregistrer et exécuter SQL)**.
Dans l'onglet **Real-time analytics (Analyse en temps réel)**, vous pouvez voir tous les flux intégrés à l'application que l'application a créés et vérifier les données.
# Exemple : extraction des valeurs les plus fréquentes (TOP\$1K\$1ITEMS\$1TUMBLING)
Cet exemple Amazon Kinesis Data Analytics montre comment utiliser la fonction `TOP_K_ITEMS_TUMBLING` pour extraire les valeurs les plus fréquentes dans une fenêtre bascule. Pour plus d’informations, consultez la section [Fonction `TOP_K_ITEMS_TUMBLING`](https://docs.aws.amazon.com/kinesisanalytics/latest/sqlref/top-k.html) dans le manuel *Référence SQL du service géré Amazon pour Apache Flink*.
La fonction `TOP_K_ITEMS_TUMBLING` est utile pour regrouper des dizaines ou des centaines de milliers de clés et si vous souhaitez réduire l'utilisation de vos ressources. La fonction produit le même résultat qu'un regroupement avec les clauses `GROUP BY` et `ORDER BY` .
Dans cet exemple, vous écrivez les enregistrements suivants dans un flux de données Amazon Kinesis :
```
{"TICKER": "TBV"}
{"TICKER": "INTC"}
{"TICKER": "MSFT"}
{"TICKER": "AMZN"}
...
```
Vous créez ensuite une application Kinesis Data Analytics dans AWS Management Console le, avec le flux de données Kinesis comme source de diffusion. Le processus de découverte lit les exemples d'enregistrements de la source de streaming et en déduit un schéma intégré à l'application avec une colonne (`TICKER`), comme illustré ci-dessous.
![\[Capture d'écran de la console montrant le schéma intégré à l'application avec une colonne de symbole boursier.\]](http://docs.aws.amazon.com/fr_fr/kinesisanalytics/latest/dev/images/ex_topk_schema.png)
Vous utilisez le code de l'application à l'aide de la fonction `TOP_K_VALUES_TUMBLING` pour créer un regroupement des données avec fenêtres. Vous insérez ensuite les données obtenues dans un autre flux intégré à l'application, comme indiqué dans la capture d'écran suivante :
![\[Capture d'écran de la console montrant les données obtenues dans un flux intégré à l'application.\]](http://docs.aws.amazon.com/fr_fr/kinesisanalytics/latest/dev/images/ex_topk.png)
Dans la procédure suivante, vous créez une application Kinesis Data Analytics qui extrait les valeurs les plus fréquentes dans le flux d’entrée.
**Topics**
+ [Étape 1 : Création d’un flux de données Kinesis](#examples-window-topkitems-1)
+ [Étape 2 : Création d’une application Kinesis Data Analytics](#examples-window-topkitems-2)
## Étape 1 : Création d’un flux de données Kinesis
Créez un flux de données Amazon Kinesis et remplissez les enregistrements comme suit :
1. [Connectez-vous à la console Kinesis AWS Management Console et ouvrez-la à https://console.aws.amazon.com l'adresse /kinesis.](https://console.aws.amazon.com/kinesis)
1. Choisissez **Data Streams (Flux de données)** dans le volet de navigation.
1. Choisissez **Create Kinesis stream (Créer un flux Kinesis)**, puis créez un flux avec une seule partition. Pour de plus amples informations, consultez [Créer un flux](https://docs.aws.amazon.com/streams/latest/dev/learning-kinesis-module-one-create-stream.html) dans le *Guide du développeur Amazon Kinesis Data Streams*.
1. Pour écrire des enregistrements sur un flux de données Kinesis dans un environnement de production, nous vous recommandons d'utiliser [Kinesis Client Library](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-kpl.html) ou les [API de flux de données Kinesis](https://docs.aws.amazon.com/streams/latest/dev/developing-producers-with-sdk.html). Pour plus de simplicité, cet exemple utilise le script Python ci-dessous pour générer des enregistrements. Exécutez le code pour remplir les exemples d'enregistrements du symbole boursier. Ce code simple écrit de façon continue un enregistrement du symbole boursier aléatoire dans le flux. Laissez le script s'exécuter pour pouvoir générer le schéma d'application lors d'une étape ultérieure.
```
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"))
```
## Étape 2 : Création d’une application Kinesis Data Analytics
Créez une application Kinesis Data Analytics comme suit :
1. Ouvrez le service géré pour la console Apache Flink à l'adresse [ https://console.aws.amazon.com/kinesisanalytics.](https://console.aws.amazon.com/kinesisanalytics)
1. Choisissez **Create application (Créer une application)**, saisissez un nom d'application, puis sélectionnez **Create application (Créer une application)**.
1. Sur la page de détails de l'application, choisissez **Connect streaming data (Connecter des données de diffusion)** pour vous connecter à la source.
1. Sur la page **Connect to source (Se connecter à la source)**, procédez comme suit :
1. Choisissez le flux que vous avez créé dans la section précédente.
1. Choisissez **Discover schema (Découvrir le schéma)**. Attendez que la console affiche le schéma déduit et les exemples d'enregistrements qui sont utilisés pour déduire le schéma pour le flux intégré à l'application créé. Le schéma déduit comporte une colonne.
1. Choisissez **Save schema and update stream samples (Enregistrer le schéma et mettre à jour les exemples de flux)**. Une fois que la console a enregistré le schéma, choisissez **Exit (Quitter)**.
1. Choisissez **Save and continue (Enregistrer et continuer)**.
1. Sur la page de détails de l'application, choisissez **Go to SQL editor (Accéder à l'éditeur SQL)**. Pour lancer l'application, choisissez **Yes, start application (Oui, démarrer l'application)** dans la boîte de dialogue qui s'affiche.
1. Dans l'éditeur SQL, écrivez le code d'application et vérifiez les résultats comme suit :
1. Copiez le code d'application suivant et collez-le dans l'éditeur:
```
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. Choisissez **Save and run SQL (Enregistrer et exécuter SQL)**.
Dans l'onglet **Real-time analytics (Analyse en temps réel)**, vous pouvez voir tous les flux intégrés à l'application que l'application a créés et vérifier les données.
# Exemple : regroupement d'une partie des résultats à partir d'une requête
Si un flux de données Amazon Kinesis contient des enregistrements qui disposent d’une heure d’événement ne correspondant pas exactement à l’heure de l’intégration, une sélection de résultats dans une fenêtre bascule contiendra, dans la fenêtre, les enregistrements qui sont arrivés mais qui ne se sont pas nécessairement produits. Dans ce cas, la fenêtre bascule contient uniquement une partie des résultats que vous souhaitez. Vous pouvez utiliser plusieurs approches pour résoudre ce problème :
+ Utilisez uniquement une fenêtre bascule et regroupez une partie des résultats dans le traitement ultérieur via une base de données ou un entrepôt de données utilisant des opérations de « mises à jour/insertions ». Cette approche est efficace dans le traitement d'une application. Elle gère les données tardives indéfiniment pour les opérateurs d'agrégation (`sum`, `min`, `max`, etc.). L'inconvénient de cette approche est que vous devez développer et maintenir une logique d'application supplémentaire dans la couche de base de données.
+ Utilisez une fenêtre bascule et défilante qui produit une partie des résultats tôt, mais continue également à produire des résultats complets au cours de la période de la fenêtre défilante. Cette approche gère les données tardives grâce à un remplacement plutôt qu'une « mise à jour/insertion » afin qu'aucune autre logique d'application n'ait besoin d'être ajoutée dans la couche de base de données. L'inconvénient de cette approche est qu'elle utilise davantage d'unités de traitement Kinesis KPUs () et qu'elle produit toujours deux résultats, ce qui peut ne pas fonctionner dans certains cas d'utilisation.
Pour plus d'informations sur les fenêtres bascules et défilantes, consultez [Requêtes à fenêtres](windowed-sql.md).
Dans la procédure suivante, le regroupement de la fenêtre bascule génère deux résultats partiels (envoyés au `CALC_COUNT_SQL_STREAM` dans le flux intégré à l'application) qui doivent être combinés pour produire un résultat final. L'application génère ensuite un second regroupement (envoyé au `DESTINATION_SQL_STREAM` dans le flux intégré à l'application) qui combine les deux résultats partiels.
**Pour créer une application qui regroupe une partie des résultats à l'aide d'une heure d'évènement**
1. [Connectez-vous à la console Kinesis AWS Management Console et ouvrez-la à https://console.aws.amazon.com l'adresse /kinesis.](https://console.aws.amazon.com/kinesis)
1. Choisissez **Data Analytics (Analyse des données)** dans le volet de navigation. Créez une application Kinesis Data Analytics comme décrit dans le didacticiel [Mise en route avec les applications Amazon Kinesis Data Analytics pour SQL](getting-started.md).
1. Dans l'éditeur SQL, remplacez le code d'application par les éléments suivants :
```
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);
```
L'instruction `SELECT` du code d'application filtre les lignes de `SOURCE_SQL_STREAM_001` pour obtenir les changements de cours d'action supérieurs à 1 % et insère ces lignes dans un autre flux intégré à l'application, `CHANGE_STREAM`, à l'aide d'une pompe.
1. Choisissez **Save and run SQL (Enregistrer et exécuter SQL)**.
La première pompe génère un flux vers `CALC_COUNT_SQL_STREAM` similaire à ce qui suit. Notez que l'ensemble des résultats est incomplet :
![\[Capture d'écran de la console montrant une partie des résultats.\]](http://docs.aws.amazon.com/fr_fr/kinesisanalytics/latest/dev/images/ex_partial_0.png)
La deuxième pompe génère ensuite un flux vers `DESTINATION_SQL_STREAM` contenant l'ensemble des résultats :
![\[Capture d'écran de la console montrant l'ensemble des résultats.\]](http://docs.aws.amazon.com/fr_fr/kinesisanalytics/latest/dev/images/ex_partial_1.png)