As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá. # Gerenciar fluxos de dados no nos dispositivos principais do Greengrass AWS IoT Greengrass O gerenciador de fluxo torna mais eficiente e confiável a transferência de dados de IoT de alto volume para o. Nuvem AWS O gerenciador de fluxo processa fluxos de dados no AWS IoT Greengrass Core antes de exportá-los para o. Nuvem AWS O Stream Manager se integra a cenários de ponta comuns, como inferência de aprendizado de máquina (ML), em que o dispositivo AWS IoT Greengrass principal processa e analisa os dados antes de exportá-los para os destinos de armazenamento locais Nuvem AWS ou para os destinos de armazenamento. O Stream Manager fornece uma interface comum para simplificar o desenvolvimento de componentes personalizados, de forma que você não precise criar uma funcionalidade personalizada de gerenciamento de fluxos. Seus componentes podem usar um mecanismo padronizado para processar fluxos de alto volume e gerenciar políticas locais de retenção de dados. É possível definir políticas para o tipo de armazenamento, tamanho e retenção de dados para cada fluxo para controlar como o gerenciador de fluxos processa e exporta dados. O gerenciador de fluxos funciona em ambientes de conectividade intermitente ou limitada. Você pode definir o uso da largura de banda, o comportamento do tempo limite e como o AWS IoT Greengrass Core manipula os dados de streaming quando estão conectados ou desconectados. Você também pode definir prioridades para controlar a ordem em que o AWS IoT Greengrass Core exporta fluxos para a Nuvem AWS. Isso permite que você manipular dados críticos mais cedo do que outros dados. Você pode configurar o gerenciador de fluxo para exportar dados automaticamente Nuvem AWS para armazenamento ou processamento e análise adicionais. O gerenciador de fluxos oferece suporte para exportações para os seguintes destinos da Nuvem AWS : + Canais em AWS IoT Analytics. AWS IoT Analytics permite realizar análises avançadas em seus dados para ajudar a tomar decisões comerciais e melhorar os modelos de aprendizado de máquina. Para obter mais informações, consulte [O que é o AWS IoT Analytics?](https://docs.aws.amazon.com/iotanalytics/latest/userguide/welcome.html) no * Guia do usuário do AWS IoT Analytics *. + Fluxos no Amazon Kinesis Data Streams Você pode usar o Kinesis Data Streams para agregar dados de alto volume e carregá-los em um data warehouse ou cluster. MapReduce Para obter mais informações, consulte [O que é o Amazon Kinesis Data Streams?](https://docs.aws.amazon.com/streams/latest/dev/what-is-this-service.html) no *Guia do desenvolvedor do Amazon Kinesis Data Streams*. + Propriedades de ativos em AWS IoT SiteWise. AWS IoT SiteWise permite coletar, organizar e analisar dados de equipamentos industriais em grande escala. Para obter mais informações, consulte [O que é AWS IoT SiteWise?](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/what-is-sitewise.html) no *Guia do AWS IoT SiteWise usuário*. + Objetos no Amazon Simple Storage Service (Amazon S3) É possível utilizar o Amazon S3 para armazenar e recuperar grandes volumes de dados. Para obter mais informações, consulte [O que é o Amazon S3?](https://docs.aws.amazon.com/AmazonS3/latest/dev/Welcome.html) no *Guia do desenvolvedor do Amazon Simple Storage Service*. ## Fluxo de trabalho do gerenciamento de streams Suas aplicações para IoT interagem com o gerenciador de fluxos pelo Stream Manager SDK. Em um fluxo de trabalho simples, um componente no AWS IoT Greengrass núcleo consome dados de IoT, como métricas de temperatura e pressão de séries temporais. O componente pode filtrar ou compactar os dados e, em seguida, chamar o Stream Manager SDK para gravar os dados em um fluxo no gerenciador de fluxos. O gerenciador de fluxo pode exportar o fluxo para o Nuvem AWS automaticamente com base nas políticas que você define para o fluxo. Os componentes também podem enviar dados diretamente para bancos de dados locais ou repositórios de armazenamento. Suas aplicações de IoT podem incluir vários componentes personalizados que leem ou gravam em fluxos. Esses componentes podem ler e gravar em fluxos para filtrar, agregar e analisar dados no dispositivo AWS IoT Greengrass principal. Isso possibilita responder rapidamente a eventos locais e extrair informações valiosas antes que os dados sejam transferidos do núcleo para os Nuvem AWS destinos locais. Para começar, implante o componente stream manager em seu dispositivo AWS IoT Greengrass principal. Na implantação, configure os parâmetros do componente gerenciador de fluxos para definir as configurações que se aplicam a todos os fluxos no dispositivo Greengrass nucleus. Use esses parâmetros para controlar como o gerenciador de fluxos armazena, processa e exporta fluxos com base em suas necessidades comerciais e restrições de ambiente. Depois de configurar o gerenciador de fluxos, você pode criar e implantar suas aplicações de IoT. Normalmente, esses são componentes personalizados que usam `StreamManagerClient` no Stream Manager SDK para criar e interagir com fluxos. Ao criar um fluxo, você pode definir políticas por fluxo, como destinos de exportação, prioridade e persistência. ## Requisitos Os seguintes requisitos são aplicados para usar o gerenciador de fluxo: + O gerenciador de streaming requer um mínimo de 70 MB de RAM, além do software AWS IoT Greengrass Core. O requisito de memória total depende da sua workload. + AWS IoT Greengrass os componentes devem usar o Stream Manager SDK para interagir com o stream manager. O Stream Manager SDK está disponível nos seguintes idiomas: + [Stream Manager SDK para Java](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-java/) (v1.1.0 ou posterior) + [Stream Manager SDK para Node.js](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-js/) (v1.1.0 ou posterior) + [Stream Manager SDK para Python](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-python/) (v1.1.0 ou posterior) + AWS IoT Greengrass os componentes devem especificar o componente do gerenciador de fluxo (`aws.greengrass.StreamManager`) como uma dependência em sua receita para usar o gerenciador de fluxo. **nota** Se você usa o gerenciador de fluxos para exportar dados para a nuvem, não pode atualizar a versão 2.0.7 desse componente para uma versão entre 2.0.8 e 2.0.11. Se você estiver implantando o gerenciador de fluxos pela primeira vez, é altamente recomendável implantar a versão mais recente do componente gerenciador de fluxos. + Se você definir destinos de Nuvem AWS exportação para um stream, deverá criar seus destinos de exportação e conceder permissões de acesso na função de [dispositivo do Greengrass](device-service-role.md). Dependendo do destino, outros requisitos também podem ser aplicados. Para obter mais informações, consulte: + [AWS IoT Analytics canais](stream-export-configurations.md#export-to-iot-analytics) + [Amazon Kinesis Data Streams](stream-export-configurations.md#export-to-kinesis) + [AWS IoT SiteWise propriedades do ativo](stream-export-configurations.md#export-to-iot-sitewise) + [Objetos do Amazon S3](stream-export-configurations.md#export-to-s3) Você é responsável pela manutenção desses Nuvem AWS recursos. ## Segurança de dados Ao usar o gerenciador de fluxo, esteja ciente das seguintes considerações de segurança. ### Segurança de dados locais AWS IoT Greengrass não criptografa dados de fluxo em repouso ou em trânsito entre componentes locais no dispositivo principal. + **Dados em repouso**. Os dados de fluxo são armazenados localmente em um diretório de armazenamento. Para segurança de dados, AWS IoT Greengrass depende de permissões de arquivo e criptografia de disco inteiro, se ativada. É possível usar o parâmetro [STREAM\$1MANAGER\$1STORE\$1ROOT\$1DIR](configure-stream-manager.md#STREAM_MANAGER_STORE_ROOT_DIR) opcional para especificar o diretório de armazenamento. Se você alterar esse parâmetro posteriormente para usar um diretório de armazenamento diferente, AWS IoT Greengrass não excluirá o diretório de armazenamento anterior nem seu conteúdo. + **Dados em trânsito localmente**. AWS IoT Greengrass não criptografa dados de fluxo em trânsito local entre fontes de dados, AWS IoT Greengrass componentes, o SDK do Stream Manager e o gerenciador de fluxo. + **Dados em trânsito para Nuvem AWS** o. Os fluxos de dados exportados pelo gerenciador de fluxo para o Nuvem AWS usam criptografia de cliente de AWS serviço padrão com Transport Layer Security (TLS). ### Autenticação de cliente Os clientes do gerenciador de fluxos usam o Stream Manager SDK para se comunicar com o gerenciador de fluxos. Quando a autenticação do cliente está habilitada, somente os componentes do Greengrass podem interagir com fluxos no gerenciador de fluxos. Quando a autenticação do cliente está desabilitada, qualquer processo em execução no dispositivo principal do Greengrass pode interagir com fluxos no gerenciador de fluxos. Você só deve desabilitar a autenticação se o seu caso de negócios exigir. Use o parâmetro [STREAM\$1MANAGER\$1AUTHENTICATE\$1CLIENT](configure-stream-manager.md#STREAM_MANAGER_AUTHENTICATE_CLIENT) para definir o modo de autenticação do cliente. É possível configurar esse parâmetro ao implantar o componente do gerenciador de fluxos nos dispositivos principais. **** |   | Habilitado | Desabilitado | | --- | --- | --- | | Valor do parâmetro | `true` (padrão e recomendado) | `false` | | Clientes permitidos | Componentes do Greengrass no dispositivo principal | Componentes do Greengrass no dispositivo principal Outros processos em execução no dispositivo de Greengrass nucleus | ## Consulte também + [Configurar o gerenciador de fluxo do AWS IoT Greengrass](configure-stream-manager.md) + [Use StreamManagerClient para trabalhar com streams](work-with-streams.md) + [Configurações de exportação para destinos compatíveis Nuvem AWS](stream-export-configurations.md) # Configurar o gerenciador de fluxo do AWS IoT Greengrass Nos dispositivos principais do Greengrass, o gerenciador de fluxos pode armazenar, processar e exportar dados de dispositivos da IoT. O gerenciador de fluxos fornece parâmetros que você usa para definir configurações de runtime. Essas configurações se aplicam a todos os fluxos no núcleo do Greengrass. Você pode usar o console do AWS IoT Greengrass ou a API para configurar as definições do gerenciador de fluxos ao implantar o componente. As alterações entrarão em vigor após a implantação ser concluída. ## Parâmetros do gerenciador de fluxo O gerenciador de fluxos fornece os seguintes parâmetros que você pode configurar ao implantar o componente em seus dispositivos principais. Todos os parâmetros são opcionais. **Diretório de armazenamento** Nome do parâmetro: `STREAM_MANAGER_STORE_ROOT_DIR` O caminho absoluto da pasta local usada para armazenar fluxos. Esse valor deve começar com uma barra (por exemplo, `/data`). Você deve especificar uma pasta existente, e o [usuário do sistema que executa o componente do gerenciador de fluxos](configure-greengrass-core-v2.md#configure-component-user) deve ter permissões para ler e gravar nessa pasta. Por exemplo, você pode executar os comandos a seguir para criar e configurar uma pasta, `/var/greengrass/streams`, que você especifica como a pasta raiz do gerenciador de fluxos. Esses comandos permitem que o usuário padrão do sistema, `ggc_user`, leia e grave nessa pasta. ``` sudo mkdir /var/greengrass/streams sudo chown ggc_user /var/greengrass/streams sudo chmod 700 /var/greengrass/streams ``` Para obter informações sobre como proteger dados de fluxo, consulte [Segurança de dados locais](manage-data-streams.md#stream-manager-security-stream-data). Padrão: `/greengrass/v2/work/aws.greengrass.StreamManager` **Porta do servidor** Nome do parâmetro: `STREAM_MANAGER_SERVER_PORT` O número da porta local usado para se comunicar com o gerenciador de fluxo. O padrão é “”. `8088`. Você pode especificar `0` para usar uma porta disponível aleatória. **Autenticar cliente** Nome do parâmetro: `STREAM_MANAGER_AUTHENTICATE_CLIENT` Indica se os clientes devem ser autenticados de modo a interagir com o gerenciador de fluxo. Toda a interação entre clientes e o gerenciador de fluxos é controlada pelo Stream Manager SDK. Esse parâmetro determina quais clientes podem chamar o Stream Manager SDK para trabalhar com fluxos. Para obter mais informações, consulte [Autenticação de cliente](manage-data-streams.md#stream-manager-security-client-authentication). Os valores válidos são `true` ou `false`. O padrão é `true` (recomendado). + `true`. Permite somente componentes do Greengrass como clientes. Os componentes usam protocolos internos do núcleo do AWS IoT Greengrass para autenticação com o Stream Manager SDK. + `false`. Permite que qualquer processo executado no AWS IoT Greengrass Core seja um cliente. Não defina o valor como `false` a menos que seu caso de negócios exija. Por exemplo, use `false` somente se os processos que não são do dispositivo principal precisarem se comunicar diretamente com o gerenciador de fluxos. **Largura máxima de banda** Nome do parâmetro: `STREAM_MANAGER_EXPORTER_MAX_BANDWIDTH` A média da largura máxima de banda (em kilobits por segundo) que pode ser usada para exportar dados. O padrão permite o uso ilimitado da largura de banda disponível. **Tamanho do grupo de threads** Nome do parâmetro: `STREAM_MANAGER_EXPORTER_THREAD_POOL_SIZE` O número máximo de threads ativos que podem ser usados para exportar dados. O padrão é “”. `5`. O tamanho ideal depende do hardware, do volume do fluxo e do número planejado de fluxos de exportação. Se a velocidade de exportação for lenta, você poderá ajustar essa configuração para encontrar o tamanho ideal para seu hardware e caso de negócios. A CPU e a memória do hardware do dispositivo de núcleo são fatores limitantes. Para iniciar, você pode tentar definir esse valor igual ao número de núcleos do processador no dispositivo. Tenha cuidado para não definir um tamanho superior ao que o seu hardware pode suportar. Cada fluxo consome recursos de hardware, então tente limitar o número de fluxos de exportação em dispositivos restritos. **Argumentos JVM** Nome do parâmetro: `JVM_ARGS` Argumentos personalizados da Java Virtual Machine para passar para o gerenciador de fluxo na startup. Se houver vários argumentos, separe-os por espaços. Só use esse parâmetro quando precisar substituir as configurações padrão usadas pela JVM. Por exemplo, talvez seja necessário aumentar o tamanho do heap padrão caso você planeje exportar um grande número de fluxos. **Nível de registro** Nome do parâmetro: `LOG_LEVEL` O nível de registro em log para o componente. Escolha entre os seguintes níveis de log, ordenados por nível: + `TRACE` + `DEBUG` + `INFO` + `WARN` + `ERROR` Padrão: `INFO` **Tamanho mínimo para upload de várias partes** Nome do parâmetro: `STREAM_MANAGER_EXPORTER_S3_DESTINATION_MULTIPART_UPLOAD_MIN_PART_SIZE_BYTES` O tamanho mínimo (em bytes) de uma parte em um upload multiparte para o Amazon S3. O gerenciador de fluxo usa essa configuração e o tamanho do arquivo de entrada para determinar como agrupar dados em lote em uma solicitação PUT de várias partes. O valor mínimo e padrão é de `5242880` bytes (5 MB). O gerenciador de fluxo usa a propriedade `sizeThresholdForMultipartUploadBytes` do fluxo para determinar se deve exportar para o Amazon S3 como um upload de uma ou várias partes. Os componentes do Greengrass definidos pelo usuário definem esse limite quando criam um fluxo que exporta para o Amazon S3. O limite padrão é 5 MB. ## Consulte também + [Gerenciar fluxos de dados no nos dispositivos principais do Greengrass](manage-data-streams.md) + [Use StreamManagerClient para trabalhar com streams](work-with-streams.md) + [Configurações de exportação para destinos compatíveis Nuvem AWS](stream-export-configurations.md) # Criar componentes personalizados que usam o Gerenciador de fluxos Use o Gerenciador de fluxos em componentes personalizados do Greengrass para armazenar, processar e exportar dados de dispositivos do IoT. Siga os procedimentos e os exemplos desta seção para criar fórmulas, artefatos e aplicações de componente que funcionem com o gerenciador de fluxos. Para obter informações sobre como desenvolver e testar componentes, consulte [Criar componentes do AWS IoT Greengrass](create-components.md). **Topics** + [Definir fórmulas de componente que usam o Gerenciador de fluxos](#stream-manager-recipes) + [Conectar-se ao gerenciador de fluxos no código da aplicação](#connect-to-stream-manager) ## Definir fórmulas de componente que usam o Gerenciador de fluxos Para usar o Gerenciador de fluxos em um componente personalizado, você deve definir o componente `aws.greengrass.StreamManager` como uma dependência. Você também deve fornecer o Stream Manager SDK. Conclua as tarefas a seguir para fazer download e usar o Stream Manager SDK no idioma de sua escolha. ### Usar o Stream Manager SDK para Java O Stream Manager SDK para Java está disponível como um arquivo JAR que você pode usar para compilar seu componente. Depois disso, você pode criar um JAR de aplicação que inclua o Stream Manager SDK, definir esse JAR como artefato de componente e executar o JAR no ciclo de vida do componente. **Para usar o Stream Manager SDK para Java** 1. Faça download do [arquivo JAR do Stream Manager SDK para Java](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-java/blob/main/sdk/aws-greengrass-stream-manager-sdk-java.jar). 1. Siga um destes procedimentos para criar os artefatos de componente da aplicação Java e o arquivo JAR do Stream Manager SDK: + Compile a aplicação como um arquivo JAR que inclua o JAR do Stream Manager SDK e execute esse arquivo JAR na fórmula do componente. + Defina o JAR do Stream Manager SDK como um artefato de componente. Adicione esse artefato ao classpath quando você executar a aplicação na fórmula do componente. A fórmula do componente deve ser semelhante ao exemplo a seguir. Esse componente executa uma versão modificada do exemplo [StreamManagerS3.java](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-java/blob/main/samples/StreamManagerS3/src/main/java/com/amazonaws/greengrass/examples/StreamManagerS3.java), que `StreamManagerS3.jar` inclui o Stream Manager SDK JAR. ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.StreamManagerS3Java", "ComponentVersion": "1.0.0", "ComponentDescription": "Uses stream manager to upload a file to an S3 bucket.", "ComponentPublisher": "Amazon", "ComponentDependencies": { "aws.greengrass.StreamManager": { "VersionRequirement": "^2.0.0" } }, "Manifests": [ { "Lifecycle": { "Run": "java -jar {artifacts:path}/StreamManagerS3.jar" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Java/1.0.0/StreamManagerS3.jar" } ] } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.StreamManagerS3Java ComponentVersion: 1.0.0 ComponentDescription: Uses stream manager to upload a file to an S3 bucket. ComponentPublisher: Amazon ComponentDependencies: aws.greengrass.StreamManager: VersionRequirement: "^2.0.0" Manifests: - Lifecycle: Run: java -jar {artifacts:path}/StreamManagerS3.jar Artifacts: - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Java/1.0.0/StreamManagerS3.jar ``` ------ Para obter informações sobre como desenvolver e testar componentes, consulte [Criar componentes do AWS IoT Greengrass](create-components.md). ### Usar o Stream Manager SDK para Python O Stream Manager SDK para Python está disponível como um código-fonte que você pode incluir em seu componente. Crie um arquivo ZIP do Stream Manager SDK, defina esse arquivo como um artefato do componente e instale os requisitos do SDK no ciclo de vida do componente. **Para usar o Stream Manager SDK para Python** 1. Clone ou baixe o repositório [aws-greengrass-stream-manager-sdk-python](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-python). ``` git clone git@github.com:aws-greengrass/aws-greengrass-stream-manager-sdk-python.git ``` 1. Crie um arquivo ZIP que inclua a pasta `stream_manager`, que contém o código-fonte do Stream Manager SDK para Python. Você pode fornecer esse arquivo ZIP como um artefato de componente que o software AWS IoT Greengrass Core descompacta ao instalar seu componente. Faça o seguinte: 1. Abra a pasta que contém o repositório que você clonou ou baixou na etapa anterior. ``` cd aws-greengrass-stream-manager-sdk-python ``` 1. Compacte a pasta `stream_manager` em um arquivo ZIP chamado `stream_manager_sdk.zip`. ------ #### [ Linux or Unix ] ``` zip -rv stream_manager_sdk.zip stream_manager ``` ------ #### [ Windows Command Prompt (CMD) ] ``` tar -acvf stream_manager_sdk.zip stream_manager ``` ------ #### [ PowerShell ] ``` Compress-Archive stream_manager stream_manager_sdk.zip ``` ------ 1. Verifique se o arquivo `stream_manager_sdk.zip` contém a pasta `stream_manager` e o respectivo conteúdo. Execute o comando a seguir para listar o conteúdo do arquivo ZIP. ------ #### [ Linux or Unix ] ``` unzip -l stream_manager_sdk.zip ``` ------ #### [ Windows Command Prompt (CMD) ] ``` tar -tf stream_manager_sdk.zip ``` ------ A saída deve ser semelhante à seguinte. ``` Archive: aws-greengrass-stream-manager-sdk-python/stream_manager.zip Length Date Time Name --------- ---------- ----- ---- 0 02-24-2021 20:45 stream_manager/ 913 02-24-2021 20:45 stream_manager/__init__.py 9719 02-24-2021 20:45 stream_manager/utilinternal.py 1412 02-24-2021 20:45 stream_manager/exceptions.py 1004 02-24-2021 20:45 stream_manager/util.py 0 02-24-2021 20:45 stream_manager/data/ 254463 02-24-2021 20:45 stream_manager/data/__init__.py 26515 02-24-2021 20:45 stream_manager/streammanagerclient.py --------- ------- 294026 8 files ``` 1. Copie os artefatos do Stream Manager SDK para a pasta de artefatos do componente. Além do arquivo ZIP do Stream Manager SDK, seu componente usa o arquivo `requirements.txt` do SDK para instalar as dependências do Stream Manager SDK. *\$1/greengrass-components*Substitua pelo caminho para a pasta que você usa para desenvolvimento local. ------ #### [ Linux or Unix ] ``` cp {stream_manager_sdk.zip,requirements.txt} ~/greengrass-components/artifacts/com.example.StreamManagerS3Python/1.0.0/ ``` ------ #### [ Windows Command Prompt (CMD) ] ``` robocopy . %USERPROFILE%\greengrass-components\artifacts\com.example.StreamManagerS3Python\1.0.0 stream_manager_sdk.zip robocopy . %USERPROFILE%\greengrass-components\artifacts\com.example.StreamManagerS3Python\1.0.0 requirements.txt ``` ------ #### [ PowerShell ] ``` cp .\stream_manager_sdk.zip,.\requirements.txt ~\greengrass-components\artifacts\com.example.StreamManagerS3Python\1.0.0\ ``` ------ 1. Crie a fórmula do seu componente. Na fórmula, faça o seguinte: 1. Defina `stream_manager_sdk.zip` e `requirements.txt` como artefatos. 1. Defina a aplicação Python como um artefato. 1. No ciclo de vida da instalação, instale os requisitos do Stream Manager SDK de `requirements.txt`. 1. No ciclo de vida da execução, anexe o Stream Manager SDK a `PYTHONPATH` e execute a aplicação Python. A fórmula do componente deve ser semelhante ao exemplo a seguir. Esse componente executa o [stream\$1manager\$1s3.py](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-python/blob/main/samples/stream_manager_s3.py) de exemplo. ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.StreamManagerS3Python", "ComponentVersion": "1.0.0", "ComponentDescription": "Uses stream manager to upload a file to an S3 bucket.", "ComponentPublisher": "Amazon", "ComponentDependencies": { "aws.greengrass.StreamManager": { "VersionRequirement": "^2.0.0" } }, "Manifests": [ { "Platform": { "os": "linux" }, "Lifecycle": { "install": "pip3 install --user -r {artifacts:path}/requirements.txt", "Run": "export PYTHONPATH=$PYTHONPATH:{artifacts:decompressedPath}/stream_manager_sdk; python3 {artifacts:path}/stream_manager_s3.py" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_sdk.zip", "Unarchive": "ZIP" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_s3.py" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/requirements.txt" } ] }, { "Platform": { "os": "windows" }, "Lifecycle": { "install": "pip3 install --user -r {artifacts:path}/requirements.txt", "Run": "set \"PYTHONPATH=%PYTHONPATH%;{artifacts:decompressedPath}/stream_manager_sdk\" & py -3 {artifacts:path}/stream_manager_s3.py" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_sdk.zip", "Unarchive": "ZIP" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_s3.py" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/requirements.txt" } ] } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.StreamManagerS3Python ComponentVersion: 1.0.0 ComponentDescription: Uses stream manager to upload a file to an S3 bucket. ComponentPublisher: Amazon ComponentDependencies: aws.greengrass.StreamManager: VersionRequirement: "^2.0.0" Manifests: - Platform: os: linux Lifecycle: install: pip3 install --user -r {artifacts:path}/requirements.txt Run: | export PYTHONPATH=$PYTHONPATH:{artifacts:decompressedPath}/stream_manager_sdk python3 {artifacts:path}/stream_manager_s3.py Artifacts: - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_sdk.zip Unarchive: ZIP - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_s3.py - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/requirements.txt - Platform: os: windows Lifecycle: install: pip3 install --user -r {artifacts:path}/requirements.txt Run: | set "PYTHONPATH=%PYTHONPATH%;{artifacts:decompressedPath}/stream_manager_sdk" py -3 {artifacts:path}/stream_manager_s3.py Artifacts: - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_sdk.zip Unarchive: ZIP - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_s3.py - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/requirements.txt ``` ------ Para obter informações sobre como desenvolver e testar componentes, consulte [Criar componentes do AWS IoT Greengrass](create-components.md). ### Use o Stream Manager SDK para JavaScript O Stream Manager SDK para JavaScript está disponível como código-fonte que você pode incluir em seu componente. Crie um arquivo ZIP do Stream Manager SDK, defina esse arquivo como um artefato do componente e instale o SDK no ciclo de vida do componente. **Para usar o SDK do Stream Manager para JavaScript** 1. Clone ou baixe o repositório [aws-greengrass-stream-manager-sdk-js](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-js). ``` git clone git@github.com:aws-greengrass/aws-greengrass-stream-manager-sdk-js.git ``` 1. Crie um arquivo ZIP que contenha a `aws-greengrass-stream-manager-sdk` pasta, que contém o código-fonte do SDK do Stream Manager para JavaScript. Você pode fornecer esse arquivo ZIP como um artefato de componente que o software AWS IoT Greengrass Core descompacta ao instalar seu componente. Faça o seguinte: 1. Abra a pasta que contém o repositório que você clonou ou baixou na etapa anterior. ``` cd aws-greengrass-stream-manager-sdk-js ``` 1. Compacte a pasta `aws-greengrass-stream-manager-sdk` em um arquivo ZIP chamado `stream-manager-sdk.zip`. ------ #### [ Linux or Unix ] ``` zip -rv stream-manager-sdk.zip aws-greengrass-stream-manager-sdk ``` ------ #### [ Windows Command Prompt (CMD) ] ``` tar -acvf stream-manager-sdk.zip aws-greengrass-stream-manager-sdk ``` ------ #### [ PowerShell ] ``` Compress-Archive aws-greengrass-stream-manager-sdk stream-manager-sdk.zip ``` ------ 1. Verifique se o arquivo `stream-manager-sdk.zip` contém a pasta `aws-greengrass-stream-manager-sdk` e o respectivo conteúdo. Execute o comando a seguir para listar o conteúdo do arquivo ZIP. ------ #### [ Linux or Unix ] ``` unzip -l stream-manager-sdk.zip ``` ------ #### [ Windows Command Prompt (CMD) ] ``` tar -tf stream-manager-sdk.zip ``` ------ A saída deve ser semelhante à seguinte. ``` Archive: stream-manager-sdk.zip Length Date Time Name --------- ---------- ----- ---- 0 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/ 369 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/package.json 1017 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/util.js 8374 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/utilInternal.js 1937 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/exceptions.js 0 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/data/ 353343 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/data/index.js 22599 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/client.js 216 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/index.js --------- ------- 387855 9 files ``` 1. Copie o artefato do Stream Manager SDK para a pasta de artefatos do componente. *\$1/greengrass-components*Substitua pelo caminho para a pasta que você usa para desenvolvimento local. ------ #### [ Linux or Unix ] ``` cp stream-manager-sdk.zip ~/greengrass-components/artifacts/com.example.StreamManagerS3JS/1.0.0/ ``` ------ #### [ Windows Command Prompt (CMD) ] ``` robocopy . %USERPROFILE%\greengrass-components\artifacts\com.example.StreamManagerS3JS\1.0.0 stream-manager-sdk.zip ``` ------ #### [ PowerShell ] ``` cp .\stream-manager-sdk.zip ~\greengrass-components\artifacts\com.example.StreamManagerS3JS\1.0.0\ ``` ------ 1. Crie a fórmula do seu componente. Na fórmula, faça o seguinte: 1. Defina `stream-manager-sdk.zip` como um artefato. 1. Defina seu JavaScript aplicativo como um artefato. 1. No ciclo de vida da instalação, instale o Stream Manager SDK do artefato `stream-manager-sdk.zip`. Esse comando `npm install` cria uma pasta `node_modules` que contém o Stream Manager SDK e as respectivas dependências. 1. No ciclo de vida de execução, anexe a `node_modules` pasta e execute seu `NODE_PATH` aplicativo. JavaScript A fórmula do componente deve ser semelhante ao exemplo a seguir. Esse componente executa o exemplo do [StreamManagerS3](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-js/blob/main/samples/StreamManagerS3/index.js). ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.StreamManagerS3JS", "ComponentVersion": "1.0.0", "ComponentDescription": "Uses stream manager to upload a file to an S3 bucket.", "ComponentPublisher": "Amazon", "ComponentDependencies": { "aws.greengrass.StreamManager": { "VersionRequirement": "^2.0.0" } }, "Manifests": [ { "Platform": { "os": "linux" }, "Lifecycle": { "install": "npm install {artifacts:decompressedPath}/stream-manager-sdk/aws-greengrass-stream-manager-sdk", "Run": "export NODE_PATH=$NODE_PATH:{work:path}/node_modules; node {artifacts:path}/index.js" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3JS/1.0.0/stream-manager-sdk.zip", "Unarchive": "ZIP" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3JS/1.0.0/index.js" } ] }, { "Platform": { "os": "windows" }, "Lifecycle": { "install": "npm install {artifacts:decompressedPath}/stream-manager-sdk/aws-greengrass-stream-manager-sdk", "Run": "set \"NODE_PATH=%NODE_PATH%;{work:path}/node_modules\" & node {artifacts:path}/index.js" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3JS/1.0.0/stream-manager-sdk.zip", "Unarchive": "ZIP" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3JS/1.0.0/index.js" } ] } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.StreamManagerS3JS ComponentVersion: 1.0.0 ComponentDescription: Uses stream manager to upload a file to an S3 bucket. ComponentPublisher: Amazon ComponentDependencies: aws.greengrass.StreamManager: VersionRequirement: "^2.0.0" Manifests: - Platform: os: linux Lifecycle: install: npm install {artifacts:decompressedPath}/stream-manager-sdk/aws-greengrass-stream-manager-sdk Run: | export NODE_PATH=$NODE_PATH:{work:path}/node_modules node {artifacts:path}/index.js Artifacts: - URI: s3://DOC-EXAMPLE-BUCKET/artifacts/com.example.StreamManagerS3JS/1.0.0/stream-manager-sdk.zip Unarchive: ZIP - URI: s3://DOC-EXAMPLE-BUCKET/artifacts/com.example.StreamManagerS3JS/1.0.0/index.js - Platform: os: windows Lifecycle: install: npm install {artifacts:decompressedPath}/stream-manager-sdk/aws-greengrass-stream-manager-sdk Run: | set "NODE_PATH=%NODE_PATH%;{work:path}/node_modules" node {artifacts:path}/index.js Artifacts: - URI: s3://DOC-EXAMPLE-BUCKET/artifacts/com.example.StreamManagerS3JS/1.0.0/stream-manager-sdk.zip Unarchive: ZIP - URI: s3://DOC-EXAMPLE-BUCKET/artifacts/com.example.StreamManagerS3JS/1.0.0/index.js ``` ------ Para obter informações sobre como desenvolver e testar componentes, consulte [Criar componentes do AWS IoT Greengrass](create-components.md). ## Conectar-se ao gerenciador de fluxos no código da aplicação Para se conectar ao gerenciador de fluxos em sua aplicação, crie uma instância de `StreamManagerClient` do Stream Manager SDK. Esse cliente se conecta ao componente Gerenciador de fluxos na porta padrão 8088 ou na porta que você especificar. Para obter informações sobre como usar o `StreamManagerClient` após a criação de uma instância, consulte [Use StreamManagerClient para trabalhar com streams](work-with-streams.md). **Example Exemplo: conectar-se ao Gerenciador de fluxos pela porta padrão** ``` import com.amazonaws.greengrass.streammanager.client.StreamManagerClient; public class MyStreamManagerComponent { void connectToStreamManagerWithDefaultPort() { StreamManagerClient client = StreamManagerClientFactory.standard().build(); // Use the client. } } ``` ``` from stream_manager import ( StreamManagerClient ) def connect_to_stream_manager_with_default_port(): client = StreamManagerClient() # Use the client. ``` ``` const { StreamManagerClient } = require('aws-greengrass-stream-manager-sdk'); function connectToStreamManagerWithDefaultPort() { const client = new StreamManagerClient(); // Use the client. } ``` **Example Exemplo: conectar-se ao Gerenciador de fluxos por uma porta diferente do padrão** Se você configurar o Gerenciador de fluxos com uma porta diferente do padrão, deverá usar a [comunicação entre processos](interprocess-communication.md) para recuperar a porta da configuração do componente. O parâmetro de configuração `port` contém o valor que você especifica em `STREAM_MANAGER_SERVER_PORT` ao implantar o Gerenciador de fluxos. ``` void connectToStreamManagerWithCustomPort() { EventStreamRPCConnection eventStreamRpcConnection = IPCUtils.getEventStreamRpcConnection(); GreengrassCoreIPCClient greengrassCoreIPCClient = new GreengrassCoreIPCClient(eventStreamRpcConnection); List keyPath = new ArrayList<>(); keyPath.add("port"); GetConfigurationRequest request = new GetConfigurationRequest(); request.setComponentName("aws.greengrass.StreamManager"); request.setKeyPath(keyPath); GetConfigurationResponse response = greengrassCoreIPCClient.getConfiguration(request, Optional.empty()).getResponse().get(); String port = response.getValue().get("port").toString(); System.out.print("Stream Manager is running on port: " + port); final StreamManagerClientConfig config = StreamManagerClientConfig.builder() .serverInfo(StreamManagerServerInfo.builder().port(Integer.parseInt(port)).build()).build(); StreamManagerClient client = StreamManagerClientFactory.standard().withClientConfig(config).build(); // Use the client. } ``` ``` import awsiot.greengrasscoreipc from awsiot.greengrasscoreipc.model import ( GetConfigurationRequest ) from stream_manager import ( StreamManagerClient ) TIMEOUT = 10 def connect_to_stream_manager_with_custom_port(): # Use IPC to get the port from the stream manager component configuration. ipc_client = awsiot.greengrasscoreipc.connect() request = GetConfigurationRequest() request.component_name = "aws.greengrass.StreamManager" request.key_path = ["port"] operation = ipc_client.new_get_configuration() operation.activate(request) future_response = operation.get_response() response = future_response.result(TIMEOUT) stream_manager_port = str(response.value["port"]) # Use port to create a stream manager client. stream_client = StreamManagerClient(port=stream_manager_port) # Use the client. ``` # Use StreamManagerClient para trabalhar com streams Os componentes do Greengrass definidos pelo usuário em execução no dispositivo principal do Greengrass podem usar o objeto `StreamManagerClient` no Stream Manager SDK para criar e interagir com os fluxos no [Gerenciador de fluxos](manage-data-streams.md). Quando um componente cria um fluxo, ele define os Nuvem AWS destinos, a priorização e outras políticas de exportação e retenção de dados para o fluxo. Para enviar os dados para o Gerenciador de fluxos, os componentes anexam os dados ao fluxo. Se um destino de exportação for definido para o fluxo, o gerenciador de fluxo exportará o fluxo automaticamente. **nota** Normalmente, os clientes do Gerenciador de fluxos são componentes do Greengrass definidos pelo usuário. Se o seu caso de negócios exigir, você também poderá permitir que os processos que não são de componente e são executados no núcleo do Greengrass (por exemplo, um contêiner do Docker) interajam com o Gerenciador de fluxos. Para obter mais informações, consulte [Autenticação de cliente](manage-data-streams.md#stream-manager-security-client-authentication). Os snippets neste tópico mostram como os clientes chamam o `StreamManagerClient` para trabalhar com fluxos. Para obter detalhes de implementação sobre os métodos e seus argumentos, use os links para a referência do SDK listada após cada snippet. Se você usa o Gerenciador de fluxos em uma função do Lambda, ela deve instanciar `StreamManagerClient` fora do manipulador de função. Se instanciado no manipulador, a função cria um `client` e uma conexão para o gerenciador de fluxo sempre que for invocado. **nota** Se você instanciar `StreamManagerClient` no manipulador, você deve chamar explicitamente o método `close()` quando o `client` concluir seu trabalho. Caso contrário, o `client` mantém a conexão aberta e outro thread em execução até que o script seja encerrado. `StreamManagerClient` comporta as operações a seguir: + [Criar stream de mensagens](#streammanagerclient-create-message-stream) + [Anexar mensagem](#streammanagerclient-append-message) + [Ler Mensagens](#streammanagerclient-read-messages) + [Listar fluxos](#streammanagerclient-list-streams) + [Descrever stream de mensagens](#streammanagerclient-describe-message-stream) + [Atualize o fluxo de mensagens](#streammanagerclient-update-message-stream) + [Excluir stream de mensagens](#streammanagerclient-delete-message-stream) ## Criar stream de mensagens Para criar um fluxo, um componente do Greengrass definido pelo usuário chama o método create e especifica um objeto `MessageStreamDefinition`. Esse objeto especifica o nome exclusivo do fluxo e define como o gerenciador de fluxo deve lidar com novos dados quando o tamanho máximo do fluxo for atingido. Você pode usar `MessageStreamDefinition` e os tipos de dados (como `ExportDefinition`, `StrategyOnFull` e `Persistence`) para definir outras propriedades de fluxo. Isso inclui: + O destino AWS IoT Analytics, Kinesis Data AWS IoT SiteWise Streams e Amazon S3, destinos para exportações automáticas. Para obter mais informações, consulte [Configurações de exportação para destinos compatíveis Nuvem AWS](stream-export-configurations.md). + Prioridade da exportação. O gerenciador de fluxo exporta fluxos de prioridade mais alta antes de fluxos de prioridade mais baixa. + Tamanho máximo do lote e intervalo de lote para AWS IoT Analytics Kinesis Data Streams e destinos AWS IoT SiteWise . O gerenciador de fluxo exporta mensagens quando qualquer condição é atendida. + Time-to-live (TTL). O tempo necessário para garantir que os dados de fluxo estejam disponíveis para processamento. Você deve certificar-se de que os dados podem ser consumidos nesse período de tempo. Esta não é uma política de exclusão. Os dados podem não ser excluídos imediatamente após o período de TTL. + Persistência do fluxo. Selecione salvar fluxos no sistema de arquivos para persistir os dados nas reinicializações do núcleo ou salve os fluxos na memória. + Número de sequência inicial. Especifique o número de sequência da mensagem a ser usada como mensagem inicial na exportação. Para obter mais informações sobre `MessageStreamDefinition`, consulte a referência do SDK para a sua linguagem de destino: + [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html)no Java SDK + [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)no SDK do Node.js + [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition)no SDK do Python **nota** O `StreamManagerClient` também fornece um destino alvo que você pode usar para exportar fluxos para um servidor HTTP. Este destino deve ser usado apenas para fins de teste. Ele não é estável e nem compatível para uso em ambientes de produção. Depois que um fluxo é criado, os componentes do Greengrass podem [anexar mensagens](#streammanagerclient-append-message) ao fluxo para enviar dados para exportação e [ler mensagens](#streammanagerclient-read-messages) do fluxo para processamento local. O número de fluxos criados depende dos seus recursos de hardware e caso de negócios. Uma estratégia é criar um stream para cada canal de destino no stream AWS IoT Analytics de dados do Kinesis, embora você possa definir vários destinos para um stream. Um fluxo tem longa duração. ### Requisitos Essa operação tem os seguintes requisitos: + Versão mínima do Stream Manager SDK: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ### Exemplos O snippet a seguir cria um fluxo chamado `StreamName`. Ele define as propriedades de fluxo em `MessageStreamDefinition` e nos tipos de dados subordinados. ------ #### [ Python ] ``` client = StreamManagerClient() try: client.create_message_stream(MessageStreamDefinition( name="StreamName", # Required. max_size=268435456, # Default is 256 MB. stream_segment_size=16777216, # Default is 16 MB. time_to_live_millis=None, # By default, no TTL is enabled. strategy_on_full=StrategyOnFull.OverwriteOldestData, # Required. persistence=Persistence.File, # Default is File. flush_on_write=False, # Default is false. export_definition=ExportDefinition( # Optional. Choose where/how the stream is exported to the Nuvem AWS. kinesis=None, iot_analytics=None, iot_sitewise=None, s3_task_executor=None ) )) except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` [Referência do SDK do Python: create\$1message\$1stream \$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.create_message_stream) ------ #### [ Java ] ``` try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { client.createMessageStream( new MessageStreamDefinition() .withName("StreamName") // Required. .withMaxSize(268435456L) // Default is 256 MB. .withStreamSegmentSize(16777216L) // Default is 16 MB. .withTimeToLiveMillis(null) // By default, no TTL is enabled. .withStrategyOnFull(StrategyOnFull.OverwriteOldestData) // Required. .withPersistence(Persistence.File) // Default is File. .withFlushOnWrite(false) // Default is false. .withExportDefinition( // Optional. Choose where/how the stream is exported to the Nuvem AWS. new ExportDefinition() .withKinesis(null) .withIotAnalytics(null) .withIotSitewise(null) .withS3(null) ) ); } catch (StreamManagerException e) { // Properly handle exception. } ``` Referência do SDK Java: \$1 [createMessageStream[MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#createMessageStream-com.amazonaws.greengrass.streammanager.model.MessageStreamDefinition-) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { await client.createMessageStream( new MessageStreamDefinition() .withName("StreamName") // Required. .withMaxSize(268435456) // Default is 256 MB. .withStreamSegmentSize(16777216) // Default is 16 MB. .withTimeToLiveMillis(null) // By default, no TTL is enabled. .withStrategyOnFull(StrategyOnFull.OverwriteOldestData) // Required. .withPersistence(Persistence.File) // Default is File. .withFlushOnWrite(false) // Default is false. .withExportDefinition( // Optional. Choose where/how the stream is exported to the Nuvem AWS. new ExportDefinition() .withKinesis(null) .withIotAnalytics(null) .withIotSiteWise(null) .withS3(null) ) ); } 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 do Node.js: \$1 [createMessageStream[MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#createMessageStream) ------ Para obter mais informações sobre como configurar destinos de exportação, consulte [Configurações de exportação para destinos compatíveis Nuvem AWS](stream-export-configurations.md). ## Anexar mensagem Para enviar dados ao Gerenciador de fluxos para exportação, os componentes do Greengrass anexam os dados ao fluxo de destino. O destino da exportação determina o tipo de dados a ser passado para esse método. ### Requisitos Essa operação tem os seguintes requisitos: + Versão mínima do Stream Manager SDK: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ### Exemplos #### AWS IoT Analytics ou destinos de exportação do Kinesis Data Streams O snippet a seguir anexa uma mensagem ao fluxo chamado `StreamName`. Para AWS IoT Analytics nossos destinos do Kinesis Data Streams, seus componentes do Greengrass acrescentam um blob de dados. Esse snippet tem os seguintes requisitos: + Versão mínima do Stream Manager SDK: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ------ #### [ Python ] ``` client = StreamManagerClient() try: sequence_number = client.append_message(stream_name="StreamName", data=b'Arbitrary bytes data') except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` Referência do SDK em Python: [append\$1message](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.append_message) ------ #### [ Java ] ``` try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { long sequenceNumber = client.appendMessage("StreamName", "Arbitrary byte array".getBytes()); } catch (StreamManagerException e) { // Properly handle exception. } ``` Referência do SDK em Java: [appendMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { const sequenceNumber = await client.appendMessage("StreamName", Buffer.from("Arbitrary byte array")); } 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 em Node.js: [appendMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage) ------ #### AWS IoT SiteWise destinos de exportação O snippet a seguir anexa uma mensagem ao fluxo chamado `StreamName`. Para AWS IoT SiteWise destinos, seus componentes do Greengrass acrescentam um objeto serializado. `PutAssetPropertyValueEntry` Para obter mais informações, consulte [Exportando para AWS IoT SiteWise](stream-export-configurations.md#export-streams-to-sitewise). **nota** Quando você envia dados para AWS IoT SiteWise, seus dados devem atender aos requisitos da `BatchPutAssetPropertyValue` ação. Para obter mais informações, consulte [BatchPutAssetPropertyValue](https://docs.aws.amazon.com/iot-sitewise/latest/APIReference/API_BatchPutAssetPropertyValue.html) na *Referência de APIs do AWS IoT SiteWise *. Esse snippet tem os seguintes requisitos: + Versão mínima do Stream Manager SDK: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ------ #### [ Python ] ``` client = StreamManagerClient() try: # SiteWise requires unique timestamps in all messages and also needs timestamps not earlier # than 10 minutes in the past. Add some randomness to time and offset. # Note: To create a new asset property data, you should use the classes defined in the # greengrasssdk.stream_manager module. time_in_nanos = TimeInNanos( time_in_seconds=calendar.timegm(time.gmtime()) - random.randint(0, 60), offset_in_nanos=random.randint(0, 10000) ) variant = Variant(double_value=random.random()) asset = [AssetPropertyValue(value=variant, quality=Quality.GOOD, timestamp=time_in_nanos)] putAssetPropertyValueEntry = PutAssetPropertyValueEntry(entry_id=str(uuid.uuid4()), property_alias="PropertyAlias", property_values=asset) sequence_number = client.append_message(stream_name="StreamName", Util.validate_and_serialize_to_json_bytes(putAssetPropertyValueEntry)) except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` [Referência do SDK do Python: append\$1message \$1 [PutAssetPropertyValueEntry](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.PutAssetPropertyValueEntry)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.append_message) ------ #### [ Java ] ``` try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) { Random rand = new Random(); // Note: To create a new asset property data, you should use the classes defined in the // com.amazonaws.greengrass.streammanager.model.sitewise package. List entries = new ArrayList<>() ; // IoTSiteWise requires unique timestamps in all messages and also needs timestamps not earlier // than 10 minutes in the past. Add some randomness to time and offset. final int maxTimeRandomness = 60; final int maxOffsetRandomness = 10000; double randomValue = rand.nextDouble(); TimeInNanos timestamp = new TimeInNanos() .withTimeInSeconds(Instant.now().getEpochSecond() - rand.nextInt(maxTimeRandomness)) .withOffsetInNanos((long) (rand.nextInt(maxOffsetRandomness))); AssetPropertyValue entry = new AssetPropertyValue() .withValue(new Variant().withDoubleValue(randomValue)) .withQuality(Quality.GOOD) .withTimestamp(timestamp); entries.add(entry); PutAssetPropertyValueEntry putAssetPropertyValueEntry = new PutAssetPropertyValueEntry() .withEntryId(UUID.randomUUID().toString()) .withPropertyAlias("PropertyAlias") .withPropertyValues(entries); long sequenceNumber = client.appendMessage("StreamName", ValidateAndSerialize.validateAndSerializeToJsonBytes(putAssetPropertyValueEntry)); } catch (StreamManagerException e) { // Properly handle exception. } ``` Referência do SDK Java: [appendMessage \$1 [PutAssetPropertyValueEntry](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/sitewise/PutAssetPropertyValueEntry.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { const maxTimeRandomness = 60; const maxOffsetRandomness = 10000; const randomValue = Math.random(); // Note: To create a new asset property data, you should use the classes defined in the // aws-greengrass-core-sdk StreamManager module. const timestamp = new TimeInNanos() .withTimeInSeconds(Math.round(Date.now() / 1000) - Math.floor(Math.random() - maxTimeRandomness)) .withOffsetInNanos(Math.floor(Math.random() * maxOffsetRandomness)); const entry = new AssetPropertyValue() .withValue(new Variant().withDoubleValue(randomValue)) .withQuality(Quality.GOOD) .withTimestamp(timestamp); const putAssetPropertyValueEntry = new PutAssetPropertyValueEntry() .withEntryId(`${ENTRY_ID_PREFIX}${i}`) .withPropertyAlias("PropertyAlias") .withPropertyValues([entry]); const sequenceNumber = await client.appendMessage("StreamName", util.validateAndSerializeToJsonBytes(putAssetPropertyValueEntry)); } 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 do Node.js: [appendMessage \$1 [PutAssetPropertyValueEntry](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.PutAssetPropertyValueEntry.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage) ------ #### Destinos de exportação do Amazon S3 O snippet a seguir anexa uma tarefa de exportação ao fluxo chamada `StreamName`. Para os destinos do Amazon S3, os componentes do Greengrass anexam um objeto serializado `S3ExportTaskDefinition` que contém informações sobre o arquivo de entrada de origem e o objeto do Amazon S3 de destino. Se o objeto especificado não existir, o gerenciador de fluxo criará o objeto para você. Para obter mais informações, consulte [Exportar para o Amazon S3](stream-export-configurations.md#export-streams-to-s3). Esse snippet tem os seguintes requisitos: + Versão mínima do Stream Manager SDK: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ------ #### [ Python ] ``` client = StreamManagerClient() try: # Append an Amazon S3 Task definition and print the sequence number. s3_export_task_definition = S3ExportTaskDefinition(input_url="URLToFile", bucket="BucketName", key="KeyName") sequence_number = client.append_message(stream_name="StreamName", Util.validate_and_serialize_to_json_bytes(s3_export_task_definition)) except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` [https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.append_message](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.append_message) ------ #### [ Java ] ``` try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) { // Append an Amazon S3 export task definition and print the sequence number. S3ExportTaskDefinition s3ExportTaskDefinition = new S3ExportTaskDefinition() .withBucket("BucketName") .withKey("KeyName") .withInputUrl("URLToFile"); long sequenceNumber = client.appendMessage("StreamName", ValidateAndSerialize.validateAndSerializeToJsonBytes(s3ExportTaskDefinition)); } catch (StreamManagerException e) { // Properly handle exception. } ``` [Referência do SDK Java: [appendMessage \$1](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-) S3 ExportTaskDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/S3ExportTaskDefinition.html) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { // Append an Amazon S3 export task definition and print the sequence number. const taskDefinition = new S3ExportTaskDefinition() .withBucket("BucketName") .withKey("KeyName") .withInputUrl("URLToFile"); const sequenceNumber = await client.appendMessage("StreamName", util.validateAndSerializeToJsonBytes(taskDefinition))); } 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 do Node.js: [appendMessage \$1](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage) S3 ExportTaskDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.S3ExportTaskDefinition.html) ------ ## Ler Mensagens Ler mensagens de um fluxo. ### Requisitos Essa operação tem os seguintes requisitos: + Versão mínima do Stream Manager SDK: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ### Exemplos O snippet a seguir lê mensagens do fluxo chamado `StreamName`. O método de leitura usa um objeto `ReadMessagesOptions` opcional que especifica o número de sequência a partir do qual começar a ler, os números mínimo e máximo a ler e um tempo limite para ler mensagens. ------ #### [ Python ] ``` client = StreamManagerClient() try: message_list = client.read_messages( stream_name="StreamName", # By default, if no options are specified, it tries to read one message from the beginning of the stream. options=ReadMessagesOptions( desired_start_sequence_number=100, # Try to read from sequence number 100 or greater. By default, this is 0. min_message_count=10, # Try to read 10 messages. If 10 messages are not available, then NotEnoughMessagesException is raised. By default, this is 1. max_message_count=100, # Accept up to 100 messages. By default this is 1. read_timeout_millis=5000 # Try to wait at most 5 seconds for the min_messsage_count to be fulfilled. By default, this is 0, which immediately returns the messages or an exception. ) ) except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` [Referência do SDK do Python: read\$1messages \$1 [ReadMessagesOptions](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.ReadMessagesOptions)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.read_messages) ------ #### [ Java ] ``` try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { List messages = client.readMessages("StreamName", // By default, if no options are specified, it tries to read one message from the beginning of the stream. new ReadMessagesOptions() // Try to read from sequence number 100 or greater. By default this is 0. .withDesiredStartSequenceNumber(100L) // Try to read 10 messages. If 10 messages are not available, then NotEnoughMessagesException is raised. By default, this is 1. .withMinMessageCount(10L) // Accept up to 100 messages. By default this is 1. .withMaxMessageCount(100L) // Try to wait at most 5 seconds for the min_messsage_count to be fulfilled. By default, this is 0, which immediately returns the messages or an exception. .withReadTimeoutMillis(Duration.ofSeconds(5L).toMillis()) ); } catch (StreamManagerException e) { // Properly handle exception. } ``` Referência do SDK Java: [readMessages \$1 [ReadMessagesOptions](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/ReadMessagesOptions.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#readMessages-java.lang.String-com.amazonaws.greengrass.streammanager.model.ReadMessagesOptions-) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { const messages = await client.readMessages("StreamName", // By default, if no options are specified, it tries to read one message from the beginning of the stream. new ReadMessagesOptions() // Try to read from sequence number 100 or greater. By default this is 0. .withDesiredStartSequenceNumber(100) // Try to read 10 messages. If 10 messages are not available, then NotEnoughMessagesException is thrown. By default, this is 1. .withMinMessageCount(10) // Accept up to 100 messages. By default this is 1. .withMaxMessageCount(100) // Try to wait at most 5 seconds for the minMessageCount to be fulfilled. By default, this is 0, which immediately returns the messages or an exception. .withReadTimeoutMillis(5 * 1000) ); } 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 do Node.js: [readMessages \$1 [ReadMessagesOptions](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.ReadMessagesOptions.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#readMessages) ------ ## Listar fluxos Obtenha a lista de fluxos no gerenciador de fluxos. ### Requisitos Essa operação tem os seguintes requisitos: + Versão mínima do Stream Manager SDK: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ### Exemplos O snippet a seguir obtém uma lista dos fluxos (por nome) no gerenciador de fluxo. ------ #### [ Python ] ``` client = StreamManagerClient() try: stream_names = client.list_streams() except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` Referência do SDK em Python: [list\$1streams](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.list_streams) ------ #### [ Java ] ``` try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { List streamNames = client.listStreams(); } catch (StreamManagerException e) { // Properly handle exception. } ``` Referência do SDK em Java: [ListStreams](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#listStreams--) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { const streams = await client.listStreams(); } 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 em Node.js: [listStreams](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#listStreams) ------ ## Descrever stream de mensagens Obtenha metadados sobre um fluxo, incluindo a definição, o tamanho e o status de exportação do fluxo. ### Requisitos Essa operação tem os seguintes requisitos: + Versão mínima do Stream Manager SDK: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ### Exemplos O snippet a seguir obtém metadados sobre o fluxo chamado `StreamName`, incluindo a definição, o tamanho e o status do exportador do fluxo. ------ #### [ Python ] ``` client = StreamManagerClient() try: stream_description = client.describe_message_stream(stream_name="StreamName") if stream_description.export_statuses[0].error_message: # The last export of export destination 0 failed with some error # Here is the last sequence number that was successfully exported stream_description.export_statuses[0].last_exported_sequence_number if (stream_description.storage_status.newest_sequence_number > stream_description.export_statuses[0].last_exported_sequence_number): pass # The end of the stream is ahead of the last exported sequence number except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` Referência do SDK em Python: [describe\$1message\$1stream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.describe_message_stream) ------ #### [ Java ] ``` try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { MessageStreamInfo description = client.describeMessageStream("StreamName"); String lastErrorMessage = description.getExportStatuses().get(0).getErrorMessage(); if (lastErrorMessage != null && !lastErrorMessage.equals("")) { // The last export of export destination 0 failed with some error. // Here is the last sequence number that was successfully exported. description.getExportStatuses().get(0).getLastExportedSequenceNumber(); } if (description.getStorageStatus().getNewestSequenceNumber() > description.getExportStatuses().get(0).getLastExportedSequenceNumber()) { // The end of the stream is ahead of the last exported sequence number. } } catch (StreamManagerException e) { // Properly handle exception. } ``` Referência do Java SDK: [describeMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#describeMessageStream-java.lang.String-) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { const description = await client.describeMessageStream("StreamName"); const lastErrorMessage = description.exportStatuses[0].errorMessage; if (lastErrorMessage) { // The last export of export destination 0 failed with some error. // Here is the last sequence number that was successfully exported. description.exportStatuses[0].lastExportedSequenceNumber; } if (description.storageStatus.newestSequenceNumber > description.exportStatuses[0].lastExportedSequenceNumber) { // The end of the stream is ahead of the last exported sequence number. } } 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 do Node.js: [describeMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#describeMessageStream) ------ ## Atualize o fluxo de mensagens Atualize as propriedades de um fluxo existente. Talvez você queira atualizar um fluxo se seus requisitos mudarem após a criação do fluxo. Por exemplo: + Adicione uma nova [configuração de exportação](stream-export-configurations.md) para um Nuvem AWS destino. + Aumente o tamanho máximo de um fluxo para alterar a forma como os dados são exportados ou retidos. Por exemplo, o tamanho do fluxo em combinação com sua estratégia em configurações completas pode resultar na exclusão ou rejeição dos dados antes que o gerenciador de fluxo possa processá-los. + Pause e retome as exportações; por exemplo, se as tarefas de exportação forem demoradas e você quiser racionar seus dados de upload. Os componentes do Greengrass seguem este processo de alto nível para atualizar um fluxo: 1. [Obter a descrição do fluxo.](#streammanagerclient-describe-message-stream) 1. Atualizar as propriedades de destino nos objetos correspondentes `MessageStreamDefinition` e subordinados. 1. Passar o atualizado `MessageStreamDefinition`. Certifique-se de incluir as definições completas do objeto para o fluxo atualizado. As propriedades indefinidas revertem para os valores padrão. Você pode especificar o número de sequência da mensagem a ser usada como mensagem inicial na exportação. ### Requisitos Essa operação tem os seguintes requisitos: + Versão mínima do Stream Manager SDK: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ### Exemplos O snippet a seguir atualiza o fluxo chamado `StreamName`. Ele atualiza várias propriedades de um fluxo que é exportado para o Kinesis Data Streams. ------ #### [ Python ] ``` client = StreamManagerClient() try: message_stream_info = client.describe_message_stream(STREAM_NAME) message_stream_info.definition.max_size=536870912 message_stream_info.definition.stream_segment_size=33554432 message_stream_info.definition.time_to_live_millis=3600000 message_stream_info.definition.strategy_on_full=StrategyOnFull.RejectNewData message_stream_info.definition.persistence=Persistence.Memory message_stream_info.definition.flush_on_write=False message_stream_info.definition.export_definition.kinesis= [KinesisConfig( # Updating Export definition to add a Kinesis Stream configuration. identifier=str(uuid.uuid4()), kinesis_stream_name=str(uuid.uuid4()))] client.update_message_stream(message_stream_info.definition) except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` Referência do SDK do Python: \$1 [updateMessageStream[MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.update_message_stream) ------ #### [ Java ] ``` try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) { MessageStreamInfo messageStreamInfo = client.describeMessageStream(STREAM_NAME); // Update the message stream with new values. client.updateMessageStream( messageStreamInfo.getDefinition() .withStrategyOnFull(StrategyOnFull.RejectNewData) // Required. Updating Strategy on full to reject new data. // Max Size update should be greater than initial Max Size defined in Create Message Stream request .withMaxSize(536870912L) // Update Max Size to 512 MB. .withStreamSegmentSize(33554432L) // Update Segment Size to 32 MB. .withFlushOnWrite(true) // Update flush on write to true. .withPersistence(Persistence.Memory) // Update the persistence to Memory. .withTimeToLiveMillis(3600000L) // Update TTL to 1 hour. .withExportDefinition( // Optional. Choose where/how the stream is exported to the Nuvem AWS. messageStreamInfo.getDefinition().getExportDefinition(). // Updating Export definition to add a Kinesis Stream configuration. .withKinesis(new ArrayList() {{ add(new KinesisConfig() .withIdentifier(EXPORT_IDENTIFIER) .withKinesisStreamName("test")); }}) ); } catch (StreamManagerException e) { // Properly handle exception. } ``` Referência do SDK Java: [update\$1message\$1stream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#updateMessageStream-java.lang.String-) \$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { const messageStreamInfo = await c.describeMessageStream(STREAM_NAME); await client.updateMessageStream( messageStreamInfo.definition // Max Size update should be greater than initial Max Size defined in Create Message Stream request .withMaxSize(536870912) // Default is 256 MB. Updating Max Size to 512 MB. .withStreamSegmentSize(33554432) // Default is 16 MB. Updating Segment Size to 32 MB. .withTimeToLiveMillis(3600000) // By default, no TTL is enabled. Update TTL to 1 hour. .withStrategyOnFull(StrategyOnFull.RejectNewData) // Required. Updating Strategy on full to reject new data. .withPersistence(Persistence.Memory) // Default is File. Update the persistence to Memory .withFlushOnWrite(true) // Default is false. Updating to true. .withExportDefinition( // Optional. Choose where/how the stream is exported to the Nuvem AWS. messageStreamInfo.definition.exportDefinition // Updating Export definition to add a Kinesis Stream configuration. .withKinesis([new KinesisConfig().withIdentifier(uuidv4()).withKinesisStreamName(uuidv4())]) ) ); } 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 do Node.js: \$1 [updateMessageStream[MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#updateMessageStream) ------ ### Restrições para a atualização de fluxos As restrições a seguir se aplicam ao atualizar fluxos. A menos que indicado na lista a seguir, as atualizações entrarão em vigor imediatamente. + Não é possível atualizar a persistência de um fluxo. Para alterar esse comportamento, [exclua o fluxo](#streammanagerclient-delete-message-stream) e [crie um fluxo](#streammanagerclient-create-message-stream) que defina a nova política de persistência. + Você só pode atualizar o tamanho máximo de um fluxo sob as seguintes condições: + O tamanho máximo deve ser maior que o tamanho atual do fluxo. Para encontrar essas informações, [descreva o fluxo](#streammanagerclient-describe-message-stream) e, em seguida, verifique o status de armazenamento do objeto `MessageStreamInfo` retornado. + O tamanho máximo deve ser maior ou igual ao tamanho do segmento do fluxo. + Você pode atualizar o tamanho do segmento do fluxo para um valor menor que o tamanho máximo do fluxo. A configuração atualizada se aplica aos novos segmentos. + As atualizações da propriedade tempo de vida (TTL) se aplicam às novas operações de anexação. Se você diminuir esse valor, o gerenciador de fluxo também poderá excluir segmentos existentes que excedam o TTL. + As atualizações da estratégia em toda a propriedade se aplicam às novas operações de anexação. Se você definir a estratégia para substituir os dados mais antigos, o gerenciador de fluxo também poderá substituir os segmentos existentes com base na nova configuração. + As atualizações na propriedade “descartar após gravação” se aplicam às novas mensagens. + As atualizações nas configurações de exportação se aplicam às novas exportações. A solicitação de atualização deve incluir todas as configurações de exportação às quais você deseja oferecer suporte. Caso contrário, o gerenciador de fluxo as excluirá. + Ao atualizar uma configuração de exportação, especifique o identificador da configuração de exportação de destino. + Para adicionar uma configuração de exportação, especifique um identificador exclusivo para a nova configuração de exportação. + Para excluir uma configuração de exportação, omita a configuração de exportação. + Para [atualizar](#streammanagerclient-update-message-stream) o número da sequência inicial de uma configuração de exportação em um fluxo, você deve especificar um valor menor que o número de sequência mais recente. Para encontrar essas informações, [descreva o fluxo](#streammanagerclient-describe-message-stream) e, em seguida, verifique o status de armazenamento do objeto `MessageStreamInfo` retornado. ## Excluir stream de mensagens Exclui um fluxo. Quando você exclui um fluxo, todos os dados armazenados para o fluxo são excluídos do disco. ### Requisitos Essa operação tem os seguintes requisitos: + Versão mínima do Stream Manager SDK: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ### Exemplos O snippet a seguir exclui o fluxo chamado `StreamName`. ------ #### [ Python ] ``` client = StreamManagerClient() try: client.delete_message_stream(stream_name="StreamName") except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` Referência do SDK do Python: [deleteMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.delete_message_stream) ------ #### [ Java ] ``` try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { client.deleteMessageStream("StreamName"); } catch (StreamManagerException e) { // Properly handle exception. } ``` Referência do SDK em Java: [delete\$1message\$1stream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#deleteMessageStream-java.lang.String-) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { await client.deleteMessageStream("StreamName"); } 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 do Node.js: [deleteMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#deleteMessageStream) ------ ## Consulte também + [Gerenciar fluxos de dados no nos dispositivos principais do Greengrass](manage-data-streams.md) + [Configurar o gerenciador de fluxo do AWS IoT Greengrass](configure-stream-manager.md) + [Configurações de exportação para destinos compatíveis Nuvem AWS](stream-export-configurations.md) + `StreamManagerClient` na referência do Stream Manager SDK: + [Python](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html) + [Java](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html) + [Node.js](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html) # Configurações de exportação para destinos compatíveis 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](work-with-streams.md#streammanagerclient-create-message-stream) ou [atualiza um fluxo](work-with-streams.md#streammanagerclient-create-message-stream), 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. ![\[Diagrama do modelo de objeto do tipo de ExportDefinition propriedade.\]](http://docs.aws.amazon.com/pt_br/greengrass/v2/developerguide/images/stream-manager-exportconfigs.png) É possível 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 stream manager tenta continuamente exportar dados para o Nuvem AWS em intervalos de até cinco minutos. Não há um limite máximo para o número de novas tentativas. **nota** O `StreamManagerClient` também fornece um destino alvo que você pode usar para exportar fluxos para um servidor HTTP. Este destino deve ser usado apenas para fins de teste. Ele não é estável e nem compatível para uso em ambientes de produção. **Topics** + [AWS IoT Analytics canais](#export-to-iot-analytics) + [Amazon Kinesis Data Streams](#export-to-kinesis) + [AWS IoT SiteWise propriedades do ativo](#export-to-iot-sitewise) + [Objetos do Amazon S3](#export-to-s3) Você é responsável pela manutenção desses Nuvem AWS recursos. ## AWS IoT Analytics canais O gerenciador de fluxo suporta exportações automáticas para AWS IoT Analytics. AWS IoT Analytics permite realizar análises avançadas em seus dados para ajudar a tomar decisões comerciais e melhorar os modelos de aprendizado de máquina. Para obter mais informações, consulte [O que é AWS IoT Analytics?](https://docs.aws.amazon.com/iotanalytics/latest/userguide/welcome.html) no *Guia do AWS IoT Analytics usuário*. 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: + [Io TAnalytics Config no SDK](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.IoTAnalyticsConfig) do Python + [Io TAnalytics Config no Java](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTAnalyticsConfig.html) SDK + [Io TAnalytics Config no SDK](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTAnalyticsConfig.html) do Node.js ### Requisitos Esse destino de exportação tem os seguintes requisitos: + Os canais de destino AWS IoT Analytics devem estar no mesmo dispositivo principal Conta da AWS e Região da AWS no dispositivo principal do Greengrass. + O [Autorize os dispositivos principais a interagir com os serviços AWS](device-service-role.md) deve conceder a permissão `iotanalytics:BatchPutMessage` para os canais de destino. Por exemplo: ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "iotanalytics:BatchPutMessage" ], "Resource": [ "arn:aws:iotanalytics:us-east-1:123456789012:channel/channel_1_name", "arn:aws:iotanalytics:us-east-1:123456789012:channel/channel_2_name" ] } ] } ``` ------ É possível conceder acesso granular ou condicional aos recursos, por exemplo, usando um esquema de nomeação `*` curinga. Para obter mais informações, consulte [Adicionando e removendo políticas do IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) no *Guia do usuário do IAM*. ### Exportando para AWS IoT Analytics Para criar um fluxo que exporta para AWS IoT Analytics, seus componentes do Greengrass [criam um fluxo](work-with-streams.md#streammanagerclient-create-message-stream) com uma definição de exportação que inclui um ou mais `IoTAnalyticsConfig` objetos. 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](work-with-streams.md#streammanagerclient-append-message) 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. MapReduce Para obter mais informações, consulte [O que é o Amazon Kinesis Data Streams?](https://docs.aws.amazon.com/streams/latest/dev/what-is-this-service.html) 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: + [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.KinesisConfig)no SDK do Python + [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/KinesisConfig.html)no Java SDK + [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.KinesisConfig.html)no SDK do Node.js ### Requisitos Esse destino de exportação tem os seguintes requisitos: + Os streams de destino no Kinesis Data Streams devem estar no mesmo dispositivo principal do Conta da AWS Região da AWS Greengrass. + (Recomendado) O Gerenciador de fluxos v2.2.1 melhora a performance da exportação de fluxos para destinos do Kinesis Data Streams. Para usar as melhorias nesta versão mais recente, atualize seu [componente do gerenciador de fluxos](stream-manager-component.md) para a versão 2.2.1 e use a política `kinesis:ListShards` no [perfil de troca de tokens do Greengrass](device-service-role.md). + O [Autorize os dispositivos principais a interagir com os serviços AWS](device-service-role.md) deve conceder a permissão `kinesis:PutRecords` para os fluxos de dados de destino. Por exemplo: ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "kinesis:PutRecords" ], "Resource": [ "arn:aws:kinesis:us-east-1:123456789012:stream/stream_1_name", "arn:aws:kinesis:us-east-1:123456789012:stream/stream_2_name" ] } ] } ``` ------ É possível conceder acesso granular ou condicional aos recursos, por exemplo, usando um esquema de nomeação `*` curinga. Para obter mais informações, consulte [Adicionando e removendo políticas do IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) no *Guia do usuário do IAM*. ### Exportação do Kinesis Data Streams Para criar um fluxo que exporte para o Kinesis Data Streams, os componentes do Greengrass [criam um fluxo](work-with-streams.md#streammanagerclient-create-message-stream) 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](work-with-streams.md#streammanagerclient-append-message) 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. ## AWS IoT SiteWise propriedades do ativo O gerenciador de fluxo suporta exportações automáticas para AWS IoT SiteWise. AWS IoT SiteWise permite coletar, organizar e analisar dados de equipamentos industriais em grande escala. Para obter mais informações, consulte [O que é AWS IoT SiteWise?](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/what-is-sitewise.html) no *Guia do AWS IoT SiteWise 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: + [Io TSite WiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.IoTSiteWiseConfig) no SDK do Python + [Io TSite WiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTSiteWiseConfig.html) no Java SDK + [Io TSite WiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTSiteWiseConfig.html) no SDK do Node.js **nota** AWS também fornece AWS IoT SiteWise componentes, que oferecem uma solução pré-construída que você pode usar para transmitir dados de fontes OPC-UA. Para obter mais informações, consulte [Coletor IoT SiteWise OPC UA](iotsitewise-opcua-collector-component.md). ### Requisitos Esse destino de exportação tem os seguintes requisitos: + As propriedades do ativo de destino AWS IoT SiteWise devem estar no mesmo Conta da AWS Região da AWS dispositivo central do Greengrass. **nota** Para ver a lista de Região da AWS s que AWS IoT SiteWise oferecem suporte, consulte [AWS IoT SiteWise endpoints e cotas](https://docs.aws.amazon.com/general/latest/gr/iot-sitewise.html#iot-sitewise_region) na Referência *AWS geral*. + O [Autorize os dispositivos principais a interagir com os serviços AWS](device-service-role.md) deve conceder a permissão `iotsitewise:BatchPutAssetPropertyValue` para as propriedades do ativo do destino. O exemplo de política a seguir usa a chave de condição `iotsitewise:assetHierarchyPath` para conceder acesso a um ativo raiz de destino e seus ativos secundários. Você pode remover o `Condition` da política para permitir o acesso a todos os seus AWS IoT SiteWise ativos ou especificar ARNs ativos individuais. ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": "iotsitewise:BatchPutAssetPropertyValue", "Resource": "*", "Condition": { "StringLike": { "iotsitewise:assetHierarchyPath": [ "/root node asset ID", "/root node asset ID/*" ] } } } ] } ``` ------ É possível conceder acesso granular ou condicional aos recursos, por exemplo, usando um esquema de nomeação `*` curinga. Para obter mais informações, consulte [Adicionando e removendo políticas do IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) no *Guia do usuário do IAM*. Para obter informações de segurança importantes, consulte a [ BatchPutAssetPropertyValue autorização](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/security_iam_service-with-iam.html#security_iam_service-with-iam-id-based-policies-batchputassetpropertyvalue-action) no *Guia AWS IoT SiteWise do usuário*. ### Exportando para AWS IoT SiteWise Para criar um fluxo que exporta para AWS IoT SiteWise, seus componentes do Greengrass [criam um fluxo](work-with-streams.md#streammanagerclient-create-message-stream) com uma definição de exportação que inclui um ou mais `IoTSiteWiseConfig` objetos. 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](work-with-streams.md#streammanagerclient-append-message-sitewise) para destinos de exportação do AWS IoT SiteWise . **nota** Quando você envia dados para AWS IoT SiteWise, seus dados devem atender aos requisitos da `BatchPutAssetPropertyValue` ação. Para obter mais informações, consulte [BatchPutAssetPropertyValue](https://docs.aws.amazon.com/iot-sitewise/latest/APIReference/API_BatchPutAssetPropertyValue.html) na *Referência de APIs 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. É possível 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: + Para exportações quase em tempo real, defina configurações baixas de tamanho de lote e intervalo e anexe os dados ao fluxo quando forem recebidos. + Para otimizar o agrupamento em lotes, mitigar as restrições de largura de banda ou minimizar os custos, seus componentes do Greengrass podem agrupar os pontos de dados timestamp-quality-value (TQV) recebidos para uma única propriedade do ativo antes de anexar os dados ao stream. Uma estratégia é agrupar entradas para até 10 (dez) combinações diferentes de propriedade e ativo, ou aliases de propriedade, em uma mensagem, em vez de enviar mais de uma entrada para a mesma propriedade. Isso ajuda o gerenciador de fluxo a permanecer dentro das [cotas do AWS IoT SiteWise](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/quotas.html). ## Objetos do Amazon S3 O gerenciador de fluxo é compatível com exportações automáticas para o Amazon S3. É possível utilizar o Amazon S3 para armazenar e recuperar grandes volumes de dados. Para obter mais informações, consulte [O que é o Amazon S3?](https://docs.aws.amazon.com/AmazonS3/latest/dev/Welcome.html) 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: + [S3 ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.S3ExportTaskExecutorConfig) no SDK do Python + [S3 ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/S3ExportTaskExecutorConfig.html) no Java SDK + [S3 ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.S3ExportTaskExecutorConfig.html) no SDK do Node.js ### Requisitos Esse destino de exportação tem os seguintes requisitos: + Os buckets do Amazon S3 de destino devem estar no mesmo dispositivo principal Conta da AWS do Greengrass. + Se uma função do Lambda executada no modo de **contêiner do Greengrass** grava arquivos de entrada no diretório de arquivos de entrada, é necessário montar o diretório como um volume no contêiner com permissões de gravação. Esse procedimento garante que os arquivos sejam gravados no sistema de arquivos raiz e fiquem visíveis para o componente Gerenciador de fluxos, executado fora do contêiner. + Se um componente de contêiner do Docker grava arquivos de entrada no diretório de arquivos de entrada, é necessário montar o diretório como um volume no contêiner com permissões de gravação. Esse procedimento garante que os arquivos sejam gravados no sistema de arquivos raiz e fiquem visíveis para o componente Gerenciador de fluxos, executado fora do contêiner. + O [Autorize os dispositivos principais a interagir com os serviços AWS](device-service-role.md) deve conceder as permissões a seguir para os buckets de destino. Por exemplo: ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "s3:PutObject", "s3:AbortMultipartUpload", "s3:ListMultipartUploadParts" ], "Resource": [ "arn:aws:s3:::bucket-1-name/*", "arn:aws:s3:::bucket-2-name/*" ] } ] } ``` ------ É possível conceder acesso granular ou condicional aos recursos, por exemplo, usando um esquema de nomeação `*` curinga. Para obter mais informações, consulte [Adicionando e removendo políticas do IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) no *Guia do usuário do IAM*. ### 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. **nota** O bucket de destino já deve existir no seu Conta da AWS. Se um objeto para a chave especificada não existir, o gerenciador de fluxo criará o objeto para você. O gerenciador de fluxo usa a propriedade de limite de upload de várias partes, a configuração do [tamanho mínimo das partes](configure-stream-manager.md#stream-manager-minimum-part-size) 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 DateTimeFormatter cadeias de caracteres [Java](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html) válidas em espaços reservados. `!{timestamp:value}` É possível 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 ``` **nota** Se você quiser monitorar o status de exportação de um fluxo, primeiro crie um fluxo de status e, em seguida, configure o fluxo de exportação para usá-lo. Para obter mais informações, consulte [Monitorar tarefas de exportação](#monitor-export-status-s3). #### Gerenciar dados de entrada É possível criar códigos que as aplicações 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. 1. Um processo local recebe dados de dispositivos ou periféricos e, em seguida, grava os dados em arquivos em um diretório no dispositivo principal. Esses são os arquivos de entrada para o gerenciador de fluxo. 1. O componente do Greengrass verifica o diretório e [anexa uma tarefa de exportação](work-with-streams.md#streammanagerclient-append-message-export-task) ao fluxo de destino quando um novo arquivo é criado. A tarefa é um objeto `S3ExportTaskDefinition` serializado em JSON que especifica a URL do arquivo de entrada, o bucket e a chave do Amazon S3 de destino, além dos metadados opcionais do usuário. 1. O gerenciador de fluxo lê o arquivo de entrada e exporta os dados para o Amazon S3 na ordem das tarefas anexadas. O bucket de destino já deve existir no seu Conta da AWS. Se um objeto para a chave especificada não existir, o gerenciador de fluxo criará o objeto para você. 1. O componente do Greengrass [lê as mensagens](work-with-streams.md#streammanagerclient-read-messages) de um fluxo de status para monitorar o status da exportação. Depois que as tarefas de exportação forem concluídas, o componente do Greengrass poderá excluir os arquivos de entrada correspondentes. Para obter mais informações, consulte [Monitorar tarefas de exportação](#monitor-export-status-s3). ### Monitorar tarefas de exportação É possível criar códigos que as aplicações 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](work-with-streams.md#streammanagerclient-create-message-stream) para usar como fluxo de status. É possível configurar as políticas de tamanho e retenção do fluxo para controlar a vida útil das mensagens de status. Por exemplo: + Defina `Persistence` como `Memory` se você não quiser armazenar as mensagens de status. + Defina `StrategyOnFull` como `OverwriteOldestData` para que as novas mensagens de status não sejam perdidas. 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 com suporte a seguir variam do menos detalhado (`ERROR`) ao mais detalhado (`TRACE`). O padrão é `INFO`. + `ERROR` + `WARN` + `INFO` + `DEBUG` + `TRACE` 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. 1. Conforme descrito no fluxo de trabalho anterior, um componente do [anexa uma tarefa de exportação](work-with-streams.md#streammanagerclient-append-message-export-task) a um fluxo configurado para gravar mensagens de status sobre as tarefas de exportação em um fluxo de status. A operação de append retorna um número de sequência que representa a ID da tarefa. 1. U componente do Greengrass [lê as mensagens](work-with-streams.md#streammanagerclient-read-messages) sequencialmente do fluxo de status e as filtra com base no nome do fluxo e no ID da tarefa, ou com base em uma propriedade da tarefa de exportação do contexto da mensagem. Por exemplo, o componente do Greengrass pode filtrar por URL do arquivo de entrada da tarefa de exportação, que é representada pelo objeto `S3ExportTaskDefinition` no contexto da mensagem. Os códigos de status a seguir indicam que uma tarefa de exportação atingiu um estado concluído: + `Success`. O upload foi concluído com êxito. + `Failure`. O gerenciador de fluxo encontrou um erro, por exemplo, o bucket especificado não existe. Depois de resolver o problema, você pode reanexar a tarefa de exportação ao fluxo. + `Canceled`. A tarefa foi interrompida porque a definição de fluxo ou exportação foi excluída ou o período time-to-live (TTL) da tarefa expirou. **nota** A tarefa também pode ter um status de `InProgress` ou `Warning`. O gerenciador de fluxo emite avisos quando um evento retorna um erro que não afeta a execução da tarefa. Por exemplo, uma falha na limpeza de um upload parcial retorna um aviso. 1. Depois que as tarefas de exportação forem concluídas, o componente do Greengrass poderá excluir os arquivos de entrada correspondentes. 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 do Python: read\$1messages \$1 [StatusMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.StatusMessage)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.read_messages) ------ #### [ 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 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 Java: [readMessages \$1 [StatusMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/StatusMessage.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#readMessages-java.lang.String-com.amazonaws.greengrass.streammanager.model.ReadMessagesOptions-) ------ #### [ 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 do Node.js: [readMessages \$1 [StatusMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StatusMessage.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#readMessages) ------