AWS IoT Greengrass Version 1 è entrato nella fase di estensione della vita utile il 30 giugno 2023. Per ulteriori informazioni, consulta la politica AWS IoT Greengrass V1 di manutenzione. Dopo questa data, AWS IoT Greengrass V1 non rilascerà aggiornamenti che forniscano funzionalità, miglioramenti, correzioni di bug o patch di sicurezza. I dispositivi che funzionano AWS IoT Greengrass V1 non subiranno interruzioni e continueranno a funzionare e a connettersi al cloud. Ti consigliamo vivamente di eseguire la migrazione a AWS IoT Greengrass Version 2, che aggiunge nuove importanti funzionalità e supporto per piattaforme aggiuntive.

Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

Configurazioni di esportazione per supportateCloud AWSdestinazioni

Utilizzo di funzioni Lambda definite dall'utenteStreamManagerClientnellaAWS IoT GreengrassUn SDK di base per interagire con stream manager. Quando una funzione Lambdacrea un flussooaggiorna un flusso, passa aMessageStreamDefinitionoggetto che rappresenta le proprietà del flusso, inclusa la definizione di esportazione. LaExportDefinitionobject contiene le configurazioni di esportazione definite per lo stream. Stream Manager utilizza queste configurazioni di esportazione per determinare dove e come esportare lo stream.

È possibile definire zero o più configurazioni di esportazione in un flusso, incluse più configurazioni di esportazione per un singolo tipo di destinazione. Ad esempio, è possibile esportare un flusso in dueAWS IoT AnalyticsUn flusso di dati Kinesis.

In caso di tentativi di esportazione non riusciti, stream manager tenta continuamente di esportare i dati nelCloud AWSa intervalli fino a cinque minuti. Il numero di tentativi di ripetizione non presenta un limite massimo.

Sei responsabile per il mantenimento di questiCloud AWSrisorse AWS.

Canali AWS IoT Analytics

Stream Manager supporta le esportazioni automatiche inAWS IoT Analytics.AWS IoT Analyticsconsente di eseguire analisi avanzate sui dati per aiutare a prendere decisioni aziendali e migliorare i modelli di machine learning. Per ulteriori informazioni, consultaChe cos'èAWS IoT Analytics?nellaAWS IoT AnalyticsGuida per l’utente di.

NellaAWS IoT GreengrassCore SDK, le tue funzioni Lambda utilizzanoIoTAnalyticsConfigper definire la configurazione di esportazione per questo tipo di destinazione. Per ulteriori informazioni, consulta la documentazione sugli SDK relativa alla tua lingua di destinazione:

Requisiti

Questa destinazione di esportazione presenta i seguenti requisiti:

Esportazione di inAWS IoT Analytics

Per creare un flusso che esporta inAWS IoT Analytics, le funzioni LambdaPer creare un flussocon una definizione di esportazione che include una o piùIoTAnalyticsConfigoggetti. Questo oggetto definisce le impostazioni di esportazione, come il canale di destinazione, la dimensione del batch, l'intervallo batch e la priorità.

Quando le funzioni Lambda ricevono i dati dai dispositivi, essiAggiunta di messaggiche contengono un blob di dati per il flusso di destinazione.

Quindi, stream manager esporta i dati in base alle impostazioni batch e alla priorità definite nelle configurazioni di esportazione del flusso.

 

Amazon Kinesis

Stream manager supporta le esportazioni automatiche in Amazon Kinesis Data Streams. Kinesis Data Streams viene comunemente utilizzato per aggregare elevati volumi di dati e caricarli in un data warehouse o cluster di riduzione della mappa. Per ulteriori informazioni, consultaCos'è Amazon Kinesis Data Streams?nellaGuida per gli sviluppatori di Amazon Kinesis.

NellaAWS IoT GreengrassCore SDK, le tue funzioni Lambda utilizzanoKinesisConfigper definire la configurazione di esportazione per questo tipo di destinazione. Per ulteriori informazioni, consulta la documentazione sugli SDK relativa alla tua lingua di destinazione:

Requisiti

Questa destinazione di esportazione presenta i seguenti requisiti:

Esportare in Kinesis Data Streams

Per creare uno stream che esporta in Kinesis Data Streams, le funzioni LambdaPer creare un flussocon una definizione di esportazione che include una o piùKinesisConfigoggetti. Questo oggetto definisce le impostazioni di esportazione, come il flusso di dati di destinazione, la dimensione del batch, l'intervallo batch e la priorità.

Quando le funzioni Lambda ricevono i dati dai dispositivi, essiAggiunta di messaggiche contengono un blob di dati per il flusso di destinazione. Quindi, stream manager esporta i dati in base alle impostazioni batch e alla priorità definite nelle configurazioni di esportazione del flusso.

Stream Manager genera un UUID casuale univoco come chiave di partizione per ogni record caricato su Amazon Kinesis.

 

AWS IoT SiteWiseproprietà degli asset

Stream Manager supporta le esportazioni automatiche inAWS IoT SiteWise.AWS IoT SiteWiseconsente di raccogliere, organizzare e analizzare i dati provenienti da apparecchiature industriali su larga scala. Per ulteriori informazioni, consultaChe cos'èAWS IoT SiteWise?nellaAWS IoT SiteWiseGuida per l’utente di.

