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

Os componentes do Greengrass definidos pelo usuário usam o StreamManagerClient no Stream Manager SDK para interagir com o Gerenciador de fluxos. Quando um componente cria um fluxo ou atualiza um fluxo, ele 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 Stream Manager SDK, os componentes do Greengrass 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, os componentes do Greengrass 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 os componentes do Greengrass recebem dados de dispositivos, eles anexam 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 é muito usado para agregar dados de alto volume e carregá-los em um data warehouse ou cluster do MapReduce. Para obter mais informações, consulte O que é o Amazon Kinesis Data Streams? no Guia do desenvolvedor do Amazon Kinesis.

No Stream Manager SDK, os componentes do Greengrass 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, os componentes do Greengrass 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 os componentes do Greengrass recebem dados de dispositivos, eles anexam 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 Stream Manager SDK, os componentes do Greengrass 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, os componentes do Greengrass 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 os componentes do Greengrass recebem dados de propriedades de ativos dos dispositivos, eles anexam mensagens que contêm um blob de 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 fluxos e a lógica do componente do Greengrass 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 Stream Manager SDK, os componentes do Greengrass 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, os componentes do Greengrass 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, os componentes do Greengrass 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 fluxos executa as tarefas na sequência em que elas são anexadas ao fluxo.

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 os componentes do Greengrass 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. Os componentes do Greengrass devem criar um fluxo de status e configurar o fluxo de exportação para gravar as 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. Essa configuração instrui o Gerenciador de fluxos a gravar as mensagens de status sobre as tarefas de exportação no 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 os componentes do Greengrass podem usar um fluxo de status para monitorar o status da exportação.

O exemplo a seguir mostra como um componente do Greengrass pode ler e processar as mensagens de status.

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.

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

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

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