AWS IoT Greengrass Version 1 est entré dans la phase de durée de vie prolongée le 30 juin 2023. Pour plus d'informations, consultez la politique de AWS IoT Greengrass V1 maintenance. Après cette date, AWS IoT Greengrass V1 ne publiera pas de mises à jour fournissant des fonctionnalités, des améliorations, des corrections de bogues ou des correctifs de sécurité. Les appareils qui fonctionnent AWS IoT Greengrass V1 sous tension ne seront pas perturbés et continueront à fonctionner et à se connecter au cloud. Nous vous recommandons vivement de migrer vers AWS IoT Greengrass Version 2, qui ajoute de nouvelles fonctionnalités importantes et prend en charge des plateformes supplémentaires.

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 prises en chargeAWS Clouddestinations

Utilisation des fonctions Lambda définies par l'utilisateurStreamManagerClientdans leAWS IoT GreengrassKit SDK principal pour interagir avec le gestionnaire de flux. Quand une fonction Lambdacrée un fluxoumet à jour un flux, il passe unMessageStreamDefinitionobjet qui représente les propriétés du flux, y compris la définition d'exportation. LeExportDefinitioncontient les configurations d'exportation définies pour le flux. Stream Manager utilise ces configurations d'exportation pour déterminer où et comment exporter le flux.

Vous pouvez définir zéro configuration d'exportation ou plus sur un flux, y compris plusieurs configurations d'exportation pour un seul type de destination. Par exemple, vous pouvez exporter un flux vers deuxAWS IoT Analyticset un flux de données Kinesis.

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

Vous êtes responsable de la maintenance de cesAWS CloudAWS.

Canaux AWS IoT Analytics

Stream Manager prend en charge les exportations automatiques versAWS IoT Analytics.AWS IoT Analyticsvous permet d'effectuer des analyses avancées de vos données pour aider à prendre des décisions commerciales et à améliorer les modèles d'apprentissage automatique. Pour de plus amples informations, veuillez consulterPrésentation d'AWS IoT Analytics?dans leAWS IoT AnalyticsGuide de l'utilisateur.

DansAWS IoT GreengrassSDK Core, vos fonctions Lambda utilisent leIoTAnalyticsConfigpour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence SDK de votre langue cible :

Prérequis

Cette destination d'exportation possède les critères suivants :

Exportation vers AWS IoT Analytics

Pour créer un flux qui exporte versAWS IoT Analytics, vos fonctions LambdaCréer un fluxavec une définition d'exportation comprenant une ou plusieursIoTAnalyticsConfigobjets. Cet objet définit les paramètres d'exportation, tels que le canal cible, la taille du lot, l'intervalle de lot et la priorité.

Lorsque vos fonctions Lambda reçoivent des données provenant d'appareils, ellesAjouter des messagesqui contiennent un blob de données vers le flux cible.

Le gestionnaire de flux 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.

 

Amazon Kinesis Data Streams

Le gestionnaire de flux prend en charge les exportations automatiques vers Amazon Kinesis Data Streams. Kinesis Data Streams est couramment utilisé pour agréger des données à volume élevé et les charger dans un entrepôt de données ou un cluster de réduction de mappe. Pour de plus amples informations, veuillez consulterQu'est-ce qu'Amazon Kinesis Data Streams ?dans leManuel du développeur Amazon Kinesis.

DansAWS IoT GreengrassSDK Core, vos fonctions Lambda utilisent leKinesisConfigpour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence SDK de votre langue cible :

Prérequis

Cette destination d'exportation possède les critères suivants :

Exporter vers Kinesis Data Streams

Pour créer un flux qui exporte vers Kinesis Data Streams, votre Lambda fonctionneCréer un fluxavec une définition d'exportation comprenant une ou plusieursKinesisConfigobjets. Cet objet définit les paramètres d'exportation, tels que le flux de données cible, la taille du lot, l'intervalle de lot et la priorité.

Lorsque vos fonctions Lambda reçoivent des données provenant d'appareils, ellesAjouter des messagesqui contiennent un blob de données vers le flux cible. Le gestionnaire de flux 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 téléchargé sur Amazon Kinesis.

 