NellaAWS IoT GreengrassCore SDK, le tue funzioni Lambda utilizzanoIoTSiteWiseConfigper definire la configurazione di esportazione per questo tipo di destinazione. Per ulteriori informazioni, consulta la documentazione sugli SDK relativa alla tua lingua di destinazione:

Requisiti

Questa destinazione di esportazione presenta i seguenti requisiti:

Esportazione di inAWS IoT SiteWise

Per creare un flusso che esporta inAWS IoT SiteWise, le funzioni LambdaPer creare un flussocon una definizione di esportazione che include una o piùIoTSiteWiseConfigoggetti. Questo oggetto definisce le impostazioni di esportazione, come la dimensione del batch, l'intervallo batch e la priorità.

Quando le funzioni Lambda ricevono i dati delle proprietà delle risorse dai dispositivi, aggiungono messaggi che contengono i dati al flusso di destinazione. I messaggi sono serializzati JSONPutAssetPropertyValueEntryoggetti che contengono valori di proprietà per una o più proprietà degli asset. Per ulteriori informazioni, consultaAggiunta di un messaggioperAWS IoT SiteWiseEsportazione di destinazioni.

Quindi, stream manager esporta i dati in base alle impostazioni batch e alla priorità definite nelle configurazioni di esportazione del flusso.

 

È possibile regolare le impostazioni dello stream manager e la logica della funzione Lambda per progettare la strategia di esportazione. Ad esempio:

 

Oggetti Amazon S3

Stream manager supporta le esportazioni automatiche in Amazon S3. Puoi utilizzare Amazon S3 per archiviare e recuperare grandi quantità di dati. Per ulteriori informazioni, consultaChe cos'è Amazon S3?nellaGuida per sviluppatori di Amazon Simple Storage.

NellaAWS IoT GreengrassCore SDK, le tue funzioni Lambda utilizzanoS3ExportTaskExecutorConfigper definire la configurazione di esportazione per questo tipo di destinazione. Per ulteriori informazioni, consulta la documentazione sugli SDK relativa alla tua lingua di destinazione:

Requisiti

Questa destinazione di esportazione presenta i seguenti requisiti:

Esportazione di in Amazon S3

Per creare uno stream che esporta in Amazon S3, le funzioni Lambda utilizzanoS3ExportTaskExecutorConfigoggetto per configurare il criterio di esportazione. Il criterio definisce le impostazioni di esportazione, come la soglia di caricamento multiparte e la priorità. Per le esportazioni di Amazon S3, stream Manager carica i dati che legge dai file locali sul dispositivo principale. Per avviare un caricamento, le funzioni Lambda aggiungono un'attività di esportazione al flusso di destinazione. L'attività di esportazione contiene informazioni sul file di input e sull'oggetto Amazon S3 di destinazione. Stream Manager esegue le attività nella sequenza in cui vengono aggiunte allo stream.

Questo flusso di lavoro di alto livello è mostrato nel seguente diagramma.

Stream Manager utilizza la proprietà soglia di caricamento multiparte,dimensione minima della parteimpostazione e dimensione del file di input per determinare come caricare i dati. La soglia di caricamento in più parti deve essere maggiore o uguale alla dimensione minima della parte. Se si desidera caricare i dati in parallel, è possibile creare più flussi.

Le chiavi che specificano gli oggetti Amazon S3 di destinazione possono includere valideFormatter Java DateTimeFormatterstringhe in!{timestamp:value}segnaposto. Puoi utilizzare questi segnaposto con timestamp per partizionare i dati in Amazon S3 in base al momento in cui sono stati caricati i dati del file di input. Ad esempio, il seguente nome della chiave viene risolto in un valore comemy-key/2020/12/31/data.txt.

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

Gestione dei dati di input

È possibile creare codice utilizzato dalle applicazioni IoT per gestire il ciclo di vita dei dati di input. Il flusso di lavoro di esempio seguente mostra come è possibile utilizzare le funzioni Lambda per gestire questi dati.

Monitoraggio delle attività di

Puoi creare codice utilizzato dalle applicazioni IoT per monitorare lo stato delle esportazioni Amazon S3. Le funzioni Lambda devono creare un flusso di stato e quindi configurare il flusso di esportazione per scrivere gli aggiornamenti di stato nel flusso di stato. Un singolo flusso di stato può ricevere aggiornamenti di stato da più flussi esportati in Amazon S3.

Per prima cosa,Per creare un flussoda utilizzare come flusso di stato. È possibile configurare i criteri di dimensione e conservazione per lo stream per controllare la durata dei messaggi di stato. Ad esempio:

Quindi, crea o aggiorna il flusso di esportazione per utilizzare il flusso di stato. In particolare, impostare la proprietà di configurazione dello stato del flussoS3ExportTaskExecutorConfigEsportazione della configurazione. Ciò indica allo stream manager di scrivere messaggi di stato sulle attività di esportazione nel flusso di stato. NellaStatusConfigobject, specificare il nome del flusso di stato e il livello di verbosità. I seguenti valori supportati vanno da quelli meno dettagliati (ERROR) al più prolisso (TRACE). Il valore predefinito è INFO.

 

Il flusso di lavoro di esempio seguente mostra come le funzioni Lambda potrebbero utilizzare un flusso di stato per monitorare lo stato dell'esportazione.

L'esempio seguente mostra come una funzione Lambda potrebbe leggere ed elaborare i messaggi di stato.

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.

Riferimento 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.
}

Riferimento SDK: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.
});

Riferimento SDK Node.js:readMessages|StatusMessage