O AWS IoT Greengrass Version 1 entrou na fase de vida útil prolongada em 30 de junho de 2023. Para obter mais informações, consulte política de manutenção do AWS IoT Greengrass V1. Após essa data, o AWS IoT Greengrass V1 não lançará atualizações que forneçam recursos, aprimoramentos, correções de erros ou patches de segurança. Os dispositivos que funcionam com o AWS IoT Greengrass V1 não serão interrompidos e continuarão operando e se conectando à nuvem. É altamente recomendável que você migre para AWS IoT Greengrass Version 2, o que adicionará novos recursos significativos e suporte para plataformas adicionais.

Configurações de exportação para destinos compatíveis do Nuvem AWS

As funções do Lambda definidas pelo usuário utilizam StreamManagerClient no SDK do AWS IoT Greengrass Core para interagir com o gerenciador de fluxo. Quando uma função do Lambda cria um fluxo ou atualiza um fluxo, ela passa um objeto MessageStreamDefinition que representa as propriedades do fluxo, incluindo a definição de exportação. O objeto ExportDefinition contém as configurações de exportação definidas para o fluxo. O gerenciador de fluxo usa essas configurações de exportação para determinar onde e como exportar o fluxo.

Você pode definir zero ou mais configurações de exportação em um fluxo, incluindo várias configurações de exportação para um único tipo de destino. Por exemplo, você pode exportar um fluxo para dois canais do AWS IoT Analytics e um fluxo de dados do Kinesis.

Para tentativas de exportação malsucedidas, o gerenciador de fluxo tenta continuamente exportar dados para a Nuvem AWS em intervalos de até cinco minutos. Não há um limite máximo para o número de novas tentativas.

Você é responsável pela manutenção desses recursos da Nuvem AWS.

Canais do AWS IoT Analytics

O gerenciador de fluxo fornece suporte a exportações automáticas para o AWS IoT Analytics. O AWS IoT Analytics permite realizar análises avançadas em seus dados para ajudar a tomar decisões de negócios e aprimorar os modelos de machine learning. Para obter mais informações, consulte O que é o AWS IoT Analytics? no AWS IoT Analytics Guia do usuário do .

No SDK do AWS IoT Greengrass Core, suas funções do Lambda usam IoTAnalyticsConfig para definir a configuração de exportação para esse tipo de destino. Para mais informações, consulte a referência do SDK para seu idioma de destino:

Requisitos

Esse destino de exportação tem os seguintes requisitos:

Exportando para o AWS IoT Analytics

Para criar um fluxo que exporte para o AWS IoT Analytics, suas funções do Lambda criam um fluxo com uma definição de exportação que inclui um ou mais objetos IoTAnalyticsConfig. Esse objeto define as configurações de exportação, como canal de destino, tamanho do lote, intervalo do lote e prioridade.

Quando suas funções do Lambda recebem dados de dispositivos, elas acrescentam mensagens que contêm um blob de dados ao fluxo de destino.

Em seguida, o gerenciador de fluxo exporta os dados com base nas configurações de lote e na prioridade definidas nas configurações de exportação do fluxo.

 

Amazon Kinesis Data Streams

O gerenciador de fluxos é compatível com exportações automáticas para o Amazon Kinesis Data Streams. O Kinesis Data Streams é comumente usado para agregar dados de alto volume e carregá-los em um data warehouse ou cluster de redução de mapas. Para obter mais informações, consulte O que é o Amazon Kinesis Data Streams? no Guia do desenvolvedor do Amazon Kinesis.

No SDK do AWS IoT Greengrass Core, suas funções do Lambda usam KinesisConfig para definir a configuração de exportação para esse tipo de destino. Para mais informações, consulte a referência do SDK para seu idioma de destino:

Requisitos

Esse destino de exportação tem os seguintes requisitos:

Exportação do Kinesis Data Streams

Para criar um fluxo que exporte para o Kinesis Data Streams, suas funções do Lambda criam um fluxo com uma definição de exportação que inclui um ou mais objetos KinesisConfig. Esse objeto define as configurações de exportação, como fluxo de dados, tamanho do lote, intervalo do lote e prioridade.

Quando suas funções do Lambda recebem dados de dispositivos, elas acrescentam mensagens que contêm um blob de dados ao fluxo de destino. Em seguida, o gerenciador de fluxo exporta os dados com base nas configurações de lote e na prioridade definidas nas configurações de exportação do fluxo.

O gerenciador de fluxo gera uma UUID exclusiva e aleatória como chave de partição para cada registro carregado no Amazon Kinesis.

 

Propriedades do ativo AWS IoT SiteWise

O gerenciador de fluxo fornece suporte a exportações automáticas para o AWS IoT SiteWise. O AWS IoT SiteWise permite coletar, organizar e analisar dados de equipamentos industriais em escala. Para mais informações, consulte O que é o AWS IoT SiteWise? no AWS IoT SiteWise Guia do usuário.

No SDK do AWS IoT Greengrass Core, suas funções do Lambda usam IoTSiteWiseConfig para definir a configuração de exportação para esse tipo de destino. Para mais informações, consulte a referência do SDK para seu idioma de destino:

Requisitos

Esse destino de exportação tem os seguintes requisitos:

Exportando para o AWS IoT SiteWise

Para criar um fluxo que exporte para o AWS IoT SiteWise, suas funções do Lambda criam um fluxo com uma definição de exportação que inclui um ou mais objetos IoTSiteWiseConfig. Esse objeto define as configurações de exportação, como tamanho do lote, intervalo do lote e prioridade.

Quando suas funções do Lambda recebem dados de propriedades de ativos de dispositivos, elas anexam mensagens que contêm os dados ao fluxo de destino. As mensagens são objetos PutAssetPropertyValueEntry serializados em JSON que contêm valores de propriedade para uma ou mais propriedades de ativos. Para obter mais informações, consulte Anexar mensagem para destinos de exportação do AWS IoT SiteWise.

Em seguida, o gerenciador de fluxo exporta os dados com base nas configurações de lote e na prioridade definidas nas configurações de exportação do fluxo.

 

Você pode ajustar as configurações do gerenciador de fluxo e a lógica da função do Lambda para criar sua estratégia de exportação. Por exemplo:

 

Objetos do Amazon S3

O gerenciador de fluxo é compatível com exportações automáticas para o Amazon S3. Você pode utilizar o Amazon S3 para armazenar e recuperar grandes volumes de dados. Para obter mais informações, consulte O que é o Amazon S3? no Guia do desenvolvedor do Amazon Simple Storage Service.

No SDK do AWS IoT Greengrass Core, suas funções do Lambda usam S3ExportTaskExecutorConfig para definir a configuração de exportação para esse tipo de destino. Para mais informações, consulte a referência do SDK para seu idioma de destino:

Requisitos

Esse destino de exportação tem os seguintes requisitos:

Exportar para o Amazon S3

Para criar um fluxo que exporte para o Amazon S3, suas funções do Lambda usam o objeto S3ExportTaskExecutorConfig para configurar a política de exportação. A política define as configurações de exportação, como o limite e a prioridade de upload em várias partes. Para exportações do Amazon S3, o gerenciador de fluxo carrega dados que ele lê de arquivos locais no dispositivo principal. Para iniciar um upload, suas funções do Lambda anexam uma tarefa de exportação ao fluxo de destino. A tarefa de exportação contém informações sobre o arquivo de entrada e o objeto de destino do Amazon S3. O gerenciador de fluxo executa tarefas na sequência em que elas são anexadas ao fluxo.

Esse fluxo de alto nível é mostrado no diagrama a seguir.

O gerenciador de fluxo usa a propriedade de limite de upload de várias partes, a configuração do tamanho mínimo das partes e o tamanho do arquivo de entrada para determinar como fazer upload dos dados. O limite de upload de várias partes deve ser maior que o tamanho mínimo das partes. Se você quiser fazer upload de dados em paralelo, pode criar vários fluxos.

As chaves que especificam seus objetos de destino do Amazon S3 podem incluir strings Java DateTimeFormatter válidas em espaços reservados de !{timestamp:value}. Você pode usar esses espaços reservados de data e hora para particionar dados no Amazon S3 com base na hora em que os dados do arquivo de entrada foram carregados. Por exemplo, o nome da chave a seguir é resolvido para um valor como my-key/2020/12/31/data.txt.

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

Gerenciar dados de entrada

Você pode criar códigos que os aplicativos de IoT usam para gerenciar o ciclo de vida dos dados de entrada. O exemplo de fluxo de trabalho a seguir mostra como você pode usar as funções do Lambda para gerenciar esses dados.

Monitorar tarefas de exportação

Você pode criar códigos que os aplicativos de IoT usam para monitorar o status das suas exportações do Amazon S3. Suas funções do Lambda devem criar um fluxo de status e, em seguida, configurar o fluxo de exportação para gravar atualizações de status no fluxo de status. Um único fluxo de status pode receber atualizações de status de vários fluxos que são exportados para o Amazon S3.

Primeiro, crie um fluxo para usar como fluxo de status. Você pode configurar as políticas de tamanho e retenção do fluxo para controlar a vida útil das mensagens de status. Por exemplo:

Em seguida, crie ou atualize o fluxo de exportação para usar o fluxo de status. Especificamente, defina a propriedade de configuração de status da configuração de exportação S3ExportTaskExecutorConfig do fluxo. Isso faz com que o gerenciador de fluxo grave mensagens de status sobre as tarefas de exportação para o fluxo de status. No objeto StatusConfig, especifique o nome do fluxo de status e o nível de detalhe. Os valores suportados a seguir variam do menos detalhado (ERROR) ao mais detalhado (TRACE). O padrão é INFO.

 

O exemplo de fluxo de trabalho a seguir mostra como as funções do Lambda podem usar um fluxo de status para monitorar o status de exportação.

O exemplo a seguir mostra como uma função do Lambda pode ler e processar mensagens de status.

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.

Referência do SDK para 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.
}

Referência do SDK para 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.
});

Referência do SDK para Node.js: readMessages | StatusMessage