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.
# Advanced (Avancé) AWS Glue concepts de streaming
Dans les applications modernes axées sur les données, l’importance des données diminue au fil du temps et leur valeur passe d’une valeur prédictive à une valeur réactive. Par conséquent, les clients souhaitent traiter les données en temps réel pour prendre des décisions plus rapidement. Lorsqu’il s’agit de flux de données en temps réel, tels que ceux provenant de capteurs IoT, les données peuvent arriver de manière désordonnée ou subir des retards de traitement en raison de la latence du réseau ou d’autres défaillances liées à la source lors de l’ingestion. Dans le cadre de la plateforme AWS Glue, le streaming AWS Glue s’appuie sur ces fonctionnalités pour fournir un ETL de streaming évolutif et sans serveur, alimenté par le streaming structuré Apache Spark, permettant aux utilisateurs de traiter les données en temps réel.
Dans cette rubrique, nous allons explorer les concepts de streaming avancés et les fonctionnalités du streaming AWS Glue.
## Considérations temporelles lors du traitement des flux
Il existe quatre notions de temps lors du traitement des flux :
+ **Event-time**— L'heure à laquelle l'événement s'est produit. Dans la plupart des cas, ce champ est intégré aux données de l’événement elles-mêmes, à la source.
+ **Event-time-window**— L'intervalle de temps entre deux événements. Comme le montre le schéma ci-dessus, **W1** est une fenêtre horaire d’événement comprise entre 17 h 00 et 17 h 10. Chaque fenêtre horaire d’événement est un regroupement de plusieurs événements.
+ **Trigger-time**— Le délai de déclenchement contrôle la fréquence du traitement des données et de la mise à jour des résultats. Heure à laquelle le traitement par microlots a commencé.
+ **Ingestion-time**— Heure à laquelle les données du flux ont été ingérées dans le service de streaming. Si l’heure de l’événement n’est pas intégrée à l’événement lui-même, cette heure peut être utilisée pour le fenêtrage dans certains cas.
## Fenêtrage
Le fenêtrage est une technique qui permet de regrouper et d’agréger plusieurs événements par fenêtre horaire d’événement. Dans les exemples suivants, nous découvrirons les avantages du fenêtrage et les circonstances dans lesquelles vous l’utiliseriez.
Selon le cas d’utilisation métier, Spark prend en charge trois types de fenêtres temporelles.
+ Fenêtre bascule : série de fenêtres horaires d’événements de taille fixe qui ne se chevauchent pas et que vous agrégez.
+ Fenêtre défilante : similaire aux fenêtres bascules en ce sens qu’elles sont « de taille fixe », mais les fenêtres peuvent se chevaucher ou défiler tant que la durée du défilement est inférieure à la durée de la fenêtre elle-même.
+ Fenêtre de session : commence par un événement de données d’entrée et continue à s’étendre tant qu’elle reçoit des entrées dans un intervalle ou pendant une durée d’inactivité. La longueur d’une fenêtre de session peut être statique ou dynamique, selon les entrées.
### Fenêtre bascule
La fenêtre bascule est une série de fenêtres horaires d’événements de taille fixe qui ne se chevauchent pas et que vous agrégez. Penchons-nous sur un exemple concret pour comprendre.
La société ABC Auto souhaite réaliser une campagne marketing pour une nouvelle marque de voitures de sport. Elle veut choisir une ville regroupant le plus de fans de voitures de sport. Pour atteindre cet objectif, elle publie une courte publicité de 15 secondes présentant la voiture sur son site Web. Tous les « clics » et la « ville » correspondante sont enregistrés et diffusés. Amazon Kinesis Data Streams Nous voulons compter le nombre de clics dans une fenêtre de 10 minutes et le regrouper par ville pour voir quelle ville est la plus demandée. Voici la sortie de l’agrégation.
| window\_start\_time | window\_end\_time | city | total\_clicks |
| --- | --- | --- | --- |
| 10/07/2023 17:00:00 | 10/07/2023 17:10:00 | Dallas | 75 |
| 10/07/2023 17:00:00 | 10/07/2023 17:10:00 | Chicago | 10 |
| 10/07/2023 17:20:00 | 10/07/2023 17:30:00 | Dallas | 20 |
| 10/07/2023 17:20:00 | 10/07/2023 17:30:00 | Chicago | 50 |
Comme expliqué ci-dessus, ces fenêtres horaires d’événements sont différentes des intervalles d’heure de déclenchement. Par exemple, même si votre heure de déclenchement est toutes les minutes, les résultats en sortie n’afficheront que des fenêtres d’agrégation de 10 minutes sans chevauchement. Pour l’optimisation, il est préférable d’aligner l’intervalle de déclenchement sur la fenêtre horaire d’événement.
Dans le tableau ci-dessus, Dallas a enregistré 75 clics au cours de la fenêtre 17 h 00 – 17 h 10, tandis que Chicago a enregistré 10 clics. De plus, il n’y a aucune donnée pour la fenêtre 17 h 10 – 17 h 20 pour aucune ville, donc cette fenêtre est omise.
Vous pouvez désormais effectuer une analyse plus approfondie de ces données dans l’application d’analytique en aval afin de déterminer la ville la plus privilégiée pour mener la campagne marketing.
**Utiliser des fenêtres qui s'enfoncent AWS Glue**
1. Créez un Amazon Kinesis Data Streams DataFrame et lisez-le. Exemple :
```
parsed_df = kinesis_raw_df \
.selectExpr('CAST(data AS STRING)') \
.select(from_json("data", ticker_schema).alias("data")) \
.select('data.event_time','data.ticker','data.trade','data.volume', 'data.price')
```
1. Traitez les données dans une fenêtre bascule. Dans l’exemple ci-dessous, les données sont regroupées en fonction du champ de saisie « event\_time » dans des fenêtres bascules de 10 minutes et en écrivant le résultat dans un lac de données Amazon S3.
```
grouped_df = parsed_df \
.groupBy(window("event_time", "10 minutes"), "city") \
.agg(sum("clicks").alias("total_clicks"))
summary_df = grouped_df \
.withColumn("window_start_time", col("window.start")) \
.withColumn("window_end_time", col("window.end")) \
.withColumn("year", year("window_start_time")) \
.withColumn("month", month("window_start_time")) \
.withColumn("day", dayofmonth("window_start_time")) \
.withColumn("hour", hour("window_start_time")) \
.withColumn("minute", minute("window_start_time")) \
.drop("window")
write_result = summary_df \
.writeStream \
.format("parquet") \
.trigger(processingTime="10 seconds") \
.option("checkpointLocation", "s3a://bucket-stock-stream/stock-stream-catalog-job/checkpoint/") \
.option("path", "s3a://bucket-stock-stream/stock-stream-catalog-job/summary_output/") \
.partitionBy("year", "month", "day") \
.start()
```
### Fenêtre défilante
Les fenêtres défilantes sont similaires aux fenêtres bascules en ce sens qu’elles sont « de taille fixe », mais les fenêtres peuvent se chevaucher ou défiler tant que la durée du défilement est inférieure à la durée de la fenêtre elle-même. En raison de la nature du défilement, une entrée peut être liée à plusieurs fenêtres.
Pour mieux comprendre, prenons l’exemple d’une banque qui souhaite détecter une éventuelle fraude par carte de crédit. Une application en streaming pourrait surveiller un flux continu de transactions par carte de crédit. Ces transactions pourraient être agrégées dans des fenêtres d’une durée de 10 minutes et toutes les 5 minutes, la fenêtre avancerait, éliminant les 5 minutes de données les plus anciennes et ajoutant les 5 dernières minutes de nouvelles données. Dans chaque fenêtre, les transactions pourraient être regroupées par pays pour vérifier l’absence de schémas suspects, comme une transaction aux États-Unis immédiatement suivie d’une autre en Australie. Par souci de simplicité, qualifions ces transactions de fraude lorsque le montant total des transactions est supérieur à 100 USD. Si un tel schéma est détecté, cela indique une fraude potentielle et la carte pourrait être gelée.
Le système de traitement des cartes de crédit envoie un flux d’événements de transaction à Kinesis pour chaque numéro de carte et pour le pays. Une AWS Glue tâche exécute l'analyse et produit le résultat agrégé suivant.
| window\_start\_time | window\_end\_time | card\_last\_four | country | total\_amount |
| --- | --- | --- | --- | --- |
| 10/07/2023 17:00:00 | 10/07/2023 17:10:00 | 6544 | ETATS-UNIS | 85 |
| 10/07/2023 17:00:00 | 10/07/2023 17:10:00 | 6544 | Australie | 10 |
| 10/07/2023 17:05:45 | 10/07/2023 17:15:45 | 6544 | ETATS-UNIS | 50 |
| 10/07/2023 17:10:45 | 10/07/2023 17:20:45 | 6544 | ETATS-UNIS | 50 |
| 10/07/2023 17:10:45 | 10/07/2023 17:20:45 | 6544 | Australie | 150 |
Sur la base de l’agrégation ci-dessus, vous pouvez voir la fenêtre de 10 minutes défiler toutes les 5 minutes, résumée au montant de la transaction. L’anomalie est détectée dans la fenêtre de 17 h 10 à 17 h 20 lorsqu’il y a une valeur aberrante, c’est-à-dire une transaction de 150 USD en Australie. AWS Glue peut détecter cette anomalie et envoyer un événement d’alerte avec la clé incriminée vers une rubrique SNS à l’aide de boto3. Par la suite, une fonction Lambda peut s’abonner à cette rubrique et prendre des mesures.
#### Traitement des données dans une fenêtre défilante
La clause `group-by` et la fonction de fenêtre sont utilisées pour implémenter la fenêtre défilante comme indiqué ci-dessous.
```
grouped_df = parsed_df \
.groupBy(window(col("event_time"), "10 minute", "5 min"), "country", "card_last_four") \
.agg(sum("tx_amount").alias("total_amount"))
```
### Fenêtre de session
Contrairement aux deux fenêtres ci-dessus qui ont une taille fixe, la longueur d’une fenêtre de session peut être statique ou dynamique, selon les entrées. Une fenêtre de session commence par un événement de données d’entrée et continue à s’étendre tant qu’elle reçoit des entrées dans un intervalle ou pendant une durée d’inactivité.
Prenons un exemple. La société ABC Hotel souhaite savoir quelle est la période la plus chargée de la semaine et proposer de meilleures offres à ses clients. Dès qu’un invité s’enregistre, une fenêtre de session est ouverte et Spark conserve un état avec agrégation pour cette fenêtre horaire d’événement. Chaque fois qu'un invité s'enregistre, un événement est généré et envoyé à Amazon Kinesis Data Streams. L’hôtel décide que s’il n’y a pas d’enregistrement pendant une période de 15 minutes, la fenêtre horaire d’événement peut être fermée. La fenêtre horaire d’événement suivante recommencera lors d’un nouvel enregistrement. La sortie ressemble à ce qui suit.
| window\_start\_time | window\_end\_time | city | total\_checkins |
| --- | --- | --- | --- |
| 2023-07-10 17:02:00 | 10/07/2023 17:30:00 | Dallas | 50 |
| 2023-07-10 17:02:00 | 10/07/2023 17:30:00 | Chicago | 25 |
| 10/07/2023 17:40:00 | 10/07/2023 18:20:00 | Dallas | 75 |
| 10/07/2023 18:50:45 | 10/07/2023 19:15:45 | Dallas | 20 |
Le premier enregistrement a eu lieu à event\_time = 17:02 (17 h 02). La fenêtre horaire d’événement d’agrégation débutera à 17 h 02. Cette agrégation se poursuivra tant que nous recevrons les événements dans un délai de 15 minutes. Dans l’exemple ci-dessus, le dernier événement que nous avons reçu a eu lieu à 17 h 15, puis il n’y a eu aucun événement pendant les 15 minutes qui ont suivi. Par conséquent, Spark a fermé cette fenêtre horaire d’événement à 17 h 15 \+ 15 min = 17 h 30 et l’a définie comme étant de 17 h 02 à 17 h 30. Il a ouvert une nouvelle fenêtre horaire d’événement à 17 h 47 lorsqu’il a reçu un nouvel événement relatif aux données d’enregistrement.
#### Traitement des données dans une fenêtre de session
La clause `group-by` et la fonction de fenêtre sont utilisées pour implémenter la fenêtre défilante.
```
grouped_df = parsed_df \
.groupBy(session_window(col("event_time"), "10 minute"), "city") \
.agg(count("check_in").alias("total_checkins"))
```
### Modes de sortie
Le mode de sortie est le mode dans lequel les résultats de la table illimitée sont écrits sur le récepteur externe. Trois modes sont disponibles. Dans l’exemple suivant, vous comptez les occurrences d’un mot alors que des lignes de données sont diffusées et traitées dans chaque microlot.
+ **Mode complet** : l’intégralité de la table des résultats sera écrite dans le récepteur après chaque traitement par microlots, même si le nombre de mots n’a pas été mis à jour dans la fenêtre horaire d’événement en cours.
+ **Mode d’ajout** : il s’agit du mode par défaut, dans lequel seuls les nouveaux mots et/ou les nouvelles lignes ajoutés à la table de résultats depuis le dernier déclenchement seront écrits dans le récepteur. Ce mode est idéal pour le streaming sans état pour des requêtes telles que map, flatMap, filter, etc.
+ **Mode de mise à jour** : seuls les mots et/ou les lignes de la table de résultats qui ont été mis à jour ou ajoutés depuis le dernier déclenchement seront écrits dans le récepteur.
**Note**
Mode de sortie = la « mise à jour » n’est pas prise en charge pour les fenêtres de session.
## Gestion des données tardives et des filigranes
Lorsque vous travaillez avec des données en temps réel, l’arrivée des données peut être retardée en raison de la latence du réseau et de défaillances en amont. Nous avons besoin d’un mécanisme pour effectuer à nouveau l’agrégation sur la fenêtre horaire d’événement manquée. Cependant, pour ce faire, l’état doit être maintenu. Dans le même temps, les anciennes données doivent être nettoyées pour limiter la taille de l’état. La version 2.1 de Spark a ajouté la prise en charge d’une fonctionnalité appelée filigrane qui maintient l’état et permet à l’utilisateur de spécifier le seuil pour les données tardives.
En référence à notre exemple de symbole boursier ci-dessus, considérons que le seuil autorisé pour les données tardives ne doit pas dépasser 10 minutes. Pour faire simple, nous supposerons que la fenêtre est à bascule, que le symbole est AMZ et que la négociation est BUY.
Dans le schéma ci-dessus, nous calculons le volume total sur une fenêtre bascule de 10 minutes. Nous avons le déclenchement à 17 h 00, 17 h 10 et 17 h 20. Au-dessus de la flèche chronologique, nous avons le flux de données d’entrée et ci-dessous se trouve la table des résultats illimités.
Au cours de la première fenêtre bascule de 10 minutes, nous avons agrégé en fonction de event\_time et le total\_volume a été calculé comme étant 30. Dans la deuxième fenêtre horaire d’événement, Spark a obtenu le premier événement de données avec event\_time = 17:02 (17 h 02). Comme il s’agit de la valeur event\_time maximale observée jusqu’à présent par Spark, le seuil du filigrane est défini il y a 10 minutes (c’est-à-dire watermark\_event\_time = 16:52). Tout événement de données avec un event\_time après 16:52 sera pris en compte pour l’agrégation limitée dans le temps et tout événement de données antérieur sera supprimé. Cela permet à Spark de maintenir un état intermédiaire pendant 10 minutes supplémentaires pour tenir compte des données tardives. Vers 17 h 08, durée chronométrée, Spark a reçu un événement avec un event\_time = 16:54, ce qui était dans les limites du seuil. Spark a donc recalculé la fenêtre horaire d’événement « 16 h 50 – 17 h 00 » et le volume total a été mis à jour de 30 à 60.
Cependant, à l’heure de déclenchement 17 h 20, lorsque Spark a reçu un événement avec event\_time = 17:15, il a défini le watermark\_event\_time = 17:05. Par conséquent, l’événement de données tardives avec event\_time = 17:03 a été considéré comme « trop tardif » et ignoré.
```
Watermark Boundary = Max(Event Time) - Watermark Threshold
```
### Utilisation de filigranes dans AWS Glue
Spark n’émet ni n’écrit les données sur le récepteur externe tant que la limite du filigrane n’est pas dépassée. Pour implémenter un filigrane dans AWS Glue, consultez l’exemple ci-dessous.
```
grouped_df = parsed_df \
.withWatermark("event_time", "10 minutes") \
.groupBy(window("event_time", "5 minutes"), "ticker") \
.agg(sum("volume").alias("total_volume"))
```