Avis de fin de support : le 7 octobre 2026, AWS le support pour AWS IoT Greengrass Version 1. Après le 7 octobre 2026, vous ne pourrez plus accéder aux AWS IoT Greengrass V1 ressources. Pour plus d'informations, rendez-vous sur Migrer depuis AWS IoT Greengrass Version 1.

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.

Configurations d'exportation pour les applications prises AWS Cloud destinations

User-defined Les fonctions Lambda sont utilisées StreamManagerClient dans le SDK AWS IoT Greengrass Core pour interagir avec le gestionnaire de flux. Lorsqu'une fonction Lambda crée un flux ou met à jour un flux, elle transmet un MessageStreamDefinition objet qui représente les propriétés du flux, y compris la définition d'exportation. L'ExportDefinitionobjet contient les configurations d'exportation définies pour le flux. Le gestionnaire de flux utilise ces configurations d'exportation pour déterminer où et comment exporter le flux.

Vous pouvez définir zéro ou plusieurs configurations d'exportation sur un flux, y compris plusieurs configurations d'exportation pour un seul type de destination. Par exemple, vous pouvez exporter un flux vers deux flux de données Kinesis et un compartiment Amazon S3.

En cas d'échec des tentatives d'exportation, le gestionnaire de flux réessaie continuellement d'exporter les données vers le, à des AWS Cloud intervalles allant jusqu'à cinq minutes. Le nombre de nouvelles tentatives n'a pas de limite maximale.

Vous êtes responsable de la maintenance de ces AWS Cloud ressources.

AWS IoT Analytics chaînes (discontinuées)

Pour supprimer les configurations AWS IoT Analytics d'exportation d'un flux existant, mettez à jour la définition de votre flux de manière à omettre cette IoTAnalyticsConfig entrée. Vous pouvez également supprimer et recréer le flux sans la configuration AWS IoT Analytics d'exportation.

Envisagez de migrer vers l'une des autres destinations d'exportation prises en charge, telles que Flux de données Amazon Kinesis ouObjets Amazon S3.

 

Flux de données Amazon Kinesis

Stream Manager prend en charge les exportations automatiques vers Amazon Kinesis Data Streams. Kinesis Data Streams est couramment utilisé pour agréger de gros volumes de données et les charger dans un entrepôt de données ou un cluster de réduction de cartes. Pour plus d'informations, consultez Qu'est-ce qu'Amazon Kinesis Data Streams ? dans le guide du développeur Amazon Kinesis.

Dans le SDK AWS IoT Greengrass Core, vos fonctions Lambda utilisent le KinesisConfig pour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence du SDK correspondant à votre langue cible :

Exigences

Cette destination d'exportation répond aux exigences suivantes :

Exportation vers Kinesis Data Streams

Pour créer un flux exporté vers Kinesis Data Streams, vos fonctions Lambda créent un flux avec une définition d'exportation qui inclut un ou plusieurs KinesisConfig objets. Cet objet définit les paramètres d'exportation, tels que le flux de données cible, la taille du lot, l'intervalle entre les lots et la priorité.

Lorsque vos fonctions Lambda reçoivent des données provenant d'appareils, elles ajoutent des messages contenant une masse de données au flux cible. Stream Manager exporte ensuite les données en fonction des paramètres de lot et de la priorité définis dans les configurations d'exportation du flux.

Stream Manager génère un UUID aléatoire unique en tant que clé de partition pour chaque enregistrement chargé sur Amazon Kinesis.

 

AWS IoT SiteWise propriétés des actifs

Stream Manager prend en charge les exportations automatiques vers AWS IoT SiteWise. AWS IoT SiteWise vous permet de collecter, d'organiser et d'analyser des données provenant d'équipements industriels à grande échelle. Pour plus d'informations, voir Qu'est-ce que c'est AWS IoT SiteWise ? dans le Guide de AWS IoT SiteWise l'utilisateur.

Dans le SDK AWS IoT Greengrass Core, vos fonctions Lambda utilisent le IoTSiteWiseConfig pour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence du SDK correspondant à votre langue cible :

Exigences

Cette destination d'exportation répond aux exigences suivantes :

Exporter vers AWS IoT SiteWise

Pour créer un flux qui exporte vers AWS IoT SiteWise, vos fonctions Lambda créent un flux avec une définition d'exportation qui inclut un ou plusieurs IoTSiteWiseConfig objets. Cet objet définit les paramètres d'exportation, tels que la taille du lot, l'intervalle entre les lots et la priorité.

Lorsque vos fonctions Lambda reçoivent des données sur les propriétés des actifs des appareils, elles ajoutent des messages contenant les données au flux cible. Les messages sont JSON-serialized PutAssetPropertyValueEntry des objets qui contiennent des valeurs de propriété pour une ou plusieurs propriétés d'actifs. Pour plus d'informations, voir Ajouter un message pour les destinations AWS IoT SiteWise d'exportation.

Stream Manager exporte ensuite les données en fonction des paramètres de lot et de la priorité définis dans les configurations d'exportation du flux.

 

