AWS IoT Greengrass Version 1 entró en la fase de vida útil prolongada el 30 de junio de 2023. Para obtener más información, consulte la política de mantenimiento de AWS IoT Greengrass V1 Después de esta fecha, AWS IoT Greengrass V1 no se publicarán actualizaciones que proporcionen funciones, mejoras, correcciones de errores o parches de seguridad. Los dispositivos que se ejecuten AWS IoT Greengrass V1 no se verán afectados y seguirán funcionando y conectándose a la nube. Le recomendamos encarecidamente que migre a AWS IoT Greengrass Version 2, ya que añade importantes funciones nuevas y es compatible con plataformas adicionales.

Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Exportación de configuraciones para Nube de AWS destinos compatibles

Las funciones Lambda StreamManagerClientdefinidas de Lambda definidas por el usuario deben utilizar el Core SDK AWS IoT Greengrass para interactuar con el administrador de secuencias. Cuando una función de Lambda crea un flujo o actualiza un flujo, pasa un objeto MessageStreamDefinition que representa las propiedades del flujo, incluida la definición de exportación. El objeto ExportDefinition contiene las configuraciones de exportación definidas para el flujo. El administrador de flujos utiliza estas configuraciones de exportación para determinar dónde y cómo exportar el flujo.

Puede definir cero o más configuraciones de exportación en un flujo, incluidas varias configuraciones de exportación para un único tipo de destino. Por ejemplo, puede exportar un flujo a dos canales AWS IoT Analytics y a un flujo de datos de Kinesis.

En caso de intentos fallidos de exportación, el administrador del flujo vuelve a intentar exportar los datos continuamente al Nube de AWS a intervalos de hasta cinco minutos. La cantidad de reintentos no tiene un límite máximo.

Usted es responsable del mantenimiento de estos recursos Nube de AWS.

Canales de AWS IoT Analytics

El administrador de flujos admite la exportación automática a AWS IoT Analytics. AWS IoT Analytics le permite realizar análisis avanzados de sus datos para ayudarle a tomar decisiones de negocios y mejorar los modelos de machine learning. Para obtener más información, consulte ¿Qué es AWS IoT Analytics? en la AWS IoT Analytics Guía del usuario.

En el AWS IoT Greengrass Core SDK, las funciones de Lambda utilizan el IoTAnalyticsConfig para definir la configuración de exportación para este tipo de destino. Para obtener más información, consulte la referencia del SDK para el lenguaje de destino.

Requisitos

Este destino de exportación tiene los siguientes requisitos:

Exportación a AWS IoT Analytics

Para crear un flujo que se exporte a AWS IoT Analytics, las funciones de Lambda crean un flujo con una definición de exportación que incluye uno o más objetos IoTAnalyticsConfig. Este objeto define los ajustes de exportación, como el canal de destino, el tamaño del lote, el intervalo del lote y la prioridad.

Cuando sus funciones de Lambda reciben datos de los dispositivos, anexan mensajes que contienen una masa de datos al flujo de destino.

A continuación, el administrador de flujos exporta los datos en función de los ajustes del lote y la prioridad definidos en las configuraciones de exportación del flujo.

 

Amazon Kinesis Data Streams

El administrador de flujos permite exportar automáticamente a Amazon Kinesis Data Streams. Kinesis Data Streams se suele utilizar para agregar grandes volúmenes de datos y cargarlos en un almacenamiento de datos o en un clúster de map-reduce. Para obtener más información, consulte Qué son los Amazon Kinesis Data Streams en la Guía para desarrolladores de Amazon Kinesis.

En el AWS IoT Greengrass Core SDK, las funciones de Lambda utilizan el KinesisConfig para definir la configuración de exportación para este tipo de destino. Para obtener más información, consulte la referencia del SDK para el lenguaje de destino.

Requisitos

Este destino de exportación tiene los siguientes requisitos:

Exportar a Kinesis Data Streams

Para crear un flujo que se exporte a Kinesis Data Streams, sus funciones de Lambda crean un flujo con una definición de exportación que incluye uno o más objetos KinesisConfig. Este objeto define los ajustes de exportación, como el flujo de datos de destino, el tamaño del lote, el intervalo del lote y la prioridad.

Cuando sus funciones de Lambda reciben datos de los dispositivos, anexan mensajes que contienen una masa de datos al flujo de destino. A continuación, el administrador de flujos exporta los datos en función de los ajustes del lote y la prioridad definidos en las configuraciones de exportación del flujo.

El administrador de flujos genera un UUID aleatorio único como clave de partición para cada registro cargado en Amazon Kinesis.

 

AWS IoT SiteWise propiedades de activos

El administrador de flujos admite la exportación automática a AWS IoT SiteWise. AWS IoT SiteWise le permite recopilar, organizar y analizar datos de equipos industriales a escala. Para obtener más información, consulte ¿Qué es AWS IoT SiteWise? en la AWS IoT SiteWiseGuía del usuario.

En el AWS IoT Greengrass Core SDK, las funciones de Lambda utilizan el IoTSiteWiseConfig para definir la configuración de exportación para este tipo de destino. Para obtener más información, consulte la referencia del SDK para el lenguaje de destino.