AWS IoT SiteWisepropriétés des ressources

Stream Manager prend en charge les exportations automatiques versAWS IoT SiteWise.AWS IoT SiteWisevous permet de collecter, d'organiser et d'analyser à grande échelle les données provenant d'équipements industriels. Pour de plus amples informations, veuillez consulterPrésentation d'AWS IoT SiteWise?dans leAWS IoT SiteWiseGuide de l'utilisateur.

DansAWS IoT GreengrassSDK Core, vos fonctions Lambda utilisent leIoTSiteWiseConfigpour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence SDK de votre langue cible :

Prérequis

Cette destination d'exportation possède les critères suivants :

Exportation vers AWS IoT SiteWise

Pour créer un flux qui exporte versAWS IoT SiteWise, vos fonctions LambdaCréer un fluxavec une définition d'exportation comprenant une ou plusieursIoTSiteWiseConfigobjets. Cet objet définit les paramètres d'exportation, tels que la taille du lot, l'intervalle de lot et la priorité.

Lorsque vos fonctions Lambda reçoivent des données de propriété de ressources provenant d'appareils, elles ajoutent des messages contenant les données au flux cible. Les messages sont sérialisés en JSONPutAssetPropertyValueEntryobjets contenant des valeurs de propriétés pour une ou plusieurs propriétés d'actifs. Pour de plus amples informations, veuillez consulterAjouter un messagepourAWS IoT SiteWisedestinations d'exportation.

Le gestionnaire de flux 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 des fonctions Lambda pour concevoir votre stratégie d'exportation. Par exemple :

 

Objets Amazon S3

Le gestionnaire de flux 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 de plus amples informations, veuillez consulterQu'est-ce qu'Amazon S3 ?dans leManuel du développeur Amazon Simple Storage Service.

DansAWS IoT GreengrassSDK Core, vos fonctions Lambda utilisent leS3ExportTaskExecutorConfigpour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence SDK de votre langue cible :

Prérequis

Cette destination d'exportation possède les critères suivants :

Exporter vers Amazon S3

Pour créer un flux qui exporte vers Amazon S3, vos fonctions Lambda utilisent leS3ExportTaskExecutorConfigpour configurer la stratégie d'exportation. La stratégie définit les paramètres d'exportation, tels que le seuil de chargement partitionné et la priorité. Pour les exportations Amazon S3, Stream Manager télécharge les données qu'il lit à partir de fichiers locaux sur le périphérique principal. Pour lancer un 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 la séquence dans laquelle elles sont ajoutées au flux.

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

Le gestionnaire de flux utilise la propriété seuil de chargement partitionné,Taille de partie minimumet la taille du fichier d'entrée pour déterminer comment charger les données. Le seuil de téléchargement partitionné doit être supérieur ou égal à la taille d'article minimale. Si vous souhaitez télécharger des données en parallel, vous pouvez créer plusieurs flux.

Les clés qui spécifient vos objets Amazon S3 cibles peuvent inclure des clés validesDateTimeFormatter datetimeJavachaînes dans!{timestamp:value}espaces réservés. Vous pouvez utiliser ces espaces réservés à l'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é télé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 utilisé par les applications IoT 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.

Surveillance des 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 d'état dans le flux d'état. Un flux d'état unique peut recevoir des mises à jour de statut provenant de plusieurs flux exportés vers Amazon S3.

Premiers,Créer un fluxà utiliser comme flux d'état. Vous pouvez configurer les stratégies de taille et 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 d'état du fluxS3ExportTaskExecutorConfigconfiguration d'exportation. Cela indique au gestionnaire de flux d'écrire des messages d'état concernant les tâches d'exportation vers le flux d'état. DansStatusConfig, spécifiez le nom du flux d'état et le niveau de verbosité. Les valeurs prises en charge suivantes vont des valeurs les moins détaillées (ERROR) au plus verbeux (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 de l'exportation.

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 kit 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 kit 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.
});

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