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.

Exporter les configurations pour les AWS Cloud destinations prises en charge

Les composants Greengrass définis par l'utilisateur sont StreamManagerClient utilisés dans le SDK Stream Manager pour interagir avec le Stream Manager. Lorsqu'un composant crée un flux ou met à jour un flux, il transmet un MessageStreamDefinition objet qui représente les propriétés du flux, y compris la définition de l'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 AWS IoT Analytics canaux et un flux de données Kinesis.

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

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

Canaux AWS IoT Analytics

Le gestionnaire de flux prend en charge les exportations automatiques versAWS IoT Analytics. AWS IoT Analyticsvous permet d'effectuer une analyse avancée de vos données afin de prendre des décisions commerciales et d'améliorer les modèles d'apprentissage automatique. Pour plus d'informations, voir Qu'est-ce que c'est AWS IoT Analytics ? dans le guide de AWS IoT Analytics l'utilisateur.

Dans le SDK Stream Manager, vos composants Greengrass utilisent IoTAnalyticsConfig le pour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence du SDK pour votre langue cible :

Prérequis

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

Exportation vers AWS IoT Analytics

Pour créer un flux exporté versAWS IoT Analytics, vos composants Greengrass créent un flux avec une définition d'exportation qui inclut un ou plusieurs IoTAnalyticsConfig objets. Cet objet définit les paramètres d'exportation, tels que le canal cible, la taille du lot, l'intervalle entre les lots et la priorité.

Lorsque vos composants Greengrass reçoivent des données provenant d'appareils, ils ajoutent des messages contenant une quantité importante de données au 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.

Flux de données Amazon Kinesis

Le gestionnaire de flux 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 MapReduce de données ou un cluster. Pour plus d'informations, consultez Qu'est-ce qu'Amazon Kinesis Data Streams ? dans le manuel Amazon Kinesis Developer Guide.

Dans le SDK Stream Manager, vos composants Greengrass utilisent KinesisConfig le pour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence du SDK pour votre langue cible :

Prérequis

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 composants Greengrass créent un flux avec une définition d'exportation qui inclut un ou plusieurs objets. KinesisConfig 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 composants Greengrass reçoivent des données provenant d'appareils, ils ajoutent des messages contenant une quantité importante de données au 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.

Le gestionnaire de flux génère un UUID unique et aléatoire comme clé de partition pour chaque enregistrement chargé sur Amazon Kinesis.

AWS IoT SiteWisepropriétés des actifs

Le gestionnaire de flux prend en charge les exportations automatiques versAWS IoT SiteWise. AWS IoT SiteWisevous permet de collecter, d'organiser et d'analyser les données des é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 Stream Manager, vos composants Greengrass utilisent IoTSiteWiseConfig le pour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence du SDK pour votre langue cible :

Prérequis

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

Exportation vers AWS IoT SiteWise

Pour créer un flux exporté versAWS IoT SiteWise, vos composants Greengrass 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 composants Greengrass reçoivent des données sur les propriétés des actifs depuis des appareils, ils ajoutent des messages contenant ces données au flux cible. Les messages sont des PutAssetPropertyValueEntry objets sérialisés en JSON 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.

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 composants Greengrass 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 plus d'informations, consultez Qu'est-ce qu'Amazon S3 ? dans le guide du développeur d'Amazon Simple Storage Service.

Dans le SDK Stream Manager, vos composants Greengrass utilisent S3ExportTaskExecutorConfig le pour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence du SDK pour votre langue cible :

Prérequis

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

Exportation vers Amazon S3

Pour créer un flux exporté vers Amazon S3, vos composants Greengrass 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é du 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 composants Greengrass 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.

Le gestionnaire de flux utilise la propriété de seuil de téléchargement en plusieurs parties, le paramètre de taille de pièce minimale et la taille du fichier d'entrée pour déterminer le mode de téléchargement des données. Le seuil de téléchargement partitionné doit être supérieur ou égal à la taille de pièce minimale. 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 les !{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é téléchargées. Par exemple, le nom de clé suivant correspond à 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 vous pouvez utiliser les composants Greengrass 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 le statut de vos exportations Amazon S3. Vos composants Greengrass doivent créer un flux d'état, puis configurer le flux d'exportation pour écrire des mises à jour de statut dans le flux d'état. Un seul flux de statut peut recevoir des mises à jour de statut provenant de plusieurs flux exportés vers Amazon S3.

Créez d'abord 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 :

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 de la configuration S3ExportTaskExecutorConfig d'exportation du flux. Ce paramètre indique au gestionnaire de flux d'écrire des messages d'état concernant les tâches d'exportation dans 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 moins détaillée (ERROR) à la plus détaillée (). TRACE L’argument par défaut est INFO.

L'exemple de flux de travail suivant montre comment les composants Greengrass peuvent utiliser un flux d'état pour surveiller le statut des exportations.

L'exemple suivant montre comment un composant Greengrass peut lire et traiter les messages d'état.

Python

import time
from stream_manager import (
    ReadMessagesOptions,
    Status,
    StatusConfig,
    StatusLevel,
    StatusMessage,
    StreamManagerClient,
)
from 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.StreamManagerClientFactory;
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 = StreamManagerClientFactory.standard().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-stream-manager-sdk'*);

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 SDK Node.js : ReadMessages | StatusMessage