Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.
# Administración de flujos de datos en los dispositivos principales de Greengrass
AWS IoT Greengrass stream manager hace que sea más eficiente y confiable transferir datos de IoT de gran volumen al Nube de AWS. Stream Manager procesa los flujos de datos en el AWS IoT Greengrass Core antes de exportarlos al Nube de AWS. Stream Manager se integra en escenarios periféricos comunes, como la inferencia del aprendizaje automático (ML), en la que el dispositivo AWS IoT Greengrass Core procesa y analiza los datos antes de exportarlos a los destinos de almacenamiento locales Nube de AWS o a los destinos de almacenamiento.
El administrador de flujos proporciona una interfaz común para simplificar el desarrollo de componentes personalizados para que no sea necesario crear una funcionalidad de administración de flujos personalizada. Sus componentes pueden usar un mecanismo estandarizado para procesar flujos de gran volumen y administrar las políticas locales de retención de datos. Puede definir políticas para el tipo de almacenamiento, el tamaño y la retención de datos según cada flujo para controlar cómo el administrador de flujos procesa y exporta datos.
El administrador de flujos trabaja en entornos con conectividad intermitente o limitada. Puede definir el uso del ancho de banda, el comportamiento de los tiempos de espera y la forma en que el AWS IoT Greengrass Core gestiona los datos de la transmisión cuando está conectado o desconectado. También puede establecer prioridades para controlar el orden en el que AWS IoT Greengrass Core exporta flujos a la Nube de AWS. Esto le permite gestionar los datos críticos antes que otros datos.
Puede configurar el administrador de transmisiones para que exporte automáticamente los datos a él Nube de AWS para su almacenamiento o posterior procesamiento y análisis. El administrador de flujos admite exportar a los siguientes destinos de la Nube de AWS :
+ Canales de entrada AWS IoT Analytics. AWS IoT Analytics le permite realizar análisis avanzados de sus datos para ayudarle a tomar decisiones empresariales y mejorar los modelos de aprendizaje automático. Para obtener más información, consulte [¿Qué es AWS IoT Analytics?](https://docs.aws.amazon.com/iotanalytics/latest/userguide/welcome.html) en la *Guía del usuario de AWS IoT Analytics *.
+ Flujos de Amazon Kinesis Data Streams. Puede usar Kinesis Data Streams para agregar grandes volúmenes de datos y cargarlos en un almacén MapReduce de datos o un clúster. Para obtener más información, consulte [Qué son los Amazon Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/what-is-this-service.html) en la *Guía para desarrolladores de Amazon Kinesis Data Streams*.
+ Propiedades de los activos en. AWS IoT SiteWise AWS IoT SiteWise le permite recopilar, organizar y analizar datos de equipos industriales a escala. Para obtener más información, consulte [¿Qué es AWS IoT SiteWise?](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/what-is-sitewise.html) en la *Guía AWS IoT SiteWise del usuario*.
+ Objetos en Amazon Simple Storage Service Amazon S3. Puede utilizar Amazon S3 para almacenar y recuperar grandes cantidades de datos. Para obtener más información, consulte [¿Qué es Amazon S3?](https://docs.aws.amazon.com/AmazonS3/latest/dev/Welcome.html) en la *Guía para desarrolladores de Amazon Simple Storage Service*.
## Flujo de trabajo de la administración de secuencias
Sus aplicaciones de IoT interactúan con el administrador de flujos a través del SDK del administrador de flujos.
En un flujo de trabajo simple, un componente del AWS IoT Greengrass núcleo consume datos de IoT, como métricas de temperatura y presión de series temporales. El componente podría filtrar o comprimir los datos y luego llamar al SDK del administrador de flujos para escribir los datos en un flujo en el administrador de flujos. Stream Manager puede exportar la transmisión a la transmisión Nube de AWS automáticamente en función de las políticas que defina para la transmisión. Los componentes también pueden enviar datos directamente a bases de datos locales o repositorios de almacenamiento.
Sus aplicaciones de IoT pueden incluir múltiples componentes personalizados que leen o escriben en flujos. Estos componentes pueden leer y escribir en las transmisiones para filtrar, agregar y analizar los datos del dispositivo AWS IoT Greengrass principal. Esto permite responder rápidamente a los eventos locales y extraer información valiosa antes de que los datos se transfieran del núcleo a los Nube de AWS destinos locales.
Para empezar, implemente el componente administrador de transmisiones en su dispositivo AWS IoT Greengrass principal. En la implementación, configure los parámetros del componente del administrador de flujos para definir los ajustes que se apliquen a todos los flujos del dispositivo principal de Greengrass. Use estos parámetros para controlar cómo el administrador de flujos almacena, procesa y exporta flujos en función de las necesidades de su negocio y las restricciones del entorno.
Después de configurar el administrador de flujos, puede crear e implementar sus aplicaciones de IoT. Por lo general, se trata de componentes personalizados que utilizan `StreamManagerClient` en el SDK del administrador de flujos para crear flujos e interactuar con ellos. Cuando crea un flujo, puede definir las políticas por flujo, como los destinos de exportación, la prioridad y la persistencia.
## Requisitos
Se aplican los siguientes requisitos para el administrador de flujos:
+ Stream Manager requiere un mínimo de 70 MB de RAM además del software AWS IoT Greengrass Core. El requisito total de memoria depende de la carga de trabajo.
+ AWS IoT Greengrass los componentes deben usar el SDK de Stream Manager para interactuar con Stream Manager. El SDK del administrador de flujos está disponible en los siguientes lenguajes:
+ [SDK del administrador de flujos para Java](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-java/) (versión 1.1.0 o posterior)
+ [SDK del administrador de flujos para Node.js](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-js/) (versión 1.1.0 o posterior)
+ [SDK del administrador de flujos para Python](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-python/) (versión 1.1.0 o posterior)
+ AWS IoT Greengrass los componentes deben especificar el componente del administrador de transmisiones (`aws.greengrass.StreamManager`) como una dependencia en su receta para usar el administrador de transmisiones.
**nota**
Si usa el administrador de flujos para exportar datos a la nube, no puede actualizar la versión 2.0.7 del componente de administrador de flujos a una versión entre la 2.0.8 y la 2.0.11. Si implementa el administrador de flujos por primera vez, le recomendamos que implemente la última versión del componente administrador de flujos.
+ Si define los destinos de Nube de AWS exportación para una transmisión, debe crear sus objetivos de exportación y conceder permisos de acceso en la función de [dispositivo de Greengrass](device-service-role.md). Según el destino, es posible que también se apliquen otros requisitos. Para obtener más información, consulte lo siguiente:
+ [AWS IoT Analytics canales](stream-export-configurations.md#export-to-iot-analytics)
+ [Amazon Kinesis Data Streams](stream-export-configurations.md#export-to-kinesis)
+ [AWS IoT SiteWise propiedades de los activos](stream-export-configurations.md#export-to-iot-sitewise)
+ [Objetos de Amazon S3](stream-export-configurations.md#export-to-s3)
Usted es responsable del mantenimiento de estos Nube de AWS recursos.
## Seguridad de los datos
Cuando utilice el administrador de secuencias, tenga en cuenta las siguientes consideraciones de seguridad.
### Seguridad de los datos locales
AWS IoT Greengrass no cifra los datos de la transmisión en reposo o en tránsito entre los componentes locales del dispositivo principal.
+ **Datos en reposo**. Los datos de secuencias se almacenan localmente en un directorio de almacenamiento. Para garantizar la seguridad de los datos, AWS IoT Greengrass se basa en los permisos de los archivos y en el cifrado de disco completo, si está activado. Puede utilizar el parámetro opcional [STREAM\$1MANAGER\$1STORE\$1ROOT\$1DIR](configure-stream-manager.md#STREAM_MANAGER_STORE_ROOT_DIR) para especificar el directorio de almacenamiento. Si cambia este parámetro más adelante para usar un directorio de almacenamiento diferente, AWS IoT Greengrass no elimina el directorio de almacenamiento anterior ni su contenido.
+ **Datos en tránsito local**. AWS IoT Greengrass no cifra los datos de transmisión en tránsito local entre las fuentes de datos, AWS IoT Greengrass los componentes, el SDK de Stream Manager y el administrador de transmisiones.
+ **Datos en tránsito hacia**. Nube de AWS Los flujos de datos exportados por el administrador de flujos para Nube de AWS utilizar el cifrado de cliente de AWS servicio estándar con Transport Layer Security (TLS).
### Autenticación del cliente
Los clientes del administrador de flujos utilizan el SDK del administrador de flujos para comunicarse con el administrador de flujos. Cuando la autenticación de cliente está habilitada, solo los componentes de Greengrass pueden interactuar con los flujos en el administrador de flujos. Cuando la autenticación de cliente está deshabilitada, cualquier proceso que se ejecute en el dispositivo principal de Greengrass puede interactuar con los flujos en el administrador de flujos. Debe deshabilitar la autenticación solo si su caso de negocio lo requiere.
Utilice el parámetro [STREAM\$1MANAGER\$1AUTHENTICATE\$1CLIENT](configure-stream-manager.md#STREAM_MANAGER_AUTHENTICATE_CLIENT) para establecer el modo de autenticación del cliente. Puede configurar este parámetro cuando implementa el componente administrador de flujos en los dispositivos principales.
****
| | Habilitado | Deshabilitado |
| --- | --- | --- |
| Valor del parámetro | `true` (predeterminado y recomendado) | `false` |
| Clientes permitidos | Componentes de Greengrass en el dispositivo principal | Componentes de Greengrass en el dispositivo principal Otros procesos que se ejecutan en el dispositivo del núcleo de Greengrass |
## Véase también
+ [Configurar el administrador de secuencias de AWS IoT Greengrass](configure-stream-manager.md)
+ [Se usa StreamManagerClient para trabajar con transmisiones](work-with-streams.md)
+ [Exportación de configuraciones para Nube de AWS destinos compatibles](stream-export-configurations.md)
# Configurar el administrador de secuencias de AWS IoT Greengrass
En los dispositivos principales de Greengrass, el administrador de flujos puede almacenar, procesar y exportar datos enviados desde dispositivos IoT. El administrador de flujos proporciona parámetros que se usan para configurar los ajustes de tiempo de ejecución. Esta configuración se aplica a todos los flujos del dispositivo principal de Greengrass. Puede usar la consola o la API de AWS IoT Greengrass para configurar los ajustes del administrador del flujos cuando implemente el componente. Los cambios surten efecto después de que se completa la implementación.
## Parámetros del administrador de secuencias
El administrador de flujos proporciona los siguientes parámetros que puede configurar al implementar el componente en sus dispositivos principales. Todos los parámetros son opcionales.
**Directorio de almacenamiento**
Nombre del parámetro: `STREAM_MANAGER_STORE_ROOT_DIR`
La ruta absoluta de la carpeta local usada para almacenar flujos. Este valor debe comenzar con una barra inclinada (por ejemplo, `/data`).
Debe especificar una carpeta existente y el [usuario del sistema que ejecuta el componente de administrador de flujos](configure-greengrass-core-v2.md#configure-component-user) debe tener permisos para leer y escribir en esta carpeta. Por ejemplo, puede ejecutar los siguientes comandos para crear y configurar una carpeta, `/var/greengrass/streams`, que especifique como carpeta raíz del administrador de flujos. Estos comandos permiten al usuario predeterminado del sistema, `ggc_user`, leer y escribir en esta carpeta.
```
sudo mkdir /var/greengrass/streams
sudo chown ggc_user /var/greengrass/streams
sudo chmod 700 /var/greengrass/streams
```
Para obtener información sobre cómo proteger los datos de secuencias, consulte [Seguridad de los datos locales](manage-data-streams.md#stream-manager-security-stream-data).
Valor predeterminado: `/greengrass/v2/work/aws.greengrass.StreamManager`
**Puerto del servidor**
Nombre del parámetro: `STREAM_MANAGER_SERVER_PORT`
El número de puerto local utilizado para comunicarse con el administrador de secuencias. El valor predeterminado es `8088`.
Puede especificar `0` para usar un puerto disponible asignado al azar.
**Autenticar cliente**
Nombre del parámetro: `STREAM_MANAGER_AUTHENTICATE_CLIENT`
Indica si los clientes deben autenticarse para interactuar con el administrador de secuencias. El SDK del administrador de flujos controla toda la interacción entre los clientes y el administrador de flujos. Este parámetro determina qué clientes pueden llamar al SDK del administrador de flujos para trabajar con flujos. Para obtener más información, consulte [Autenticación del cliente](manage-data-streams.md#stream-manager-security-client-authentication).
Los valores válidos son `true` o `false`. El valor predeterminado es `true` (recomendado).
+ `true`. Solo permite como clientes a los componentes de Greengrass. Los componentes usan protocolos de AWS IoT Greengrass Core internos para autenticarse con el SDK del administrador de flujos.
+ `false`. Permite que cualquier proceso que se ejecute en el AWS IoT Greengrass Core sea un cliente. No establezca el valor en `false`, a menos que su caso de negocio lo requiera. Por ejemplo, use `false` solo si los procesos no componentes en el dispositivo principal deben comunicarse directamente con el administrador de flujos.
**Ancho de banda máximo**
Nombre del parámetro: `STREAM_MANAGER_EXPORTER_MAX_BANDWIDTH`
El ancho de banda máximo promedio (en kilobits por segundo) que se puede utilizar para exportar datos. El valor predeterminado permite el uso ilimitado del ancho de banda disponible.
**Tamaño del grupo de subprocesos**
Nombre del parámetro: `STREAM_MANAGER_EXPORTER_THREAD_POOL_SIZE`
Cantidad máxima de subprocesos activos que se pueden utilizar para exportar datos. El valor predeterminado es `5`.
El tamaño óptimo depende del hardware, el volumen de secuencias y la cantidad planificada de secuencias de exportación. Si la velocidad de exportación es lenta, puede ajustar esta configuración para encontrar el tamaño óptimo para su hardware y su caso de negocio. La CPU y la memoria del hardware del dispositivo principal son factores limitantes. Para comenzar, puede intentar establecer este valor igual a la cantidad de núcleos de procesador en el dispositivo.
Tenga cuidado de no establecer un tamaño superior al que admite el hardware. Cada flujo utiliza recursos de hardware, por lo que debe intentar limitar la cantidad de flujos de exportación en dispositivos restringidos.
**Argumentos de JVM**
Nombre del parámetro: `JVM_ARGS`
Argumentos personalizados de la máquina virtual de Java para pasar al administrador de secuencias al inicio. Varios argumentos deben separarse por espacios.
Utilice este parámetro sólo cuando deba anular la configuración predeterminada utilizada por la JVM. Por ejemplo, puede que necesite aumentar el tamaño predeterminado del montón si planea exportar un gran número de secuencias.
**Nivel de registro**
Nombre del parámetro: `LOG_LEVEL`
El nivel de registro del componente. Elija uno de los siguientes niveles de registro, que se enumeran aquí en orden de niveles:
+ `TRACE`
+ `DEBUG`
+ `INFO`
+ `WARN`
+ `ERROR`
Valor predeterminado: `INFO`
**Tamaño mínimo para la carga de varias partes**
Nombre del parámetro: `STREAM_MANAGER_EXPORTER_S3_DESTINATION_MULTIPART_UPLOAD_MIN_PART_SIZE_BYTES`
El tamaño mínimo (en bytes) de una parte en una carga multiparte a Amazon S3. El administrador de flujos utiliza esta configuración y el tamaño del archivo de entrada para determinar cómo agrupar los datos en una solicitud PUT de varias partes. El valor predeterminado y mínimo es de `5242880` bytes (5 MB).
El administrador de flujos usa la propiedad `sizeThresholdForMultipartUploadBytes` de la transmisión para determinar si se debe exportar a Amazon S3 como una carga única o multiparte. Los componentes de Greengrass definidos por el usuario establecen este umbral cuando crean un flujo que se exporta a Amazon S3. El umbral de tamaño predeterminado es 5 MB.
## Véase también
+ [Administración de flujos de datos en los dispositivos principales de Greengrass](manage-data-streams.md)
+ [Se usa StreamManagerClient para trabajar con transmisiones](work-with-streams.md)
+ [Exportación de configuraciones para Nube de AWS destinos compatibles](stream-export-configurations.md)
# Creación de componentes personalizados que usen el administrador de flujos
Utilice el administrador de flujos en componentes personalizados de Greengrass para almacenar, procesar y exportar datos de dispositivos IoT. Utilice los procedimientos y ejemplos de esta sección para crear recetas de componentes, artefactos y aplicaciones que funcionen con el administrador de flujos. Para obtener más información sobre cómo desarrollar y probar componentes, consulte [Creación de componentes de AWS IoT Greengrass](create-components.md).
**Topics**
+ [Definición de las recetas de componentes que utilizan el administrador de flujos](#stream-manager-recipes)
+ [Conexión al administrador de flujos en el código de la aplicación](#connect-to-stream-manager)
## Definición de las recetas de componentes que utilizan el administrador de flujos
Para usar el administrador de flujos en un componente personalizado, debe definir el componente `aws.greengrass.StreamManager` como una dependencia. También debe proporcionar el SDK del administrador de flujos. Complete las siguientes tareas para descargar y usar el SDK del administrador de flujos en el lenguaje que prefiera.
### Uso del SDK del administrador de flujos para Java
El SDK del administrador de flujos para Java está disponible como un archivo JAR que puede usar para compilar su componente. A continuación, puede crear un JAR de aplicación que incluya el SDK del administrador de flujos, definir el JAR de aplicación como un artefacto de componente y ejecutar el JAR de aplicación en el ciclo de vida del componente.
**Uso del SDK del administrador de flujos para Java**
1. Descargue el [archivo JAR del SDK del administrador de flujos para Java](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-java/blob/main/sdk/aws-greengrass-stream-manager-sdk-java.jar).
1. Realice una de las siguientes acciones para crear artefactos componentes a partir de su aplicación Java y el archivo JAR del SDK del administrador de flujos:
+ Cree su aplicación como un archivo JAR que incluya el JAR del SDK del administrador de flujos y ejecute este archivo JAR en la receta de su componente.
+ Defina el JAR del SDK del administrador de flujos como un artefacto componente. Agregue ese artefacto a la ruta de clases cuando ejecute su aplicación en la receta de su componente.
La receta de su componente podría parecerse al siguiente ejemplo. Este componente ejecuta una versión modificada del ejemplo de [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` incluye el JAR del SDK de Stream Manager.
------
#### [ 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 obtener más información sobre cómo desarrollar y probar componentes, consulte [Creación de componentes de AWS IoT Greengrass](create-components.md).
### Uso del SDK del administrador de flujos para Python
El SDK del administrador de flujos para Python está disponible como código de origen que puede incluir en su componente. Cree un archivo ZIP del SDK del administrador de flujo, defina el archivo ZIP como un artefacto de un componente e instale los requisitos del SDK en el ciclo de vida del componente.
**Uso del SDK del administrador de flujos para Python**
1. Clona o descarga el repositorio [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. Cree un archivo ZIP que contenga la carpeta `stream_manager`, que incluye el código de origen del SDK del administrador de flujos para Python. Puede proporcionar este archivo ZIP como un artefacto componente que el software AWS IoT Greengrass Core descomprimirá al instalar el componente. Haga lo siguiente:
1. Abra la carpeta que contiene el repositorio que clonó o descargó en el paso anterior.
```
cd aws-greengrass-stream-manager-sdk-python
```
1. Comprima la carpeta `stream_manager` en un archivo ZIP llamado `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. Compruebe que el archivo `stream_manager_sdk.zip` contiene la carpeta `stream_manager` y su contenido. Ejecute el siguiente comando para hacer una lista del contenido del archivo ZIP.
------
#### [ Linux or Unix ]
```
unzip -l stream_manager_sdk.zip
```
------
#### [ Windows Command Prompt (CMD) ]
```
tar -tf stream_manager_sdk.zip
```
------
El resultado de debería parecerse al siguiente.
```
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 los artefactos del SDK del administrador de flujos a la carpeta de artefactos de su componente. Además del archivo ZIP del SDK del administrador de flujos, su componente utiliza el archivo `requirements.txt` del SDK para instalar las dependencias del SDK del administrador de flujos. *\$1/greengrass-components*Sustitúyalo por la ruta a la carpeta que utilizas para el desarrollo 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. Cree su receta de componentes. En la receta, haga lo siguiente:
1. Defina `stream_manager_sdk.zip` y `requirements.txt` como artefactos.
1. Defina su aplicación de Python como un artefacto.
1. En el ciclo de vida de la instalación, instale los requisitos del SDK del administrador de flujos desde `requirements.txt`.
1. En el ciclo de vida de ejecución, agregue el SDK del administrador de flujos a `PYTHONPATH` y ejecute la aplicación Python.
La receta de su componente podría parecerse al siguiente ejemplo. Este componente ejecuta el ejemplo de [stream\$1manager\$1s3.py](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-python/blob/main/samples/stream_manager_s3.py).
------
#### [ 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 obtener más información sobre cómo desarrollar y probar componentes, consulte [Creación de componentes de AWS IoT Greengrass](create-components.md).
### Usa el SDK de Stream Manager para JavaScript
El SDK de Stream Manager JavaScript está disponible como código fuente que puede incluir en su componente. Cree un archivo ZIP del SDK del administrador de flujo, defina el archivo ZIP como un artefacto de un componente e instale el SDK en el ciclo de vida del componente.
**Para usar el SDK de Stream Manager para JavaScript**
1. Clona o descarga el repositorio [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. Crea un archivo ZIP que contenga la `aws-greengrass-stream-manager-sdk` carpeta, que contiene el código fuente del SDK de Stream Manager para. JavaScript Puedes proporcionar este archivo ZIP como un artefacto componente que el software AWS IoT Greengrass principal descomprimirá al instalar el componente. Haga lo siguiente:
1. Abra la carpeta que contiene el repositorio que clonó o descargó en el paso anterior.
```
cd aws-greengrass-stream-manager-sdk-js
```
1. Comprima la carpeta `aws-greengrass-stream-manager-sdk` en un archivo ZIP llamado `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. Compruebe que el archivo `stream-manager-sdk.zip` contiene la carpeta `aws-greengrass-stream-manager-sdk` y su contenido. Ejecute el siguiente comando para hacer una lista del contenido del archivo ZIP.
------
#### [ Linux or Unix ]
```
unzip -l stream-manager-sdk.zip
```
------
#### [ Windows Command Prompt (CMD) ]
```
tar -tf stream-manager-sdk.zip
```
------
El resultado de debería parecerse al siguiente.
```
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 el artefacto del SDK del administrador de flujos a la carpeta de artefactos de su componente. *\$1/greengrass-components*Sustitúyalo por la ruta a la carpeta que utilizas para el desarrollo 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. Cree su receta de componentes. En la receta, haga lo siguiente:
1. Defina `stream-manager-sdk.zip` como un artefacto.
1. Defina su JavaScript aplicación como un artefacto.
1. En el ciclo de vida de la instalación, instale el SDK del administrador de flujos desde el artefacto `stream-manager-sdk.zip`. Este comando `npm install` crea una carpeta `node_modules` que contiene el SDK del administrador de flujos y sus dependencias.
1. En el ciclo de vida de ejecución, añada la `node_modules` carpeta a `NODE_PATH` la aplicación y JavaScript ejecútela.
La receta de su componente podría parecerse al siguiente ejemplo. Este componente ejecuta el ejemplo de [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 obtener más información sobre cómo desarrollar y probar componentes, consulte [Creación de componentes de AWS IoT Greengrass](create-components.md).
## Conexión al administrador de flujos en el código de la aplicación
Para conectarse al administrador de flujos de su aplicación, cree una instancia `StreamManagerClient` desde el SDK del administrador de flujos. Este cliente se conecta al componente del administrador de flujos en su puerto predeterminado 8088 o en el puerto que especifique. Para obtener más información sobre cómo usar `StreamManagerClient` después de crear una instancia, consulte [Se usa StreamManagerClient para trabajar con transmisiones](work-with-streams.md).
**Example Ejemplo: conectarse al administrador de flujos con el puerto predeterminado**
```
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 Ejemplo: conectarse al administrador de flujos con un puerto no predeterminado**
Si configura el administrador de flujos con un puerto distinto del predeterminado, debe utilizar la [comunicación entre procesos](interprocess-communication.md) para recuperar el puerto de la configuración del componente.
El parámetro `port` de configuración contiene el valor que se especifica en `STREAM_MANAGER_SERVER_PORT` al implementar el administrador de flujos.
```
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.
```
# Se usa StreamManagerClient para trabajar con transmisiones
Los componentes de Greengrass definidos por el usuario que se ejecutan en el dispositivo principal de Greengrass pueden utilizar el objeto `StreamManagerClient` en el SDK del Administrador de flujos para crear flujos en el [administrador de flujos](manage-data-streams.md) y luego interactuar con ellos. Cuando un componente crea una transmisión, define los Nube de AWS destinos, la priorización y otras políticas de exportación y retención de datos para la transmisión. Para enviar datos al administrador de flujos, los componentes anexan los datos al flujo. Si se define un destino de exportación para el flujo, el administrador de flujos exporta la secuencia automáticamente.
**nota**
Normalmente, los clientes del administrador de flujos son componentes de Greengrass definidos por el usuario. Si su caso de negocio lo requiere, puede permitir que los procesos que no sean componentes que se ejecutan en el núcleo de Greengrass (por ejemplo, un contenedor de Docker) interactúen con el administrador de flujos. Para obtener más información, consulte [Autenticación del cliente](manage-data-streams.md#stream-manager-security-client-authentication).
Los fragmentos de este tema muestran cómo los clientes llaman a los métodos de `StreamManagerClient` para trabajar con secuencias. Para obtener detalles sobre la implementación de los métodos y sus argumentos, utilice los vínculos a la referencia del SDK de cada fragmento.
Si utiliza el administrador de flujos en una función de Lambda, la función de Lambda debe iniciar `StreamManagerClient` fuera del controlador de funciones. Si se crea una instancia en el controlador, la función crea un `client` y una conexión al administrador de secuencias cada vez que se invoca.
**nota**
Si crea una instancia `StreamManagerClient` en el controlador, debe llamar explícitamente al método `close()` cuando el `client` complete su trabajo. De lo contrario, el `client` mantiene la conexión abierta y otro subproceso en ejecución hasta que salga del script.
`StreamManagerClient` admite las siguientes operaciones:
+ [Creación de una secuencia de mensajes](#streammanagerclient-create-message-stream)
+ [Agregar un mensaje](#streammanagerclient-append-message)
+ [Lectura de mensajes](#streammanagerclient-read-messages)
+ [Lista de secuencias](#streammanagerclient-list-streams)
+ [Descripción de una secuencia de mensajes](#streammanagerclient-describe-message-stream)
+ [Actualización de un flujo de mensajes](#streammanagerclient-update-message-stream)
+ [Eliminación de una secuencia de mensajes](#streammanagerclient-delete-message-stream)
## Creación de una secuencia de mensajes
Para crear un flujo, un componente de Greengrass definido por el usuario llama al método de creación y pasa un objeto `MessageStreamDefinition`. Este objeto especifica el nombre exclusivo del flujo y define cómo el administrador del flujo debe gestionar los nuevos datos cuando se alcanza el tamaño máximo de flujo. Puede utilizar `MessageStreamDefinition` y sus tipos de datos (como `ExportDefinition`, `StrategyOnFull` y `Persistence`) para definir otras propiedades de secuencias. Entre ellos se incluyen:
+ El destino AWS IoT Analytics, Kinesis Data Streams AWS IoT SiteWise y Amazon S3 para las exportaciones automáticas. Para obtener más información, consulte [Exportación de configuraciones para Nube de AWS destinos compatibles](stream-export-configurations.md).
+ Prioridad de exportación. El administrador de secuencias exporta secuencias de mayor prioridad antes que secuencias de menor prioridad.
+ Tamaño e intervalo de lote máximos para AWS IoT Analytics Kinesis Data Streams AWS IoT SiteWise y destinos. El administrador de secuencias exporta mensajes cuando se cumple cualquiera de las condiciones.
+ Time-to-live (TTL). La cantidad de tiempo para garantizar que los datos de secuencia están disponibles para su procesamiento. Debe asegurarse de que los datos se pueden consumir dentro de este período de tiempo. Esta no es una política de eliminación. Es posible que los datos no se eliminen inmediatamente después del período TTL.
+ Persistencia de secuencia. Elija guardar las secuencias en el sistema de archivos para conservar los datos en los reinicios del núcleo o guardar las secuencias en la memoria.
+ Inicio del número de secuencia Especifique el número de secuencia del mensaje que se utilizará como mensaje de inicio en la exportación.
Para obtener más información acerca de `MessageStreamDefinition`, consulte la referencia del SDK para el lenguaje de destino:
+ [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html)en el SDK de Java
+ [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)en el SDK de Node.js
+ [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition)en el SDK de Python
**nota**
`StreamManagerClient` también proporciona un destino que puede utilizar para exportar secuencias a un servidor HTTP. Este destino está pensado solo con fines de prueba. No es estable y no se admite para su uso en entornos de producción.
Una vez creado un flujo, los componentes de Greengrass pueden [anexar mensajes](#streammanagerclient-append-message) al flujo con el fin de enviar datos para su exportación y [leer los mensajes](#streammanagerclient-read-messages) del flujo para su procesamiento local. El número de secuencias que cree depende de sus capacidades de hardware y de su caso de negocio. Una estrategia consiste en AWS IoT Analytics crear una transmisión para cada canal de destino de la transmisión de datos de Kinesis, aunque puede definir varios destinos para una transmisión. Una secuencia tiene una vida útil duradera.
### Requisitos
Esta operación tiene los siguientes requisitos:
+ Versión mínima del SDK del Administrados de flujos: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.0
### Ejemplos
El siguiente fragmento crea una secuencia denominada `StreamName`. Define las propiedades de las secuencias en `MessageStreamDefinition` y los tipos de datos 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 Nube de 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.
```
Referencia del SDK de Python: [create\$1message\$1stream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.create_message_stream) \$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition)
------
#### [ 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 Nube de AWS.
new ExportDefinition()
.withKinesis(null)
.withIotAnalytics(null)
.withIotSitewise(null)
.withS3(null)
)
);
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Referencia del [createMessageStream](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-)SDK de Java: \$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 {
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 Nube de 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.
});
```
Referencia del SDK de Node.js: [createMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#createMessageStream)\$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)
------
Para obtener más información acerca de la configuración de destinos de exportación, consulte [Exportación de configuraciones para Nube de AWS destinos compatibles](stream-export-configurations.md).
## Agregar un mensaje
Para enviar datos al administrador de flujos para su exportación, los componentes de Greengrass anexan los datos al flujo de destino. El destino de exportación determina el tipo de datos que se van a transferir a este método.
### Requisitos
Esta operación tiene los siguientes requisitos:
+ Versión mínima del SDK del Administrados de flujos: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.0
### Ejemplos
#### AWS IoT Analytics o destinos de exportación de Kinesis Data Streams
El siguiente fragmento agrega un mensaje a la secuencia denominada `StreamName`. Para los AWS IoT Analytics destinos de Kinesis Data Streams, sus componentes de Greengrass añaden una masa de datos.
Este fragmento tiene los siguientes requisitos:
+ Versión mínima del SDK del Administrados de flujos: 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.
```
Referencia del SDK de 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.
}
```
Referencia de SDK de 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.
});
```
Referencia del Node.js SDK: [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 exportación
El siguiente fragmento agrega un mensaje a la secuencia denominada `StreamName`. Para AWS IoT SiteWise los destinos, sus componentes de Greengrass añaden un objeto serializado. `PutAssetPropertyValueEntry` Para obtener más información, consulte [Exportación a AWS IoT SiteWise](stream-export-configurations.md#export-streams-to-sitewise).
**nota**
Al enviar datos a AWS IoT SiteWise, estos deben cumplir los requisitos de la acción. `BatchPutAssetPropertyValue` Para obtener más información, consulta [BatchPutAssetPropertyValue](https://docs.aws.amazon.com/iot-sitewise/latest/APIReference/API_BatchPutAssetPropertyValue.html) en la *AWS IoT SiteWise Referencia de la API de *.
Este fragmento tiene los siguientes requisitos:
+ Versión mínima del SDK del Administrados de flujos: 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.
```
Referencia del SDK de 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.
}
```
Referencia del SDK de 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.
});
```
Referencia del SDK de 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 exportación de Amazon S3
El siguiente fragmento añade una tarea de exportación a la secuencia denominada `StreamName`. Para los destinos de Amazon S3, los componentes de Greengrass anexan un objeto `S3ExportTaskDefinition` serializado que contiene información sobre el archivo de entrada de origen y el objeto de Amazon S3 de destino. Si el objeto especificado no existe, el administrador de flujos lo crea por usted. Para obtener más información, consulte [Exportación a Amazon S3](stream-export-configurations.md#export-streams-to-s3).
Este fragmento tiene los siguientes requisitos:
+ Versión mínima del SDK del Administrados de flujos: 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.
```
[Referencia del SDK de Python: [append\$1message \$1 S3](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.append_message) ExportTaskDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.S3ExportTaskDefinition)
------
#### [ 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.
}
```
[Referencia del SDK de 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.
});
```
[Referencia del SDK de 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)
------
## Lectura de mensajes
Leer mensajes de un flujo.
### Requisitos
Esta operación tiene los siguientes requisitos:
+ Versión mínima del SDK del Administrados de flujos: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.0
### Ejemplos
El siguiente fragmento lee los mensajes de la secuencia denominada `StreamName`. El método read toma un objeto `ReadMessagesOptions` opcional que especifica el número de secuencia para comenzar a leer, los números mínimo y máximo para leer y un tiempo de espera para leer mensajes.
------
#### [ 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.
```
Referencia del SDK de 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.
}
```
Referencia del SDK de 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.
});
```
Referencia del SDK de 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)
------
## Lista de secuencias
Obtenga la lista de flujos en el administrador de flujos.
### Requisitos
Esta operación tiene los siguientes requisitos:
+ Versión mínima del SDK del Administrados de flujos: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.0
### Ejemplos
El siguiente fragmento de código obtiene una lista de las secuencias (por nombre) del administrador de secuencias.
------
#### [ Python ]
```
client = StreamManagerClient()
try:
stream_names = client.list_streams()
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
```
Referencia del SDK de 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.
}
```
Referencia de SDK de 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.
});
```
Referencia del Node.js SDK: [listStreams](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#listStreams)
------
## Descripción de una secuencia de mensajes
Obtenga metadatos sobre un flujo, incluida la definición, el tamaño y el estado de exportación del flujo.
### Requisitos
Esta operación tiene los siguientes requisitos:
+ Versión mínima del SDK del Administrados de flujos: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.0
### Ejemplos
El siguiente fragmento de código obtiene metadatos de una secuencia llamada `StreamName`, como su definición, su tamaño y los estados del exportador.
------
#### [ 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.
```
Referencia del SDK de 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.
}
```
Referencia del SDK de Java: [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.
});
```
Referencia del SDK de Node.js: [describeMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#describeMessageStream)
------
## Actualización de un flujo de mensajes
Actualiza las propiedades de un flujo existente. Es posible que desee actualizar un flujo si sus requisitos cambian después de crearlo. Por ejemplo:
+ Agrega una nueva [configuración de exportación](stream-export-configurations.md) para un Nube de AWS destino.
+ Aumente el tamaño máximo de un flujo para cambiar la forma en que se exportan o conservan los datos. Por ejemplo, el tamaño del flujo, en combinación con su estrategia en la configuración completa, puede provocar que los datos se eliminen o rechacen antes de que el administrador de flujos pueda procesarlos.
+ Pausa y reanuda las exportaciones; por ejemplo, si las tareas de exportación son prolongadas y quiere racionar los datos de carga.
Los componentes de Greengrass siguen este proceso general para actualizar un flujo:
1. [Consiga la descripción del flujo.](#streammanagerclient-describe-message-stream)
1. Actualice las propiedades de destino de los objetos correspondientes `MessageStreamDefinition` y subordinados.
1. Transfiera la actualización `MessageStreamDefinition`. Asegúrese de incluir las definiciones de objetos completas para el flujo actualizado. Las propiedades no definidas revierten a los valores predeterminados.
Puede especificar el número de secuencia del mensaje que se utilizará como mensaje de inicio en la exportación.
### Requisitos
Esta operación tiene los siguientes requisitos:
+ Versión mínima del SDK del Administrados de flujos: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.0
### Ejemplos
El siguiente fragmento de código actualiza la secuencia llamada `StreamName`. Actualiza varias propiedades de un flujo que se exporta a 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.
```
Referencia del SDK de Python: [updateMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.update_message_stream)\$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition)
------
#### [ 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 Nube de 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.
}
```
Referencia del SDK de 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 Nube de 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.
});
```
Referencia del SDK de 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)
------
### Restricciones para actualizar los flujos
Se aplican las siguientes restricciones al actualizar flujos. A menos que se indique en la siguiente lista, las actualizaciones se aplican de inmediato.
+ No puede actualizar la persistencia de un flujo. Para cambiar este comportamiento, [elimine el flujo](#streammanagerclient-delete-message-stream) y [crea un flujo](#streammanagerclient-create-message-stream) que defina la nueva política de persistencia.
+ Puede actualizar el tamaño máximo de una transmisión solo en las siguientes condiciones:
+ El tamaño máximo debe ser superior o igual al tamaño actual del flujo. Para encontrar esta información, [describa la secuencia](#streammanagerclient-describe-message-stream) y, a continuación, compruebe el estado de almacenamiento del objeto `MessageStreamInfo` devuelto.
+ El tamaño máximo debe ser superior o igual al tamaño del segmento del flujo.
+ Puede actualizar el tamaño del segmento de la transmisión a un valor inferior al tamaño máximo del flujo. La configuración actualizada se aplica a los segmentos nuevos.
+ Las actualizaciones de la propiedad tiempo de vida (TTL) se aplican a las nuevas operaciones de anexión. Si reduce este valor, es posible que el administrador de flujos también elimine los segmentos existentes que superen el TTL.
+ Las actualizaciones de la estrategia en toda la propiedad se aplican a las nuevas operaciones de anexión. Si establece la estrategia para sobrescribir los flujos de datos más antiguos, es posible que el administrador de flujos también sobrescriba los segmentos existentes en función del nuevo ajuste.
+ Las actualizaciones de la propiedad de vaciar al escribir se aplican a los mensajes nuevos.
+ Las actualizaciones de las configuraciones de exportación se aplican a las nuevas exportaciones. La solicitud de actualización debe incluir todas las configuraciones de exportación que desee admitir. De lo contrario, el administrador de flujos las elimina.
+ Al actualizar una configuración de exportación, especifique el identificador de la configuración de exportación de destino.
+ Para añadir una configuración de exportación, especifique un identificador único para la nueva configuración de exportación.
+ Para eliminar una configuración de exportación, omita la configuración de exportación.
+ Para [actualizar](#streammanagerclient-update-message-stream) el número de secuencia inicial de una configuración de exportación en un flujo, debe especificar un valor inferior al último número de secuencia. Para encontrar esta información, [describa la secuencia](#streammanagerclient-describe-message-stream) y, a continuación, compruebe el estado de almacenamiento del objeto `MessageStreamInfo` devuelto.
## Eliminación de una secuencia de mensajes
Elimina un flujo. Cuando elimina una secuencia, todos los datos almacenados para la secuencia se eliminan del disco.
### Requisitos
Esta operación tiene los siguientes requisitos:
+ Versión mínima del SDK del Administrados de flujos: Python: 1.1.0 \$1 Java: 1.1.0 \$1 Node.js: 1.1.0
### Ejemplos
El siguiente fragmento de código elimina la secuencia llamada `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.
```
Referencia del SDK de 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.
}
```
Referencia del SDK de 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.
});
```
Referencia del SDK de Node.js: [deleteMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#deleteMessageStream)
------
## Véase también
+ [Administración de flujos de datos en los dispositivos principales de Greengrass](manage-data-streams.md)
+ [Configurar el administrador de secuencias de AWS IoT Greengrass](configure-stream-manager.md)
+ [Exportación de configuraciones para Nube de AWS destinos compatibles](stream-export-configurations.md)
+ `StreamManagerClient` en la referencia del SDK del Administrador de flujos:
+ [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)
# Exportación de configuraciones para Nube de AWS destinos compatibles
Los componentes de Greengrass definidos por el usuario utilizan `StreamManagerClient` en el SDK del administrador de flujos para interactuar con el administrador de flujos. Cuando un componente [crea un flujo](work-with-streams.md#streammanagerclient-create-message-stream) o [actualiza un flujo](work-with-streams.md#streammanagerclient-create-message-stream), pasa un objeto `MessageStreamDefinition` que representa las propiedades del flujo, incluida la definición de exportación. El objeto `ExportDefinition` contiene las configuraciones de exportación definidas para el flujo. El administrador de flujos utiliza estas configuraciones de exportación para determinar dónde y cómo exportar el flujo.
![\[Diagrama del modelo de objetos del tipo de ExportDefinition propiedad.\]](http://docs.aws.amazon.com/es_es/greengrass/v2/developerguide/images/stream-manager-exportconfigs.png)
Puede definir cero o más configuraciones de exportación en un flujo, incluidas varias configuraciones de exportación para un único tipo de destino. Por ejemplo, puede exportar un flujo a dos canales AWS IoT Analytics y a un flujo de datos de Kinesis.
En caso de intentos de exportación fallidos, Stream Manager vuelve a intentar exportar los datos continuamente a Nube de AWS intervalos de hasta cinco minutos. La cantidad de reintentos no tiene un límite máximo.
**nota**
`StreamManagerClient` también proporciona un destino que puede utilizar para exportar secuencias a un servidor HTTP. Este destino está pensado solo con fines de prueba. No es estable y no se admite para su uso en entornos de producción.
**Topics**
+ [AWS IoT Analytics canales](#export-to-iot-analytics)
+ [Amazon Kinesis Data Streams](#export-to-kinesis)
+ [AWS IoT SiteWise propiedades de los activos](#export-to-iot-sitewise)
+ [Objetos de Amazon S3](#export-to-s3)
Usted es responsable del mantenimiento de estos Nube de AWS recursos.
## AWS IoT Analytics canales
El administrador de transmisiones admite exportaciones automáticas a AWS IoT Analytics. AWS IoT Analytics le permite realizar análisis avanzados de sus datos para ayudarle a tomar decisiones empresariales y mejorar los modelos de aprendizaje automático. Para obtener más información, consulte [¿Qué es AWS IoT Analytics?](https://docs.aws.amazon.com/iotanalytics/latest/userguide/welcome.html) en la *Guía AWS IoT Analytics del usuario*.
En el SDK del administrador de flujos, los componentes de Greengrass utilizan `IoTAnalyticsConfig` para definir la configuración de exportación para este tipo de destino. Para obtener más información, consulte la referencia del SDK para el lenguaje de destino.
+ [Io TAnalytics Config](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.IoTAnalyticsConfig) en el SDK de Python
+ [Io TAnalytics Config](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTAnalyticsConfig.html) en el SDK de Java
+ [Io TAnalytics Config](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTAnalyticsConfig.html) en el SDK de Node.js
### Requisitos
Este destino de exportación tiene los siguientes requisitos:
+ Los canales de destino AWS IoT Analytics deben estar en el mismo Cuenta de AWS y Región de AWS igual que el dispositivo principal de Greengrass.
+ El [Autorizar a los dispositivos principales a interactuar con AWS los servicios](device-service-role.md) debe permitir el permiso `iotanalytics:BatchPutMessage` para segmentar los canales. Por ejemplo:
------
#### [ 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"
]
}
]
}
```
------
Puede conceder acceso granular o condicional a recursos (por ejemplo, utilizando un esquema de nomenclatura con comodín `*`) Para obtener más información, consulte [Adición y eliminación de políticas de IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) en la *Guía del usuario de IAM*.
### Exportando a AWS IoT Analytics
Para crear una transmisión a la que se exporte AWS IoT Analytics, sus componentes de Greengrass [crean una transmisión](work-with-streams.md#streammanagerclient-create-message-stream) con una definición de exportación que incluye uno o más `IoTAnalyticsConfig` objetos. Este objeto define los ajustes de exportación, como el canal de destino, el tamaño del lote, el intervalo del lote y la prioridad.
Cuando sus componentes de Greengrass reciben datos de los dispositivos, [anexan mensajes](work-with-streams.md#streammanagerclient-append-message) que contienen una masa de datos al flujo de destino.
A continuación, el administrador de flujos exporta los datos en función de los ajustes del lote y la prioridad definidos en las configuraciones de exportación del flujo.
## Amazon Kinesis Data Streams
El administrador de flujos permite exportar automáticamente a Amazon Kinesis Data Streams. Kinesis Data Streams se utiliza normalmente para agregar grandes volúmenes de datos y cargarlos en un almacén MapReduce de datos o un clúster. Para obtener más información, consulte [Qué son los Amazon Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/what-is-this-service.html) en la *Guía para desarrolladores de Amazon Kinesis*.
En el SDK del administrador de flujos, los componentes de Greengrass utilizan `KinesisConfig` para definir la configuración de exportación para este tipo de destino. Para obtener más información, consulte la referencia del SDK para el lenguaje de destino.
+ [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.KinesisConfig)en el SDK de Python
+ [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/KinesisConfig.html)en el SDK de Java
+ [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.KinesisConfig.html)en el SDK de Node.js
### Requisitos
Este destino de exportación tiene los siguientes requisitos:
+ Las transmisiones de destino de Kinesis Data Streams deben estar en el Cuenta de AWS mismo dispositivo principal de Greengrass Región de AWS y en el mismo.
+ (Recomendado) La versión 2.2.1 del administrador de flujos mejora el rendimiento de la exportación de flujos a los destinos del flujo de datos de Kinesis. Para usar las mejoras de esta última versión, actualice su [componente de administrador de flujos](stream-manager-component.md) a la versión 2.2.1 y utilice la política `kinesis:ListShards` en el [rol de intercambio de tokens de Greengrass](device-service-role.md).
+ El [Autorizar a los dispositivos principales a interactuar con AWS los servicios](device-service-role.md) debe conceder el permiso `kinesis:PutRecords` para destinar flujos de datos. Por ejemplo:
------
#### [ 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"
]
}
]
}
```
------
Puede conceder acceso granular o condicional a recursos (por ejemplo, utilizando un esquema de nomenclatura con comodín `*`) Para obtener más información, consulte [Adición y eliminación de políticas de IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) en la *Guía del usuario de IAM*.
### Exportación a Kinesis Data Streams
Para crear un flujo que se exporte a Kinesis Data Streams, sus componentes de Greengrass [crean un flujo](work-with-streams.md#streammanagerclient-create-message-stream) con una definición de exportación que incluye uno o más objetos `KinesisConfig`. Este objeto define los ajustes de exportación, como el flujo de datos de destino, el tamaño del lote, el intervalo del lote y la prioridad.
Cuando sus componentes de Greengrass reciben datos de los dispositivos, [anexan mensajes](work-with-streams.md#streammanagerclient-append-message) que contienen una masa de datos al flujo de destino. A continuación, el administrador de flujos exporta los datos en función de los ajustes del lote y la prioridad definidos en las configuraciones de exportación del flujo.
El administrador de flujos genera un UUID aleatorio único como clave de partición para cada registro cargado en Amazon Kinesis.
## AWS IoT SiteWise propiedades de los activos
Stream Manager admite la exportación automática a AWS IoT SiteWise. AWS IoT SiteWise le permite recopilar, organizar y analizar datos de equipos industriales a escala. Para obtener más información, consulte [¿Qué es AWS IoT SiteWise?](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/what-is-sitewise.html) en la *Guía AWS IoT SiteWise del usuario*.
En el SDK del administrador de flujos, los componentes de Greengrass utilizan `IoTSiteWiseConfig` para definir la configuración de exportación para este tipo de destino. Para obtener más información, consulte la referencia del SDK para el lenguaje de destino.
+ [Io TSite WiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.IoTSiteWiseConfig) en el SDK de Python
+ [Io TSite WiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTSiteWiseConfig.html) en el SDK de Java
+ [Io TSite WiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTSiteWiseConfig.html) en el SDK de Node.js
**nota**
AWS también proporciona AWS IoT SiteWise componentes, que ofrecen una solución prediseñada que puede utilizar para transmitir datos desde fuentes OPC-UA. Para obtener más información, consulte [Colector IoT SiteWise OPC UA](iotsitewise-opcua-collector-component.md).
### Requisitos
Este destino de exportación tiene los siguientes requisitos:
+ Las propiedades de los activos de destino AWS IoT SiteWise deben estar en el mismo Cuenta de AWS y Región de AWS igual que el dispositivo principal de Greengrass.
**nota**
Para ver la lista de Región de AWS dispositivos AWS IoT SiteWise compatibles, consulte los [AWS IoT SiteWise puntos finales y las cuotas](https://docs.aws.amazon.com/general/latest/gr/iot-sitewise.html#iot-sitewise_region) en la Referencia *AWS general*.
+ [Autorizar a los dispositivos principales a interactuar con AWS los servicios](device-service-role.md)El `iotsitewise:BatchPutAssetPropertyValue` debe permitir que el permiso se dirija a las propiedades de los activos. El siguiente ejemplo de política utiliza la clave `iotsitewise:assetHierarchyPath` de condición para conceder acceso a un activo raíz de destino y a sus elementos secundarios. Puede eliminarlos `Condition` de la política para permitir el acceso a todos sus AWS IoT SiteWise activos o especificar ARNs activos individuales.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "iotsitewise:BatchPutAssetPropertyValue",
"Resource": "*",
"Condition": {
"StringLike": {
"iotsitewise:assetHierarchyPath": [
"/root node asset ID",
"/root node asset ID/*"
]
}
}
}
]
}
```
------
Puede conceder acceso granular o condicional a recursos (por ejemplo, utilizando un esquema de nomenclatura con comodín `*`) Para obtener más información, consulte [Adición y eliminación de políticas de IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) en la *Guía del usuario de IAM*.
Para obtener información de seguridad importante, consulte la [ BatchPutAssetPropertyValue autorización](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) en la *Guía del AWS IoT SiteWise usuario*.
### Exportación a AWS IoT SiteWise
Para crear una transmisión a la que se exporte AWS IoT SiteWise, sus componentes de Greengrass [crean una transmisión](work-with-streams.md#streammanagerclient-create-message-stream) con una definición de exportación que incluye uno o más `IoTSiteWiseConfig` objetos. Este objeto define los ajustes de exportación, como el tamaño del lote, el intervalo del lote y la prioridad.
Cuando sus componentes de Greengrass reciben datos de propiedad de activos, anexan mensajes que contienen datos al flujo de destino. Los mensajes son objetos `PutAssetPropertyValueEntry` serializados en JSON que contienen los valores de propiedad de una o más propiedades de los activos. Para obtener más información, consulte [Añadir un mensaje](work-with-streams.md#streammanagerclient-append-message-sitewise) para los AWS IoT SiteWise destinos de exportación.
**nota**
Al enviar datos a AWS IoT SiteWise, estos deben cumplir los requisitos de la `BatchPutAssetPropertyValue` acción. Para obtener más información, consulta [BatchPutAssetPropertyValue](https://docs.aws.amazon.com/iot-sitewise/latest/APIReference/API_BatchPutAssetPropertyValue.html) en la *AWS IoT SiteWise Referencia de la API de *.
A continuación, el administrador de flujos exporta los datos en función de los ajustes del lote y la prioridad definidos en las configuraciones de exportación del flujo.
Puede ajustar la configuración del administrador de flujos y la lógica del componente de Greengrass para diseñar su estrategia de exportación. Por ejemplo:
+ Para realizar exportaciones prácticamente en tiempo real, establezca ajustes del tamaño de lote e intervalos bajos y añada los datos al flujo cuando lo reciba.
+ Para optimizar el procesamiento por lotes, mitigar las restricciones de ancho de banda o minimizar los costos, sus componentes de Greengrass pueden agrupar timestamp-quality-value los puntos de datos (TQV) recibidos para una sola propiedad de activo antes de agregar los datos a la transmisión. Una estrategia consiste en agrupar las entradas de hasta 10 combinaciones diferentes de propiedades y activos (o alias de propiedades) en un mensaje, en lugar de enviar más de una entrada para la misma propiedad. Esto ayuda al Administrador de flujos a mantenerse dentro de las [cuotas de AWS IoT SiteWise](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/quotas.html).
## Objetos de Amazon S3
El administrador de flujos permite exportar automáticamente a Amazon S3. Puede utilizar Amazon S3 para almacenar y recuperar grandes cantidades de datos. Para obtener más información, consulte [¿Qué es Amazon S3?](https://docs.aws.amazon.com/AmazonS3/latest/dev/Welcome.html) en la *Guía para desarrolladores de Amazon Simple Storage Service*.
En el SDK del administrador de flujos, los componentes de Greengrass utilizan `S3ExportTaskExecutorConfig` para definir la configuración de exportación para este tipo de destino. Para obtener más información, consulte la referencia del SDK para el lenguaje de destino.
+ [S3 ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.S3ExportTaskExecutorConfig) en el SDK de Python
+ [S3 ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/S3ExportTaskExecutorConfig.html) en el SDK de Java
+ [S3 ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.S3ExportTaskExecutorConfig.html) en el SDK de Node.js
### Requisitos
Este destino de exportación tiene los siguientes requisitos:
+ Los buckets Amazon S3 de destino deben estar en el mismo lugar Cuenta de AWS que el dispositivo principal de Greengrass.
+ Si una función de Lambda que se ejecuta en el modo **contenedor de Greengrass** escribe archivos de entrada en el directorio de archivos de entrada, usted debe crear el directorio como un volumen en el contenedor con permisos de escritura. Esto garantiza que los archivos se escriban en el sistema de archivos raíz y que sean visibles en el componente administrador de flujos, que se ejecuta fuera del contenedor.
+ Si un componente contenedor de Docker escribe archivos de entrada en un directorio de archivos de entrada, usted debe montar el directorio como un volumen en el contenedor con permisos de escritura. Esto garantiza que los archivos se escriban en el sistema de archivos raíz y que sean visibles en el componente administrador de flujos, que se ejecuta fuera del contenedor.
+ El [Autorizar a los dispositivos principales a interactuar con AWS los servicios](device-service-role.md) debe permitir los siguientes permisos para los buckets de destino. Por ejemplo:
------
#### [ 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/*"
]
}
]
}
```
------
Puede conceder acceso granular o condicional a recursos (por ejemplo, utilizando un esquema de nomenclatura con comodín `*`) Para obtener más información, consulte [Adición y eliminación de políticas de IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) en la *Guía del usuario de IAM*.
### Exportación a Amazon S3
Para crear un flujo que se exporte a Amazon S3, sus componentes de Greengrass utilizan el objeto `S3ExportTaskExecutorConfig` para configurar la política de exportación. La política define los ajustes de exportación, como el umbral de carga multiparte y la prioridad. Para las exportaciones de Amazon S3, el administrador de flujos carga los datos que lee de los archivos locales en el dispositivo principal. Para iniciar una carga, sus componentes de Greengrass anexan una tarea de exportación al flujo de destino. La tarea de exportación contiene información sobre el archivo de entrada y el objeto Amazon S3 de destino. El administrador de flujos ejecuta las tareas en la secuencia en que se anexan al flujo.
**nota**
El depósito de destino ya debe existir en su. Cuenta de AWS Si no existe un objeto para la clave especificada, el administrador de flujos crea el objeto automáticamente.
El administrador de flujos utiliza la propiedad de umbral de carga multiparte, el ajuste del [tamaño mínimo de las piezas](configure-stream-manager.md#stream-manager-minimum-part-size) y el tamaño del archivo de entrada para determinar cómo cargar los datos. El umbral de carga multiparte debe ser igual o mayor que el tamaño mínimo de la pieza. Si desea cargar datos en paralelo, puede crear varios flujos.
Las claves que especifican los objetos de Amazon S3 de destino pueden incluir DateTimeFormatter cadenas [Java](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html) válidas en `!{timestamp:value}` los marcadores de posición. Puede utilizar estos marcadores de fecha y hora para particionar los datos en Amazon S3 en función de la hora en la que se cargaron los datos del archivo de entrada. Por ejemplo, el siguiente nombre de clave se resuelve en un valor como `my-key/2020/12/31/data.txt`.
```
my-key/!{timestamp:YYYY}/!{timestamp:MM}/!{timestamp:dd}/data.txt
```
**nota**
Si desea supervisar el estado de exportación de un flujo, cree primero un flujo de estado y, a continuación, configure el flujo de exportación para utilizarlo. Para obtener más información, consulte [Supervise las tareas de exportación](#monitor-export-status-s3).
#### Administración de datos de entrada
Puede crear un código que las aplicaciones de IoT usen para administrar el ciclo de vida de los datos de entrada. El siguiente ejemplo de flujo de trabajo muestra cómo se pueden utilizar los componentes de Greengrass para administrar estos datos.
1. Un proceso local recibe datos de dispositivos o periféricos y, a continuación, los escribe en los archivos de un directorio del dispositivo principal. Estos son los archivos de entrada del administrador de flujos.
1. Un componente de Greengrass escanea el directorio y [anexa una tarea de exportación](work-with-streams.md#streammanagerclient-append-message-export-task) al flujo de destino cuando se crea un archivo nuevo. La tarea es un objeto `S3ExportTaskDefinition` serializado en JSON que especifica la URL del archivo de entrada, el bucket y la clave de Amazon S3 de destino y los metadatos de usuario opcionales.
1. El administrador de flujos lee el archivo de entrada y exporta los datos a Amazon S3 en el orden de las tareas anexas. El bucket de destino ya debe existir en su. Cuenta de AWS Si no existe un objeto para la clave especificada, el administrador de flujos crea el objeto automáticamente.
1. El componente de Greengrass [lee los mensajes](work-with-streams.md#streammanagerclient-read-messages) de un flujo de estado para supervisar el estado de la exportación. Una vez finalizadas las tareas de exportación, el componente de Greengrass puede eliminar los archivos de entrada correspondientes. Para obtener más información, consulte [Supervise las tareas de exportación](#monitor-export-status-s3).
### Supervise las tareas de exportación
Puede crear un código que las aplicaciones de IoT utilizan para monitorear el estado de sus exportaciones de Amazon S3. Sus componentes de Greengrass deben crear un flujo de estado y, a continuación, configurar el flujo de exportación para escribir actualizaciones de estado en el flujo de estado. Una sola transmisión de estado puede recibir actualizaciones de estado de varias transmisiones que se exportan a Amazon S3.
En primer lugar, [cree un flujo](work-with-streams.md#streammanagerclient-create-message-stream) para utilizarlo como flujo de estado. Puede configurar las políticas de tamaño y retención del flujo para controlar la vida útil de los mensajes de estado. Por ejemplo:
+ Configure `Persistence` en `Memory` si no desea guardar los mensajes de estado.
+ Configure `StrategyOnFull` en `OverwriteOldestData` para que no se pierdan los nuevos mensajes de estado.
A continuación, cree o actualice el flujo de exportación para usar el flujo de estado. En concreto, defina la propiedad de configuración de estado de la configuración de `S3ExportTaskExecutorConfig` exportación del flujo. Esto le indica al administrador del flujo que escriba mensajes de estado sobre las tareas de exportación en el flujo de estado. En el objeto `StatusConfig`, especifique el nombre del flujo de estado y el nivel de detalle. Los siguientes valores admitidos van desde el menos detallado (`ERROR`) al más detallado (). `TRACE` El valor predeterminado es `INFO`.
+ `ERROR`
+ `WARN`
+ `INFO`
+ `DEBUG`
+ `TRACE`
El siguiente ejemplo de flujo de trabajo muestra cómo los componentes de Greengrass pueden utilizar un flujo de estado para supervisar el estado de exportación.
1. Como se describió en el flujo de trabajo anterior, un componente de Greengrass [anexa una tarea de exportación](work-with-streams.md#streammanagerclient-append-message-export-task) a un flujo que está configurado para escribir mensajes de estado sobre las tareas de exportación en un flujo de estado. La operación de incorporación devuelve un número de secuencia que representa el ID de la tarea.
1. Un componente de Greengrass [lee los mensajes](work-with-streams.md#streammanagerclient-read-messages) secuencialmente del flujo de estado y, a continuación, filtra los mensajes en función del nombre del flujo y el identificador de la tarea o en función de una propiedad de la tarea de exportación del contexto del mensaje. Por ejemplo, el componente de Greengrass puede filtrar por la URL del archivo de entrada de la tarea de exportación, que está representada por el objeto `S3ExportTaskDefinition` en el contexto del mensaje.
Los siguientes códigos de estado indican que una tarea de exportación ha alcanzado un estado completo:
+ `Success`. Se ha completado correctamente la carga.
+ `Failure`. El administrador de flujos detectó un error; por ejemplo, el bucket especificado no existe. Tras resolver el problema, puede volver a añadir la tarea de exportación al flujo.
+ `Canceled`. La tarea se detuvo porque se eliminó la definición de transmisión o exportación o porque el período time-to-live (TTL) de la tarea expiró.
**nota**
La tarea también puede tener un estado de `InProgress` o `Warning`. El administrador de flujos emite advertencias cuando un evento devuelve un error que no afecta a la ejecución de la tarea. Por ejemplo, si no se borra una carga parcial cancelada, aparecerá una advertencia.
1. Una vez finalizadas las tareas de exportación, el componente de Greengrass puede eliminar los archivos de entrada correspondientes.
En el siguiente ejemplo, se muestra cómo un componente de Greengrass puede leer y procesar los mensajes de estado.
------
#### [ 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.
```
Referencia del SDK de 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.
}
```
Referencia del SDK de 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.
});
```
Referencia del SDK de 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)
------