Vous pouvez ajuster les paramètres de votre gestionnaire de flux et la logique de la fonction Lambda pour concevoir votre stratégie d'exportation. Par exemple :

 

Objets Amazon S3

Stream Manager prend en charge les exportations automatiques vers Amazon S3. Vous pouvez utiliser Amazon S3 pour stocker et récupérer de grandes quantités de données. Pour plus d'informations, consultez Qu'est-ce qu'Amazon S3 ? dans le guide du développeur Amazon Simple Storage Service.

Dans le SDK AWS IoT Greengrass Core, vos fonctions Lambda utilisent le S3ExportTaskExecutorConfig pour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence du SDK correspondant à votre langue cible :

Exigences

Cette destination d'exportation répond aux exigences suivantes :

Exporter vers Amazon S3

Pour créer un flux exporté vers Amazon S3, vos fonctions Lambda utilisent l'S3ExportTaskExecutorConfigobjet pour configurer la politique d'exportation. La politique définit les paramètres d'exportation, tels que le seuil et la priorité de téléchargement en plusieurs parties. Pour les exportations Amazon S3, le gestionnaire de flux télécharge les données qu'il lit à partir de fichiers locaux sur l'appareil principal. Pour lancer un téléchargement, vos fonctions Lambda ajoutent une tâche d'exportation au flux cible. La tâche d'exportation contient des informations sur le fichier d'entrée et l'objet Amazon S3 cible. Le gestionnaire de flux exécute les tâches dans l'ordre dans lequel elles sont ajoutées au flux.

Ce flux de travail de haut niveau est illustré dans le schéma suivant.

Stream Manager utilise la propriété de seuil de téléchargement en plusieurs parties, le paramètre de taille de partie minimale et la taille du fichier d'entrée pour déterminer comment télécharger les données. Le seuil de téléchargement en plusieurs parties doit être supérieur ou égal à la taille minimale des parties. Si vous souhaitez télécharger des données en parallèle, vous pouvez créer plusieurs flux.

Les clés qui spécifient vos objets Amazon S3 cibles peuvent inclure des DateTimeFormatter chaînes Java valides dans des !{timestamp:value} espaces réservés. Vous pouvez utiliser ces espaces réservés d'horodatage pour partitionner les données dans Amazon S3 en fonction de l'heure à laquelle les données du fichier d'entrée ont été chargées. Par exemple, le nom de clé suivant se résout en une valeur telle quemy-key/2020/12/31/data.txt.

my-key/!{timestamp:YYYY}/!{timestamp:MM}/!{timestamp:dd}/data.txt

Gérer les données d'entrée

Vous pouvez créer du code que les applications IoT utilisent pour gérer le cycle de vie des données d'entrée. L'exemple de flux de travail suivant montre comment utiliser les fonctions Lambda pour gérer ces données.

Surveiller les tâches d'exportation

Vous pouvez créer du code que les applications IoT utilisent pour surveiller l'état de vos exportations Amazon S3. Vos fonctions Lambda doivent créer un flux d'état, puis configurer le flux d'exportation pour écrire des mises à jour de statut dans le flux de statut. Un seul flux de statut peut recevoir des mises à jour de statut provenant de plusieurs flux exportés vers Amazon S3.

Tout d'abord, créez un flux à utiliser comme flux d'état. Vous pouvez configurer la taille et les politiques de rétention du flux afin de contrôler la durée de vie des messages d'état. Par exemple :

Ensuite, créez ou mettez à jour le flux d'exportation pour utiliser le flux d'état. Spécifiquement, définissez la propriété de configuration status de la configuration S3ExportTaskExecutorConfig d'exportation du flux. Cela indique au gestionnaire de flux d'écrire des messages d'état concernant les tâches d'exportation vers le flux d'état. Dans l'StatusConfigobjet, spécifiez le nom du flux d'état et le niveau de verbosité. Les valeurs prises en charge suivantes vont de la plus simple (ERROR) à la plus détaillée (). TRACE La valeur par défaut est INFO.

 

L'exemple de flux de travail suivant montre comment les fonctions Lambda peuvent utiliser un flux d'état pour surveiller l'état des exportations.

L'exemple suivant montre comment une fonction Lambda peut lire et traiter les messages d'état.

Python

import time
from greengrasssdk.stream_manager import (
    ReadMessagesOptions,
    Status,
    StatusConfig,
    StatusLevel,
    StatusMessage,
    StreamManagerClient,
)
from greengrasssdk.stream_manager.util import Util

client = StreamManagerClient()
 