Requisitos

Este destino de exportación tiene los siguientes requisitos:

Exportación a AWS IoT SiteWise

Para crear un flujo que se exporte a AWS IoT SiteWise, las funciones de Lambda crean un flujo con una definición de exportación que incluye uno o más objetos IoTSiteWiseConfig. Este objeto define los ajustes de exportación, como el tamaño del lote, el intervalo del lote y la prioridad.

Cuando sus funciones de Lambda reciben datos de propiedad de activos, anexan mensajes que contienen los datos al flujo de destino. Los mensajes son objetos PutAssetPropertyValueEntry serializados en JSON que contienen los valores de propiedad de una o más propiedades de los activos. Para obtener más información, consulte Añadir un mensaje para los AWS IoT SiteWisedestinos de exportación.

A continuación, el administrador de flujos exporta los datos en función de los ajustes del lote y la prioridad definidos en las configuraciones de exportación del flujo.

 

Puede ajustar los ajustes del administrador de flujos y la lógica de la función de Lambda para diseñar su estrategia de exportación. Por ejemplo:

 

Objetos de Amazon S3

El administrador de flujos permite exportar automáticamente a Amazon S3. Puede utilizar Amazon S3 para almacenar y recuperar grandes cantidades de datos. Para obtener más información, consulte ¿Qué es Amazon S3? en la Guía para desarrolladores de Amazon Simple Storage Service.

En el AWS IoT Greengrass Core SDK, las funciones de Lambda utilizan el S3ExportTaskExecutorConfig para definir la configuración de exportación para este tipo de destino. Para obtener más información, consulte la referencia del SDK para el lenguaje de destino.

Requisitos

Este destino de exportación tiene los siguientes requisitos:

Exportar a Amazon S3.

Para crear un flujo que se exporte a Amazon S3, las funciones de Lambda utilizan el objeto S3ExportTaskExecutorConfig para configurar la política de exportación. La política define los ajustes de exportación, como el umbral de carga multiparte y la prioridad. Para las exportaciones de Amazon S3, el administrador de flujos carga los datos que lee de los archivos locales en el dispositivo principal. Para iniciar una carga, sus funciones de Lambda anexan una tarea de exportación al flujo de destino. La tarea de exportación contiene información sobre el archivo de entrada y el objeto Amazon S3 de destino. El administrador de flujos ejecuta las tareas en la secuencia en que se anexan al flujo.

Este flujo de alto nivel se muestra en el siguiente diagrama.

El administrador de flujos utiliza la propiedad de umbral de carga multiparte, el ajuste del tamaño mínimo de las piezas y el tamaño del archivo de entrada para determinar cómo cargar los datos. El umbral de carga multiparte debe ser igual o mayor que el tamaño mínimo de la pieza. Si desea cargar datos en paralelo, puede crear varios flujos.

Las claves que especifican los objetos de Amazon S3 de destino pueden incluir cadenas Java DateTimeFormatter válidas en los marcadores de posición !{timestamp:value}. Puede utilizar estos marcadores de fecha y hora para particionar los datos en Amazon S3 en función de la hora en la que se cargaron los datos del archivo de entrada. Por ejemplo, el siguiente nombre de clave se resuelve en un valor como my-key/2020/12/31/data.txt.

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

Administrar datos de entrada

Puede crear un código que las aplicaciones de IoT usen para administrar el ciclo de vida de los datos de entrada. El siguiente ejemplo de flujo de trabajo muestra cómo se pueden utilizar las funciones de Lambda para administrar estos datos.

Supervise las tareas de exportación

Puede crear un código que las aplicaciones de IoT utilizan para monitorear el estado de sus exportaciones de Amazon S3. Las funciones de Lambda deben crear un flujo de estado y, a continuación, configurar el flujo de exportación para escribir actualizaciones de estado en el flujo de estado. Una sola transmisión de estado puede recibir actualizaciones de estado de varias transmisiones que se exportan a Amazon S3.

En primer lugar, cree un flujo para utilizarlo como flujo de estado. Puede configurar las políticas de tamaño y retención del flujo para controlar la vida útil de los mensajes de estado. Por ejemplo:

A continuación, cree o actualice el flujo de exportación para usar el flujo de estado. En concreto, defina la propiedad de configuración de estado de la configuración de S3ExportTaskExecutorConfig exportación del flujo. Esto le indica al administrador del flujo que escriba mensajes de estado sobre las tareas de exportación en el flujo de estado. En el objeto StatusConfig, especifique el nombre del flujo de estado y el nivel de detalle. Los siguientes valores admitidos van desde el menos detallado (ERROR) al más detallado (). TRACE El valor predeterminado es INFO.

 

El siguiente ejemplo de flujo de trabajo muestra cómo las funciones de Lambda pueden utilizar un flujo de estado para supervisar el estado de la exportación.

El siguiente ejemplo muestra cómo una función de Lambda puede leer y procesar los mensajes de estado.

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.

Referencia del SDK de 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.
}

Referencia del SDK de 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.
});

Referencia del SDK de Node.js: readMessages | StatusMessage