try:
    # Read the statuses from the export status stream
    is_file_uploaded_to_s3 = False
    while not is_file_uploaded_to_s3:
        try:
            messages_list = client.read_messages(
                "StatusStreamName", ReadMessagesOptions(min_message_count=1, read_timeout_millis=1000)
            )
            for message in messages_list:
                # Deserialize the status message first.
                status_message = Util.deserialize_json_bytes_to_obj(message.payload, StatusMessage)

                # Check the status of the status message. If the status is "Success",
                # the file was successfully uploaded to S3.
                # If the status was either "Failure" or "Cancelled", the server was unable to upload the file to S3.
                # We will print the message for why the upload to S3 failed from the status message.
                # If the status was "InProgress", the status indicates that the server has started uploading
                # the S3 task.
                if status_message.status == Status.Success:
                    logger.info("Successfully uploaded file at path " + file_url + " to S3.")
                    is_file_uploaded_to_s3 = True
                elif status_message.status == Status.Failure or status_message.status == Status.Canceled:
                    logger.info(
                        "Unable to upload file at path " + file_url + " to S3. Message: " + status_message.message
                    )
                    is_file_uploaded_to_s3 = True
            time.sleep(5)
        except StreamManagerException:
            logger.exception("Exception while running")
except StreamManagerException:
    pass
    # Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
    pass
    # Properly handle errors.

Référence du SDK Python : read_messages | StatusMessage

Java

import com.amazonaws.greengrass.streammanager.client.StreamManagerClient;
import com.amazonaws.greengrass.streammanager.client.utils.ValidateAndSerialize;
import com.amazonaws.greengrass.streammanager.model.ReadMessagesOptions;
import com.amazonaws.greengrass.streammanager.model.Status;
import com.amazonaws.greengrass.streammanager.model.StatusConfig;
import com.amazonaws.greengrass.streammanager.model.StatusLevel;
import com.amazonaws.greengrass.streammanager.model.StatusMessage;

try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
    try {
        boolean isS3UploadComplete = false;
        while (!isS3UploadComplete) {
            try {
                // Read the statuses from the export status stream
                List<Message> messages = client.readMessages("StatusStreamName",
                    new ReadMessagesOptions().withMinMessageCount(1L).withReadTimeoutMillis(1000L));
                for (Message message : messages) {
                    // Deserialize the status message first.
                    StatusMessage statusMessage = ValidateAndSerialize.deserializeJsonBytesToObj(message.getPayload(), StatusMessage.class);
                    // Check the status of the status message. If the status is "Success", the file was successfully uploaded to S3.
                    // If the status was either "Failure" or "Canceled", the server was unable to upload the file to S3.
                    // We will print the message for why the upload to S3 failed from the status message.
                    // If the status was "InProgress", the status indicates that the server has started uploading the S3 task.
                    if (Status.Success.equals(statusMessage.getStatus())) {
                        System.out.println("Successfully uploaded file at path " + FILE_URL + " to S3.");
                        isS3UploadComplete = true;
                     } else if (Status.Failure.equals(statusMessage.getStatus()) || Status.Canceled.equals(statusMessage.getStatus())) {
                        System.out.println(String.format("Unable to upload file at path %s to S3. Message %s",
                            statusMessage.getStatusContext().getS3ExportTaskDefinition().getInputUrl(),
                            statusMessage.getMessage()));
                        sS3UploadComplete = true;
                    }
                }
            } catch (StreamManagerException ignored) {
            } finally {
                // Sleep for sometime for the S3 upload task to complete before trying to read the status message.
                Thread.sleep(5000);
            }
        } catch (e) {
        // Properly handle errors.
    }
} catch (StreamManagerException e) {
    // Properly handle exception.
}

Référence du SDK Java : ReadMessages | StatusMessage

Node.js

const {
    StreamManagerClient, ReadMessagesOptions,
    Status, StatusConfig, StatusLevel, StatusMessage,
    util,
} = require('aws-greengrass-core-sdk').StreamManager;

const client = new StreamManagerClient();
client.onConnected(async () => {
    try {
        let isS3UploadComplete = false;
        while (!isS3UploadComplete) {
            try {
                // Read the statuses from the export status stream
                const messages = await c.readMessages("StatusStreamName",
                    new ReadMessagesOptions()
                        .withMinMessageCount(1)
                        .withReadTimeoutMillis(1000));

                messages.forEach((message) => {
                    // Deserialize the status message first.
                    const statusMessage = util.deserializeJsonBytesToObj(message.payload, StatusMessage);
                    // Check the status of the status message. If the status is 'Success', the file was successfully uploaded to S3.
                    // If the status was either 'Failure' or 'Cancelled', the server was unable to upload the file to S3.
                    // We will print the message for why the upload to S3 failed from the status message.
                    // If the status was "InProgress", the status indicates that the server has started uploading the S3 task.
                    if (statusMessage.status === Status.Success) {
                        console.log(`Successfully uploaded file at path ${FILE_URL} to S3.`);
                        isS3UploadComplete = true;
                    } else if (statusMessage.status === Status.Failure || statusMessage.status === Status.Canceled) {
                        console.log(`Unable to upload file at path ${FILE_URL} to S3. Message: ${statusMessage.message}`);
                        isS3UploadComplete = true;
                    }
                });
                // Sleep for sometime for the S3 upload task to complete before trying to read the status message.
                await new Promise((r) => setTimeout(r, 5000));
            } catch (e) {
                // Ignored
            }
    } catch (e) {
        // Properly handle errors.
    }
});
client.onError((err) => {
    // Properly handle connection errors.
    // This is called only when the connection to the StreamManager server fails.
});

Node.js Référence du SDK : ReadMessages | StatusMessage