Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.
# Gérez les flux de données sur les appareils principaux de Greengrass
AWS IoT Greengrass le gestionnaire de flux permet de transférer de gros volumes de données IoT vers le AWS Cloud. Le gestionnaire de flux traite les flux de données sur le AWS IoT Greengrass Core avant de les exporter vers le AWS Cloud. Le gestionnaire de flux s'intègre aux scénarios périphériques courants, tels que l'inférence d'apprentissage automatique (ML), dans laquelle le périphérique AWS IoT Greengrass Core traite et analyse les données avant de les exporter vers des destinations de stockage locales AWS Cloud ou vers des destinations de stockage locales.
Le gestionnaire de flux fournit une interface commune pour simplifier le développement de composants personnalisés afin que vous n'ayez pas à créer de fonctionnalités de gestion de flux personnalisées. Vos composants peuvent utiliser un mécanisme standardisé pour traiter des flux importants et gérer les politiques locales de conservation des données. Vous pouvez définir des politiques relatives au type de stockage, à la taille et à la conservation des données pour chaque flux afin de contrôler la manière dont le gestionnaire de flux traite et exporte les données.
Le gestionnaire de flux fonctionne dans des environnements avec une connectivité intermittente ou limitée. Vous pouvez définir l'utilisation de la bande passante, le comportement du délai d'expiration et la manière dont le AWS IoT Greengrass Core gère les données de flux lorsqu'il est connecté ou déconnecté. Vous pouvez également définir des priorités pour contrôler l'ordre dans lequel le AWS IoT Greengrass Core exporte les flux vers le AWS Cloud. Cela vous permet de traiter les données critiques plus rapidement que les autres données.
Vous pouvez configurer le gestionnaire de flux pour qu'il exporte automatiquement les données vers le AWS Cloud pour les stocker ou les traiter et les analyser ultérieurement. Le gestionnaire de flux prend en charge les exportations vers les AWS Cloud destinations suivantes :
+ Canaux entrants AWS IoT Analytics. AWS IoT Analytics vous permet d'effectuer une analyse avancée de vos données afin de prendre des décisions commerciales et d'améliorer les modèles d'apprentissage automatique. Pour plus d’informations, consultez [Présentation de AWS IoT Analytics](https://docs.aws.amazon.com/iotanalytics/latest/userguide/welcome.html) dans le *Guide de l’utilisateur AWS IoT Analytics *.
+ Streams dans Amazon Kinesis Data Streams. Vous pouvez utiliser Kinesis Data Streams pour agréger de gros volumes de données et les charger dans un entrepôt MapReduce de données ou un cluster. Pour plus d'informations, consultez [Présentation d'Amazon Kinesis Data Streams](https://docs.aws.amazon.com/streams/latest/dev/what-is-this-service.html) dans le *Guide du développeur Amazon Kinesis Data Streams*.
+ Propriétés des actifs dans AWS IoT SiteWise. AWS IoT SiteWise vous permet de collecter, d'organiser et d'analyser les données des équipements industriels à grande échelle. Pour plus d'informations, voir [Qu'est-ce que c'est AWS IoT SiteWise ?](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/what-is-sitewise.html) dans le *guide de AWS IoT SiteWise l'utilisateur*.
+ Objets dans Amazon Simple Storage Service (Amazon S3). Vous pouvez utiliser Amazon S3 pour stocker et récupérer de grandes quantités de données. Pour plus d'informations, consultez [Qu'est-ce qu'Amazon S3 ?](https://docs.aws.amazon.com/AmazonS3/latest/dev/Welcome.html) dans le *guide du développeur Amazon Simple Storage Service*.
## Flux de travail de gestion des flux
Vos applications IoT interagissent avec le gestionnaire de flux via le SDK Stream Manager.
Dans un flux de travail simple, un composant AWS IoT Greengrass central consomme des données IoT, telles que des séries chronologiques de mesures de température et de pression. Le composant peut filtrer ou compresser les données, puis appeler le SDK Stream Manager pour écrire les données dans un flux dans le gestionnaire de flux. Le gestionnaire de flux peut exporter le flux AWS Cloud automatiquement en fonction des politiques que vous définissez pour le flux. Les composants peuvent également envoyer des données directement vers des bases de données locales ou des référentiels de stockage.
Vos applications IoT peuvent inclure plusieurs composants personnalisés qui lisent ou écrivent dans des flux. Ces composants peuvent lire et écrire dans des flux pour filtrer, agréger et analyser les données sur le périphérique AWS IoT Greengrass principal. Cela permet de réagir rapidement aux événements locaux et d'extraire des informations précieuses avant que les données ne soient transférées du centre vers les AWS Cloud destinations locales.
Pour commencer, déployez le composant Stream Manager sur votre appareil AWS IoT Greengrass principal. Lors du déploiement, configurez les paramètres du composant du gestionnaire de flux pour définir les paramètres qui s'appliquent à tous les flux sur le périphérique principal Greengrass. Utilisez ces paramètres pour contrôler la manière dont le gestionnaire de flux stocke, traite et exporte les flux en fonction des besoins de votre entreprise et des contraintes environnementales.
Après avoir configuré le gestionnaire de flux, vous pouvez créer et déployer vos applications IoT. Il s'agit généralement de composants personnalisés utilisés `StreamManagerClient` dans le SDK Stream Manager pour créer des flux et interagir avec ceux-ci. Lorsque vous créez un flux, vous pouvez définir des politiques par flux, telles que les destinations d'exportation, la priorité et la persistance.
## Exigences
Les exigences suivantes s'appliquent à l'utilisation du gestionnaire de flux :
+ Le gestionnaire de flux nécessite un minimum de 70 Mo de RAM en plus du logiciel AWS IoT Greengrass Core. Vos besoins totaux en mémoire dépendent de votre charge de travail.
+ AWS IoT Greengrass les composants doivent utiliser le SDK Stream Manager pour interagir avec le Stream Manager. Le SDK Stream Manager est disponible dans les langues suivantes :
+ [SDK Stream Manager pour](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-java/) Java (v1.1.0 ou version ultérieure)
+ [SDK Stream Manager pour Node.js](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-js/) (v1.1.0 ou version ultérieure)
+ [SDK Stream Manager pour Python](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-python/) (v1.1.0 ou version ultérieure)
+ AWS IoT Greengrass les composants doivent spécifier le composant du gestionnaire de flux (`aws.greengrass.StreamManager`) en tant que dépendance dans leur recette pour utiliser le gestionnaire de flux.
**Note**
Si vous utilisez le gestionnaire de flux pour exporter des données vers le cloud, vous ne pouvez pas mettre à niveau la version 2.0.7 du composant du gestionnaire de flux vers une version comprise entre v2.0.8 et v2.0.11. Si vous déployez le gestionnaire de flux pour la première fois, nous vous recommandons vivement de déployer la dernière version du composant du gestionnaire de flux.
+ Si vous définissez des destinations AWS Cloud d'exportation pour un flux, vous devez créer vos cibles d'exportation et accorder des autorisations d'accès dans le rôle d'[appareil Greengrass](device-service-role.md). Selon la destination, d'autres exigences peuvent également s'appliquer. Pour en savoir plus, consultez :
+ [AWS IoT Analytics chaînes](stream-export-configurations.md#export-to-iot-analytics)
+ [Flux de données Amazon Kinesis](stream-export-configurations.md#export-to-kinesis)
+ [AWS IoT SiteWise propriétés des actifs](stream-export-configurations.md#export-to-iot-sitewise)
+ [Objets Amazon S3](stream-export-configurations.md#export-to-s3)
Vous êtes responsable de la maintenance de ces AWS Cloud ressources.
## Sécurité des données
Lorsque vous utilisez le gestionnaire de flux, tenez compte des considérations de sécurité suivantes.
### Sécurité des données locales
AWS IoT Greengrass ne chiffre pas les données de flux au repos ou en transit entre les composants locaux du périphérique principal.
+ **Données au repos**. Les données de flux sont stockées localement dans un répertoire de stockage. Pour la sécurité des données, AWS IoT Greengrass repose sur les autorisations de fichiers et le chiffrement complet du disque, s'il est activé. Vous pouvez utiliser le paramètre facultatif [STREAM\$1MANAGER\$1STORE\$1ROOT\$1DIR](configure-stream-manager.md#STREAM_MANAGER_STORE_ROOT_DIR) pour spécifier le répertoire de stockage. Si vous modifiez ce paramètre ultérieurement pour utiliser un autre répertoire de stockage, AWS IoT Greengrass cela ne supprime pas le répertoire de stockage précédent ni son contenu.
+ **Données en transit localement**. AWS IoT Greengrass ne chiffre pas les données de flux en transit local entre les sources de données, AWS IoT Greengrass les composants, le SDK Stream Manager et le gestionnaire de flux.
+ **Données en transit vers le AWS Cloud**. Les flux de données exportés par le gestionnaire de flux AWS Cloud utilisent le chiffrement standard du client de AWS service avec le protocole TLS (Transport Layer Security).
### Authentification client
Les clients Stream Manager utilisent le SDK Stream Manager pour communiquer avec Stream Manager. Lorsque l'authentification du client est activée, seuls les composants Greengrass peuvent interagir avec les flux dans le gestionnaire de flux. Lorsque l'authentification du client est désactivée, tout processus exécuté sur le périphérique principal de Greengrass peut interagir avec les flux dans le gestionnaire de flux. Vous ne devez désactiver l'authentification que si votre analyse de rentabilisation l'exige.
Vous utilisez le paramètre [STREAM\$1MANAGER\$1AUTHENTICATE\$1CLIENT](configure-stream-manager.md#STREAM_MANAGER_AUTHENTICATE_CLIENT) pour définir le mode d'authentification du client. Vous pouvez configurer ce paramètre lorsque vous déployez le composant Stream Manager sur les appareils principaux.
****
| | Activé | Désactivé |
| --- | --- | --- |
| Valeur de paramètre | `true` (valeur par défaut et recommandée) | `false` |
| Clients autorisés | Composants Greengrass sur l'appareil principal | Composants Greengrass sur l'appareil principal Autres processus en cours d'exécution sur l'appareil principal Greengrass |
## Consultez aussi
+ [Configuration du gestionnaire de AWS IoT Greengrass flux](configure-stream-manager.md)
+ [StreamManagerClient À utiliser pour travailler avec des flux](work-with-streams.md)
+ [Exporter les configurations pour les AWS Cloud destinations prises en charge](stream-export-configurations.md)
# Configuration du gestionnaire de AWS IoT Greengrass flux
Sur les appareils principaux de Greengrass, le gestionnaire de flux peut stocker, traiter et exporter les données des appareils IoT. Le gestionnaire de flux fournit des paramètres que vous pouvez utiliser pour configurer les paramètres d'exécution. Ces paramètres s'appliquent à tous les streams sur l'appareil principal de Greengrass. Vous pouvez utiliser la AWS IoT Greengrass console ou l'API pour configurer les paramètres du gestionnaire de flux lorsque vous déployez le composant. Les modifications prennent effet une fois le déploiement terminé.
## Paramètres du gestionnaire de flux
Le gestionnaire de flux fournit les paramètres suivants que vous pouvez configurer lorsque vous déployez le composant sur vos appareils principaux. Tous les paramètres sont facultatifs.
**Répertoire de stockage**
Nom du paramètre: `STREAM_MANAGER_STORE_ROOT_DIR`
Le chemin absolu du dossier local utilisé pour stocker les flux. Cette valeur doit commencer par une barre oblique (par exemple, `/data`).
Vous devez spécifier un dossier existant, et l'[utilisateur du système qui exécute le composant du gestionnaire de flux](configure-greengrass-core-v2.md#configure-component-user) doit être autorisé à lire et à écrire dans ce dossier. Par exemple, vous pouvez exécuter les commandes suivantes pour créer et configurer un dossier`/var/greengrass/streams`, que vous spécifiez comme dossier racine du gestionnaire de flux. Ces commandes permettent à l'utilisateur du système par défaut de lire et d'écrire dans ce dossier. `ggc_user`
```
sudo mkdir /var/greengrass/streams
sudo chown ggc_user /var/greengrass/streams
sudo chmod 700 /var/greengrass/streams
```
Pour de plus amples informations sur la sécurité des données de flux, veuillez consulter [Sécurité des données locales](manage-data-streams.md#stream-manager-security-stream-data).
Par défaut : `/greengrass/v2/work/aws.greengrass.StreamManager`
**Server port**
Nom du paramètre: `STREAM_MANAGER_SERVER_PORT`
Numéro de port local utilisé pour communiquer avec le gestionnaire de flux. L’argument par défaut est `8088`.
Vous pouvez spécifier `0` d'utiliser un port disponible de manière aléatoire.
**Authentification du client**
Nom du paramètre: `STREAM_MANAGER_AUTHENTICATE_CLIENT`
Indique si les clients doivent être authentifiés pour interagir avec le gestionnaire de flux. Toutes les interactions entre les clients et le gestionnaire de flux sont contrôlées par le SDK Stream Manager. Ce paramètre détermine quels clients peuvent appeler le SDK Stream Manager pour travailler avec des flux. Pour de plus amples informations, veuillez consulter [Authentification client](manage-data-streams.md#stream-manager-security-client-authentication).
Les valeurs valides sont `true` ou `false`. La valeur par défaut est `true` (recommandé).
+ `true`. Autorise uniquement les composants Greengrass en tant que clients. Les composants utilisent des protocoles AWS IoT Greengrass Core internes pour s'authentifier auprès du SDK Stream Manager.
+ `false`. Permet à tout processus exécuté sur le AWS IoT Greengrass Core d'être un client. Ne définissez pas la valeur sur `false` sauf si votre analyse de rentabilisation l'exige. Par exemple, à utiliser `false` uniquement si les processus non composants du périphérique principal doivent communiquer directement avec le gestionnaire de flux.
**Bande passante maximum**
Nom du paramètre: `STREAM_MANAGER_EXPORTER_MAX_BANDWIDTH`
Bande passante maximale moyenne (en kilobits par seconde) pouvant être utilisée pour exporter des données. La valeur par défaut permet une utilisation illimitée de la bande passante disponible.
**Taille du groupe de threads**
Nom du paramètre: `STREAM_MANAGER_EXPORTER_THREAD_POOL_SIZE`
Nombre maximal de threads actifs pouvant être utilisés pour exporter des données. L’argument par défaut est `5`.
La taille optimale dépend de votre matériel, du volume de flux et du nombre planifié de flux d'exportation. Si votre vitesse d'exportation est faible, vous pouvez ajuster ce paramètre afin de trouver la taille optimale en fonction de votre matériel et de votre analyse de rentabilisation. Le processeur et la mémoire de votre appareil principal sont des facteurs limitatifs. Pour commencer, vous pouvez essayer de définir cette valeur par le nombre de cœurs de processeur sur l'appareil.
Veillez à ne pas définir une taille supérieure à ce que votre matériel peut prendre en charge. Chaque flux consomme des ressources matérielles. Essayez donc de limiter le nombre de flux d'exportation sur les appareils soumis à des contraintes.
**Arguments JVM**
Nom du paramètre: `JVM_ARGS`
Arguments JVM (machine virtuelle Java) personnalisés à transmettre au gestionnaire de flux au démarrage. Les arguments doivent être séparés par des espaces.
Utilisez ce paramètre uniquement lorsque vous devez remplacer les paramètres par défaut utilisés par la JVM. Par exemple, il peut s'avérer nécessaire d'augmenter la taille de pile par défaut si vous prévoyez d'exporter un grand nombre de flux.
**Logging level (Niveau de journalisation)**
Nom du paramètre: `LOG_LEVEL`
Le niveau de journalisation du composant. Choisissez parmi les niveaux de journalisation suivants, listés ici par ordre de niveau :
+ `TRACE`
+ `DEBUG`
+ `INFO`
+ `WARN`
+ `ERROR`
Par défaut : `INFO`
**Taille minimale pour le téléchargement en plusieurs parties**
Nom du paramètre: `STREAM_MANAGER_EXPORTER_S3_DESTINATION_MULTIPART_UPLOAD_MIN_PART_SIZE_BYTES`
Taille minimale (en octets) d'une partie dans un chargement partitionné vers Amazon S3. Le gestionnaire de flux utilise ce paramètre et la taille du fichier d'entrée pour déterminer comment regrouper les données dans une requête PUT en plusieurs parties. La valeur minimale par défaut est de 5 `5242880` octets (5 Mo).
Le gestionnaire de flux utilise la `sizeThresholdForMultipartUploadBytes` propriété du flux pour déterminer s'il convient d'exporter vers Amazon S3 sous forme de téléchargement en une ou plusieurs parties. Les composants Greengrass définis par l'utilisateur définissent ce seuil lorsqu'ils créent un flux exporté vers Amazon S3. Le seuil par défaut est de 5 Mo.
## Consultez aussi
+ [Gérez les flux de données sur les appareils principaux de Greengrass](manage-data-streams.md)
+ [StreamManagerClient À utiliser pour travailler avec des flux](work-with-streams.md)
+ [Exporter les configurations pour les AWS Cloud destinations prises en charge](stream-export-configurations.md)
# Créez des composants personnalisés utilisant le gestionnaire de flux
Utilisez le gestionnaire de flux dans les composants Greengrass personnalisés pour stocker, traiter et exporter les données des appareils IoT. Utilisez les procédures et les exemples de cette section pour créer des recettes de composants, des artefacts et des applications compatibles avec le gestionnaire de flux. Pour plus d'informations sur le développement et le test de composants, consultez[Création de AWS IoT Greengrass composants](create-components.md).
**Topics**
+ [Définissez des recettes de composants utilisant le gestionnaire de flux](#stream-manager-recipes)
+ [Connectez-vous au gestionnaire de flux dans le code de l'application](#connect-to-stream-manager)
## Définissez des recettes de composants utilisant le gestionnaire de flux
Pour utiliser le gestionnaire de flux dans un composant personnalisé, vous devez définir le `aws.greengrass.StreamManager` composant en tant que dépendance. Vous devez également fournir le SDK Stream Manager. Effectuez les tâches suivantes pour télécharger et utiliser le SDK Stream Manager dans la langue de votre choix.
### Utiliser le SDK Stream Manager pour Java
Le SDK Stream Manager pour Java est disponible sous forme de fichier JAR que vous pouvez utiliser pour compiler votre composant. Vous pouvez ensuite créer un fichier JAR d'application qui inclut le SDK Stream Manager, définir le fichier JAR d'application en tant qu'artefact de composant et exécuter le fichier JAR d'application dans le cycle de vie du composant.
**Pour utiliser le SDK Stream Manager pour Java**
1. Téléchargez le fichier [JAR du SDK Stream Manager pour Java](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-java/blob/main/sdk/aws-greengrass-stream-manager-sdk-java.jar).
1. Procédez de l'une des manières suivantes pour créer des artefacts de composants à partir de votre application Java et du fichier JAR du SDK Stream Manager :
+ Créez votre application sous la forme d'un fichier JAR incluant le JAR du SDK Stream Manager, et exécutez ce fichier JAR dans votre recette de composant.
+ Définissez le JAR du SDK Stream Manager en tant qu'artefact de composant. Ajoutez cet artefact au chemin de classe lorsque vous exécutez votre application dans votre recette de composant.
La recette de vos composants peut ressembler à l'exemple suivant. Ce composant exécute une version modifiée de l'exemple [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), qui `StreamManagerS3.jar` inclut le JAR du SDK 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
```
------
Pour plus d'informations sur le développement et le test de composants, consultez[Création de AWS IoT Greengrass composants](create-components.md).
### Utiliser le SDK Stream Manager pour Python
Le SDK Stream Manager pour Python est disponible sous forme de code source que vous pouvez inclure dans votre composant. Créez un fichier ZIP du SDK Stream Manager, définissez le fichier ZIP en tant qu'artefact de composant et installez les exigences du SDK dans le cycle de vie du composant.
**Pour utiliser le SDK Stream Manager pour Python**
1. Clonez ou téléchargez le dépôt [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. Créez un fichier ZIP contenant le `stream_manager` dossier contenant le code source du SDK Stream Manager pour Python. Vous pouvez fournir ce fichier ZIP sous forme d'artefact de composant que le logiciel AWS IoT Greengrass Core décompresse lors de l'installation de votre composant. Procédez comme suit :
1. Ouvrez le dossier contenant le référentiel que vous avez cloné ou téléchargé à l'étape précédente.
```
cd aws-greengrass-stream-manager-sdk-python
```
1. Compressez le `stream_manager` dossier dans un fichier ZIP nommé`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. Vérifiez que le `stream_manager_sdk.zip` fichier contient le `stream_manager` dossier et son contenu. Exécutez la commande suivante pour répertorier le contenu du fichier ZIP.
------
#### [ Linux or Unix ]
```
unzip -l stream_manager_sdk.zip
```
------
#### [ Windows Command Prompt (CMD) ]
```
tar -tf stream_manager_sdk.zip
```
------
La sortie doit ressembler à ce qui suit :
```
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. Copiez les artefacts du SDK Stream Manager dans le dossier des artefacts de votre composant. Outre le fichier ZIP du SDK Stream Manager, votre composant utilise le `requirements.txt` fichier du SDK pour installer les dépendances du SDK Stream Manager. *\$1/greengrass-components*Remplacez-le par le chemin du dossier que vous utilisez pour le développement 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. Créez votre recette de composants. Dans la recette, procédez comme suit :
1. Définissez `stream_manager_sdk.zip` et `requirements.txt` en tant qu'artefacts.
1. Définissez votre application Python comme un artefact.
1. Au cours du cycle de vie d'installation, installez les exigences du SDK Stream Manager à partir de`requirements.txt`.
1. Au cours du cycle de vie d'exécution, ajoutez le SDK Stream Manager à votre `PYTHONPATH` application Python et exécutez-la.
La recette de vos composants peut ressembler à l'exemple suivant. Ce composant exécute l'exemple [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
```
------
Pour plus d'informations sur le développement et le test de composants, consultez[Création de AWS IoT Greengrass composants](create-components.md).
### Utilisez le SDK Stream Manager pour JavaScript
Le SDK Stream Manager pour JavaScript est disponible sous forme de code source que vous pouvez inclure dans votre composant. Créez un fichier ZIP du SDK Stream Manager, définissez le fichier ZIP en tant qu'artefact de composant et installez le SDK dans le cycle de vie du composant.
**Pour utiliser le SDK Stream Manager pour JavaScript**
1. Clonez ou téléchargez le dépôt [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. Créez un fichier ZIP contenant le `aws-greengrass-stream-manager-sdk` dossier contenant le code source du SDK Stream Manager pour JavaScript. Vous pouvez fournir ce fichier ZIP sous forme d'artefact de composant que le logiciel AWS IoT Greengrass Core décompresse lors de l'installation de votre composant. Procédez comme suit :
1. Ouvrez le dossier contenant le référentiel que vous avez cloné ou téléchargé à l'étape précédente.
```
cd aws-greengrass-stream-manager-sdk-js
```
1. Compressez le `aws-greengrass-stream-manager-sdk` dossier dans un fichier ZIP nommé`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. Vérifiez que le `stream-manager-sdk.zip` fichier contient le `aws-greengrass-stream-manager-sdk` dossier et son contenu. Exécutez la commande suivante pour répertorier le contenu du fichier ZIP.
------
#### [ Linux or Unix ]
```
unzip -l stream-manager-sdk.zip
```
------
#### [ Windows Command Prompt (CMD) ]
```
tar -tf stream-manager-sdk.zip
```
------
La sortie doit ressembler à ce qui suit :
```
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. Copiez l'artefact du SDK Stream Manager dans le dossier des artefacts de votre composant. *\$1/greengrass-components*Remplacez-le par le chemin du dossier que vous utilisez pour le développement 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. Créez votre recette de composants. Dans la recette, procédez comme suit :
1. Définissez `stream-manager-sdk.zip` comme un artefact.
1. Définissez votre JavaScript application comme un artefact.
1. Au cours du cycle de vie de l'installation, installez le SDK Stream Manager à partir de l'`stream-manager-sdk.zip`artefact. Cette `npm install` commande crée un `node_modules` dossier contenant le SDK Stream Manager et ses dépendances.
1. Dans le cycle de vie d'exécution, ajoutez le `node_modules` dossier et exécutez votre JavaScript application. `NODE_PATH`
La recette de vos composants peut ressembler à l'exemple suivant. Ce composant exécute l'exemple [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
```
------
Pour plus d'informations sur le développement et le test de composants, consultez[Création de AWS IoT Greengrass composants](create-components.md).
## Connectez-vous au gestionnaire de flux dans le code de l'application
Pour vous connecter au gestionnaire de flux dans votre application, créez une instance de `StreamManagerClient` depuis le SDK de Stream Manager. Ce client se connecte au composant du gestionnaire de flux sur son port par défaut 8088, ou sur le port que vous spécifiez. Pour plus d'informations sur la façon de l'utiliser `StreamManagerClient` après avoir créé une instance, consultez[StreamManagerClient À utiliser pour travailler avec des flux](work-with-streams.md).
**Example Exemple : Connexion au gestionnaire de flux avec le port par défaut**
```
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 Exemple : Connexion au gestionnaire de flux avec un port autre que celui par défaut**
Si vous configurez le gestionnaire de flux avec un port autre que le port par défaut, vous devez utiliser [la communication interprocessus](interprocess-communication.md) pour récupérer le port depuis la configuration du composant.
Le paramètre `port` de configuration contient la valeur que vous spécifiez `STREAM_MANAGER_SERVER_PORT` lorsque vous déployez le gestionnaire de flux.
```
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.
```
# StreamManagerClient À utiliser pour travailler avec des flux
Les composants Greengrass définis par l'utilisateur qui s'exécutent sur le périphérique principal de Greengrass peuvent utiliser `StreamManagerClient` l'objet du SDK Stream Manager pour créer des flux dans le gestionnaire de flux, [puis interagir avec les flux](manage-data-streams.md). Lorsqu'un composant crée un flux, il définit les AWS Cloud destinations, la hiérarchisation et les autres politiques d'exportation et de conservation des données pour le flux. Pour envoyer des données au gestionnaire de flux, les composants ajoutent les données au flux. Si une destination d'exportation est définie pour le flux, le gestionnaire de flux exporte le flux automatiquement.
**Note**
Généralement, les clients du gestionnaire de flux sont des composants Greengrass définis par l'utilisateur. Si votre analyse de rentabilisation l'exige, vous pouvez également autoriser les processus non liés aux composants exécutés sur le cœur de Greengrass (par exemple, un conteneur Docker) à interagir avec le gestionnaire de flux. Pour de plus amples informations, veuillez consulter [Authentification client](manage-data-streams.md#stream-manager-security-client-authentication).
Les extraits de cette rubrique vous montrent comment les clients appellent des `StreamManagerClient` méthodes pour travailler avec des flux. Pour plus de détails sur l'implémentation des méthodes et de leurs arguments, utilisez les liens vers la référence du SDK répertoriée après chaque extrait.
Si vous utilisez le gestionnaire de flux dans une fonction Lambda, votre fonction Lambda doit être instanciée `StreamManagerClient` en dehors du gestionnaire de fonctions. Si la fonction est instanciée dans le gestionnaire, elle crée un `client` et une connexion au gestionnaire de flux chaque fois qu'elle est appelée.
**Note**
Si vous effectuez une instanciation `StreamManagerClient` dans le gestionnaire, vous devez appeler explicitement la méthode `close()` lorsque le `client` termine son travail. Sinon, le `client` maintient la connexion ouverte et un autre thread actif jusqu'à ce que le script se termine.
`StreamManagerClient` prend en charge les opérations suivantes :
+ [Créer un flux de messages](#streammanagerclient-create-message-stream)
+ [Ajouter un message](#streammanagerclient-append-message)
+ [Lire des messages](#streammanagerclient-read-messages)
+ [Afficher la liste des flux](#streammanagerclient-list-streams)
+ [Décrire le flux de messages](#streammanagerclient-describe-message-stream)
+ [Mettre à jour le flux de messages](#streammanagerclient-update-message-stream)
+ [Supprimer le flux de messages](#streammanagerclient-delete-message-stream)
## Créer un flux de messages
Pour créer un flux, un composant Greengrass défini par l'utilisateur appelle la méthode create et transmet un objet. `MessageStreamDefinition` Cet objet spécifie le nom unique du flux et définit comment le gestionnaire de flux doit gérer les nouvelles données lorsque la taille maximale du flux est atteinte. Vous pouvez utiliser `MessageStreamDefinition` et ses types de données (tels que `ExportDefinition`, `StrategyOnFull` et `Persistence`) pour définir d'autres propriétés de flux. Il s’agit des licences suivantes :
+ La cible AWS IoT Analytics, Kinesis Data Streams et AWS IoT SiteWise les destinations Amazon S3 pour les exportations automatiques. Pour de plus amples informations, veuillez consulter [Exporter les configurations pour les AWS Cloud destinations prises en charge](stream-export-configurations.md).
+ Exportez la priorité. Le gestionnaire de flux exporte les flux de priorité supérieure avant les flux de priorité inférieure.
+ Taille de lot et intervalle de lot maximaux pour AWS IoT Analytics Kinesis Data Streams AWS IoT SiteWise et les destinations. Le gestionnaire de flux exporte les messages lorsque l'une ou l'autre des conditions est remplie.
+ Time-to-live (TTL). Temps nécessaire pour garantir que les données du flux sont disponibles pour le traitement. Vous devez vous assurer que les données peuvent être consommées pendant cette période. Il ne s'agit pas d'une stratégie de suppression. Les données peuvent ne pas être supprimées immédiatement après la période de TTL.
+ Persistance des flux. Choisissez d'enregistrer les flux dans le système de fichiers afin de conserver les données lors des redémarrages du noyau ou d'enregistrer les flux en mémoire.
+ Numéro de séquence de départ. Spécifiez le numéro de séquence du message à utiliser comme message de départ lors de l'exportation.
Pour plus d'informations`MessageStreamDefinition`, consultez la référence du SDK pour votre langue cible :
+ [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html)dans le SDK Java
+ [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)dans le SDK Node.js
+ [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition)dans le SDK Python
**Note**
`StreamManagerClient`fournit également une destination cible que vous pouvez utiliser pour exporter des flux vers un serveur HTTP. Cette cible n'est destinée qu'à des fins de test. Il n'est pas stable ni pris en charge pour une utilisation dans des environnements de production.
Après la création d'un flux, vos composants Greengrass peuvent [ajouter des messages](#streammanagerclient-append-message) au flux pour envoyer des données à exporter et [lire les messages](#streammanagerclient-read-messages) du flux pour un traitement local. Le nombre de flux que vous créez dépend de vos capacités matérielles et de votre analyse de rentabilisation. L'une des stratégies consiste à créer un flux pour chaque canal cible dans AWS IoT Analytics le flux de données Kinesis, bien que vous puissiez définir plusieurs cibles pour un flux. Un flux a un cycle de vie durable.
### Exigences
Cette opération répond aux exigences suivantes :
+ Version minimale du SDK Stream Manager : Python : 1.1.0 \$1 Java : 1.1.0 \$1 Node.js : 1.1.0
### Exemples
L'extrait de code suivant crée un flux nommé `StreamName`. Il définit les propriétés du flux dans les types de données `MessageStreamDefinition` et les types de données subordonnés.
------
#### [ 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 AWS Cloud.
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.
```
Référence du SDK 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 AWS Cloud.
new ExportDefinition()
.withKinesis(null)
.withIotAnalytics(null)
.withIotSitewise(null)
.withS3(null)
)
);
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Référence du SDK Java : \$1 [createMessageStream[MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#createMessageStream-com.amazonaws.greengrass.streammanager.model.MessageStreamDefinition-)
------
#### [ Node.js ]
```
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
await client.createMessageStream(
new MessageStreamDefinition()
.withName("StreamName") // Required.
.withMaxSize(268435456) // Default is 256 MB.
.withStreamSegmentSize(16777216) // Default is 16 MB.
.withTimeToLiveMillis(null) // By default, no TTL is enabled.
.withStrategyOnFull(StrategyOnFull.OverwriteOldestData) // Required.
.withPersistence(Persistence.File) // Default is File.
.withFlushOnWrite(false) // Default is false.
.withExportDefinition( // Optional. Choose where/how the stream is exported to the AWS Cloud.
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.
});
```
Référence du SDK Node.js : \$1 [createMessageStream[MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#createMessageStream)
------
Pour plus d'informations sur la configuration des destinations d'exportation, consultez[Exporter les configurations pour les AWS Cloud destinations prises en charge](stream-export-configurations.md).
## Ajouter un message
Pour envoyer des données au gestionnaire de flux à des fins d'exportation, vos composants Greengrass ajoutent les données au flux cible. La destination d'exportation détermine le type de données à transmettre à cette méthode.
### Exigences
Cette opération répond aux exigences suivantes :
+ Version minimale du SDK Stream Manager : Python : 1.1.0 \$1 Java : 1.1.0 \$1 Node.js : 1.1.0
### Exemples
#### AWS IoT Analytics ou destinations d'exportation Kinesis Data Streams
L'extrait de code suivant ajoute un message au flux nommé `StreamName`. Pour AWS IoT Analytics nos destinations Kinesis Data Streams, vos composants Greengrass ajoutent un blob de données.
Cet extrait de code répond aux exigences suivantes :
+ Version minimale du SDK Stream Manager : 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.
```
[Référence du SDK 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.
}
```
[Référence du SDK 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.
});
```
[Référence du SDK Node.js : AppendMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage)
------
#### AWS IoT SiteWise destinations d'exportation
L'extrait de code suivant ajoute un message au flux nommé `StreamName`. Pour les AWS IoT SiteWise destinations, vos composants Greengrass ajoutent un objet sérialisé. `PutAssetPropertyValueEntry` Pour de plus amples informations, veuillez consulter [Exporter vers AWS IoT SiteWise](stream-export-configurations.md#export-streams-to-sitewise).
**Note**
Lorsque vous envoyez des données à AWS IoT SiteWise, celles-ci doivent répondre aux exigences de l'`BatchPutAssetPropertyValue`action. Pour plus d’informations, consultez [BatchPutAssetPropertyValue](https://docs.aws.amazon.com/iot-sitewise/latest/APIReference/API_BatchPutAssetPropertyValue.html) dans la *Référence d’API AWS IoT SiteWise *.
Cet extrait de code répond aux exigences suivantes :
+ Version minimale du SDK Stream Manager : 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.
```
Référence du SDK 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.
}
```
Référence du SDK Java : [AppendMessage \$1 [PutAssetPropertyValueEntry](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/sitewise/PutAssetPropertyValueEntry.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-)
------
#### [ Node.js ]
```
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
const maxTimeRandomness = 60;
const maxOffsetRandomness = 10000;
const randomValue = Math.random();
// Note: To create a new asset property data, you should use the classes defined in the
// aws-greengrass-core-sdk StreamManager module.
const timestamp = new TimeInNanos()
.withTimeInSeconds(Math.round(Date.now() / 1000) - Math.floor(Math.random() - maxTimeRandomness))
.withOffsetInNanos(Math.floor(Math.random() * maxOffsetRandomness));
const entry = new AssetPropertyValue()
.withValue(new Variant().withDoubleValue(randomValue))
.withQuality(Quality.GOOD)
.withTimestamp(timestamp);
const putAssetPropertyValueEntry = new PutAssetPropertyValueEntry()
.withEntryId(`${ENTRY_ID_PREFIX}${i}`)
.withPropertyAlias("PropertyAlias")
.withPropertyValues([entry]);
const sequenceNumber = await client.appendMessage("StreamName", util.validateAndSerializeToJsonBytes(putAssetPropertyValueEntry));
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
```
Référence du SDK 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)
------
#### Destinations d'exportation Amazon S3
L'extrait suivant ajoute une tâche d'exportation au flux nommé. `StreamName` Pour les destinations Amazon S3, vos composants Greengrass ajoutent un `S3ExportTaskDefinition` objet sérialisé contenant des informations sur le fichier d'entrée source et l'objet Amazon S3 cible. Si l'objet spécifié n'existe pas, Stream Manager le crée pour vous. Pour de plus amples informations, veuillez consulter [Exportation vers Amazon S3](stream-export-configurations.md#export-streams-to-s3).
Cet extrait de code répond aux exigences suivantes :
+ Version minimale du SDK Stream Manager : 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.
```
[Référence du SDK 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.
}
```
[Référence du SDK Java : [AppendMessage \$1](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-) S3 ExportTaskDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/S3ExportTaskDefinition.html)
------
#### [ Node.js ]
```
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
// Append an Amazon S3 export task definition and print the sequence number.
const taskDefinition = new S3ExportTaskDefinition()
.withBucket("BucketName")
.withKey("KeyName")
.withInputUrl("URLToFile");
const sequenceNumber = await client.appendMessage("StreamName", util.validateAndSerializeToJsonBytes(taskDefinition)));
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
```
[Référence du SDK 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)
------
## Lire des messages
Lisez les messages d'un stream.
### Exigences
Cette opération répond aux exigences suivantes :
+ Version minimale du SDK Stream Manager : Python : 1.1.0 \$1 Java : 1.1.0 \$1 Node.js : 1.1.0
### Exemples
L'extrait de code suivant lit les messages du flux nommé `StreamName`. La méthode read utilise un objet `ReadMessagesOptions` facultatif qui spécifie le numéro de séquence à partir duquel commencer la lecture, les nombres minimum et maximum à lire et un délai d'expiration pour la lecture des messages.
------
#### [ 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.
```
Référence du SDK 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.
}
```
Référence du SDK Java : [ReadMessages \$1 [ReadMessagesOptions](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/ReadMessagesOptions.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#readMessages-java.lang.String-com.amazonaws.greengrass.streammanager.model.ReadMessagesOptions-)
------
#### [ Node.js ]
```
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
const messages = await client.readMessages("StreamName",
// By default, if no options are specified, it tries to read one message from the beginning of the stream.
new ReadMessagesOptions()
// Try to read from sequence number 100 or greater. By default this is 0.
.withDesiredStartSequenceNumber(100)
// Try to read 10 messages. If 10 messages are not available, then NotEnoughMessagesException is thrown. By default, this is 1.
.withMinMessageCount(10)
// Accept up to 100 messages. By default this is 1.
.withMaxMessageCount(100)
// Try to wait at most 5 seconds for the minMessageCount to be fulfilled. By default, this is 0, which immediately returns the messages or an exception.
.withReadTimeoutMillis(5 * 1000)
);
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
```
Référence du SDK 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)
------
## Afficher la liste des flux
Obtenez la liste des flux dans le gestionnaire de flux.
### Exigences
Cette opération répond aux exigences suivantes :
+ Version minimale du SDK Stream Manager : Python : 1.1.0 \$1 Java : 1.1.0 \$1 Node.js : 1.1.0
### Exemples
L'extrait de code suivant récupère une liste des flux (par nom) dans le gestionnaire de flux.
------
#### [ Python ]
```
client = StreamManagerClient()
try:
stream_names = client.list_streams()
except StreamManagerException:
pass
# Properly handle errors.
except ConnectionError or asyncio.TimeoutError:
pass
# Properly handle errors.
```
[Référence du SDK 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.
}
```
[Référence du SDK 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.
});
```
[Référence du SDK Node.js : ListStreams](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#listStreams)
------
## Décrire le flux de messages
Obtenez les métadonnées relatives à un flux, notamment sa définition, sa taille et son statut d'exportation.
### Exigences
Cette opération répond aux exigences suivantes :
+ Version minimale du SDK Stream Manager : Python : 1.1.0 \$1 Java : 1.1.0 \$1 Node.js : 1.1.0
### Exemples
L'extrait de code suivant récupère des métadonnées sur le flux nommé `StreamName`, en particulier la définition, la taille et les statuts d'exportation du flux.
------
#### [ 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.
```
[Référence du SDK 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.
}
```
Référence du SDK 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.
});
```
Référence du SDK Node.js : [describeMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#describeMessageStream)
------
## Mettre à jour le flux de messages
Mettez à jour les propriétés d'un flux existant. Vous souhaiterez peut-être mettre à jour un flux si vos exigences changent après sa création. Par exemple :
+ Ajoutez une nouvelle [configuration d'exportation](stream-export-configurations.md) pour une AWS Cloud destination.
+ Augmentez la taille maximale d'un flux pour modifier la façon dont les données sont exportées ou conservées. Par exemple, la taille du flux associée à votre stratégie en matière de paramètres complets peut entraîner la suppression ou le rejet de données avant que le gestionnaire de flux ne puisse les traiter.
+ Interrompez et reprenez les exportations, par exemple, si les tâches d'exportation sont longues et que vous souhaitez rationner vos données de téléchargement.
Vos composants Greengrass suivent ce processus de haut niveau pour mettre à jour un flux :
1. [Obtenez la description du stream.](#streammanagerclient-describe-message-stream)
1. Mettez à jour les propriétés cibles sur les objets correspondants `MessageStreamDefinition` et subordonnés.
1. Transmettez la mise à jour`MessageStreamDefinition`. Assurez-vous d'inclure les définitions d'objets complètes pour le flux mis à jour. Les propriétés non définies reprennent leurs valeurs par défaut.
Vous pouvez spécifier le numéro de séquence du message à utiliser comme message de départ lors de l'exportation.
### Exigences
Cette opération répond aux exigences suivantes :
+ Version minimale du SDK Stream Manager : Python : 1.1.0 \$1 Java : 1.1.0 \$1 Node.js : 1.1.0
### Exemples
L'extrait suivant met à jour le flux nommé. `StreamName` Elle met à jour plusieurs propriétés d'un flux exporté vers 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.
```
Référence du SDK Python : \$1 [updateMessageStream[MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.update_message_stream)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
MessageStreamInfo messageStreamInfo = client.describeMessageStream(STREAM_NAME);
// Update the message stream with new values.
client.updateMessageStream(
messageStreamInfo.getDefinition()
.withStrategyOnFull(StrategyOnFull.RejectNewData) // Required. Updating Strategy on full to reject new data.
// Max Size update should be greater than initial Max Size defined in Create Message Stream request
.withMaxSize(536870912L) // Update Max Size to 512 MB.
.withStreamSegmentSize(33554432L) // Update Segment Size to 32 MB.
.withFlushOnWrite(true) // Update flush on write to true.
.withPersistence(Persistence.Memory) // Update the persistence to Memory.
.withTimeToLiveMillis(3600000L) // Update TTL to 1 hour.
.withExportDefinition(
// Optional. Choose where/how the stream is exported to the AWS Cloud.
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.
}
```
Référence du SDK Java : [update\$1message\$1stream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#updateMessageStream-java.lang.String-) \$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html)
------
#### [ Node.js ]
```
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
const messageStreamInfo = await c.describeMessageStream(STREAM_NAME);
await client.updateMessageStream(
messageStreamInfo.definition
// Max Size update should be greater than initial Max Size defined in Create Message Stream request
.withMaxSize(536870912) // Default is 256 MB. Updating Max Size to 512 MB.
.withStreamSegmentSize(33554432) // Default is 16 MB. Updating Segment Size to 32 MB.
.withTimeToLiveMillis(3600000) // By default, no TTL is enabled. Update TTL to 1 hour.
.withStrategyOnFull(StrategyOnFull.RejectNewData) // Required. Updating Strategy on full to reject new data.
.withPersistence(Persistence.Memory) // Default is File. Update the persistence to Memory
.withFlushOnWrite(true) // Default is false. Updating to true.
.withExportDefinition(
// Optional. Choose where/how the stream is exported to the AWS Cloud.
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.
});
```
Référence du SDK 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)
------
### Contraintes de mise à jour des flux
Les contraintes suivantes s'appliquent lors de la mise à jour des flux. Sauf indication contraire dans la liste suivante, les mises à jour prennent effet immédiatement.
+ Vous ne pouvez pas mettre à jour la persistance d'un flux. Pour modifier ce comportement, [supprimez le flux](#streammanagerclient-delete-message-stream) et [créez un flux](#streammanagerclient-create-message-stream) qui définit la nouvelle politique de persistance.
+ Vous pouvez mettre à jour la taille maximale d'un flux uniquement dans les conditions suivantes :
+ La taille maximale doit être supérieure ou égale à la taille actuelle du flux. Pour trouver ces informations, [décrivez le flux](#streammanagerclient-describe-message-stream), puis vérifiez l'état de stockage de l'`MessageStreamInfo`objet renvoyé.
+ La taille maximale doit être supérieure ou égale à la taille du segment du flux.
+ Vous pouvez mettre à jour la taille du segment de flux à une valeur inférieure à la taille maximale du flux. Le paramètre mis à jour s'applique aux nouveaux segments.
+ Les mises à jour de la propriété Time to Live (TTL) s'appliquent aux nouvelles opérations d'ajout. Si vous diminuez cette valeur, le gestionnaire de flux peut également supprimer les segments existants qui dépassent le TTL.
+ Les mises à jour de la stratégie relative à la propriété complète s'appliquent aux nouvelles opérations d'ajout. Si vous définissez une stratégie pour remplacer les données les plus anciennes, le gestionnaire de flux peut également remplacer les segments existants en fonction du nouveau paramètre.
+ Les mises à jour de la propriété flush on write s'appliquent aux nouveaux messages.
+ Les mises à jour des configurations d'exportation s'appliquent aux nouvelles exportations. La demande de mise à jour doit inclure toutes les configurations d'exportation que vous souhaitez prendre en charge. Dans le cas contraire, le gestionnaire de flux les supprime.
+ Lorsque vous mettez à jour une configuration d'exportation, spécifiez l'identifiant de la configuration d'exportation cible.
+ Pour ajouter une configuration d'exportation, spécifiez un identifiant unique pour la nouvelle configuration d'exportation.
+ Pour supprimer une configuration d'exportation, omettez-la.
+ Pour [mettre à jour](#streammanagerclient-update-message-stream) le numéro de séquence de départ d'une configuration d'exportation dans un flux, vous devez spécifier une valeur inférieure au dernier numéro de séquence. Pour trouver ces informations, [décrivez le flux](#streammanagerclient-describe-message-stream), puis vérifiez l'état de stockage de l'`MessageStreamInfo`objet renvoyé.
## Supprimer le flux de messages
Supprime un flux. Lorsque vous supprimez un flux, toutes les données stockées dans le flux sont supprimées du disque.
### Exigences
Cette opération répond aux exigences suivantes :
+ Version minimale du SDK Stream Manager : Python : 1.1.0 \$1 Java : 1.1.0 \$1 Node.js : 1.1.0
### Exemples
L'extrait de code suivant supprime le flux nommé `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.
```
Référence du SDK 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.
}
```
[Référence du SDK 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.
});
```
Référence du SDK Node.js : [deleteMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#deleteMessageStream)
------
## Consultez aussi
+ [Gérez les flux de données sur les appareils principaux de Greengrass](manage-data-streams.md)
+ [Configuration du gestionnaire de AWS IoT Greengrass flux](configure-stream-manager.md)
+ [Exporter les configurations pour les AWS Cloud destinations prises en charge](stream-export-configurations.md)
+ `StreamManagerClient`dans la référence du SDK Stream Manager :
+ [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)
# Exporter les configurations pour les AWS Cloud destinations prises en charge
Les composants Greengrass définis par l'utilisateur sont `StreamManagerClient` utilisés dans le SDK Stream Manager pour interagir avec le Stream Manager. Lorsqu'un composant [crée un flux](work-with-streams.md#streammanagerclient-create-message-stream) ou [met à jour un flux](work-with-streams.md#streammanagerclient-create-message-stream), il transmet un `MessageStreamDefinition` objet qui représente les propriétés du flux, y compris la définition de l'exportation. L'`ExportDefinition`objet contient les configurations d'exportation définies pour le flux. Le gestionnaire de flux utilise ces configurations d'exportation pour déterminer où et comment exporter le flux.
![\[Schéma du modèle d'objet du type de ExportDefinition propriété.\]](http://docs.aws.amazon.com/fr_fr/greengrass/v2/developerguide/images/stream-manager-exportconfigs.png)
Vous pouvez définir zéro ou plusieurs configurations d'exportation sur un flux, y compris plusieurs configurations d'exportation pour un seul type de destination. Par exemple, vous pouvez exporter un flux vers deux AWS IoT Analytics canaux et un flux de données Kinesis.
En cas d'échec des tentatives d'exportation, le gestionnaire de flux essaie continuellement d'exporter les données à des AWS Cloud intervalles allant jusqu'à cinq minutes. Le nombre de nouvelles tentatives n'est pas limité.
**Note**
`StreamManagerClient`fournit également une destination cible que vous pouvez utiliser pour exporter des flux vers un serveur HTTP. Cette cible n'est destinée qu'à des fins de test. Il n'est pas stable ni pris en charge pour une utilisation dans des environnements de production.
**Topics**
+ [AWS IoT Analytics chaînes](#export-to-iot-analytics)
+ [Flux de données Amazon Kinesis](#export-to-kinesis)
+ [AWS IoT SiteWise propriétés des actifs](#export-to-iot-sitewise)
+ [Objets Amazon S3](#export-to-s3)
Vous êtes responsable de la maintenance de ces AWS Cloud ressources.
## AWS IoT Analytics chaînes
Le gestionnaire de flux prend en charge les exportations automatiques vers AWS IoT Analytics. AWS IoT Analytics vous permet d'effectuer une analyse avancée de vos données afin de prendre des décisions commerciales et d'améliorer les modèles d'apprentissage automatique. Pour plus d'informations, voir [Qu'est-ce que c'est AWS IoT Analytics ?](https://docs.aws.amazon.com/iotanalytics/latest/userguide/welcome.html) dans le *guide de AWS IoT Analytics l'utilisateur*.
Dans le SDK Stream Manager, vos composants Greengrass utilisent `IoTAnalyticsConfig` le pour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence du SDK pour votre langue cible :
+ [Io TAnalytics Config](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.IoTAnalyticsConfig) dans le SDK Python
+ [Io TAnalytics Config](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTAnalyticsConfig.html) dans le SDK Java
+ [Io TAnalytics Config](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTAnalyticsConfig.html) dans le SDK Node.js
### Exigences
Cette destination d'exportation répond aux exigences suivantes :
+ Les canaux cibles AWS IoT Analytics entrants doivent être identiques à ceux Compte AWS de Région AWS l'appareil principal Greengrass.
+ Ils [Autoriser les appareils principaux à interagir avec les AWS services](device-service-role.md) doivent `iotanalytics:BatchPutMessage` autoriser les chaînes cibles. Par exemple :
------
#### [ 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"
]
}
]
}
```
------
Vous pouvez accorder un accès granulaire ou conditionnel aux ressources, par exemple en utilisant un schéma de `*` dénomination générique. Pour plus d'informations, consultez la section [Ajout et suppression de politiques IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) dans le Guide de l'*utilisateur IAM*.
### Exporter vers AWS IoT Analytics
Pour créer un flux exporté vers AWS IoT Analytics, vos composants Greengrass [créent un flux avec une](work-with-streams.md#streammanagerclient-create-message-stream) définition d'exportation qui inclut un ou plusieurs `IoTAnalyticsConfig` objets. Cet objet définit les paramètres d'exportation, tels que le canal cible, la taille du lot, l'intervalle entre les lots et la priorité.
Lorsque vos composants Greengrass reçoivent des données provenant d'appareils, ils [ajoutent des messages](work-with-streams.md#streammanagerclient-append-message) contenant une quantité importante de données au flux cible.
Le gestionnaire de flux exporte ensuite les données en fonction des paramètres de lot et de la priorité définis dans les configurations d'exportation du flux.
## Flux de données Amazon Kinesis
Le gestionnaire de flux prend en charge les exportations automatiques vers Amazon Kinesis Data Streams. Kinesis Data Streams est couramment utilisé pour agréger de gros volumes de données et les charger dans un entrepôt MapReduce de données ou un cluster. Pour plus d'informations, consultez [Qu'est-ce qu'Amazon Kinesis Data Streams ?](https://docs.aws.amazon.com/streams/latest/dev/what-is-this-service.html) dans le manuel *Amazon Kinesis Developer Guide*.
Dans le SDK Stream Manager, vos composants Greengrass utilisent `KinesisConfig` le pour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence du SDK pour votre langue cible :
+ [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.KinesisConfig)dans le SDK Python
+ [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/KinesisConfig.html)dans le SDK Java
+ [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.KinesisConfig.html)dans le SDK Node.js
### Exigences
Cette destination d'exportation répond aux exigences suivantes :
+ Dans Kinesis Data Streams, les flux cibles doivent se trouver dans le Compte AWS même appareil que celui Région AWS de base de Greengrass.
+ (Recommandé) Le gestionnaire de flux v2.2.1 améliore les performances d'exportation de flux vers des destinations Kinesis Data Streams. Pour utiliser les améliorations de cette dernière version, mettez à niveau votre [composant de gestionnaire de flux](stream-manager-component.md) vers la version 2.2.1 et utilisez la `kinesis:ListShards` politique du rôle d'échange de jetons [Greengrass](device-service-role.md).
+ Ils [Autoriser les appareils principaux à interagir avec les AWS services](device-service-role.md) doivent `kinesis:PutRecords` autoriser le ciblage des flux de données. Par exemple :
------
#### [ 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"
]
}
]
}
```
------
Vous pouvez accorder un accès granulaire ou conditionnel aux ressources, par exemple en utilisant un schéma de `*` dénomination générique. Pour plus d'informations, consultez la section [Ajout et suppression de politiques IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) dans le Guide de l'*utilisateur IAM*.
### Exportation vers Kinesis Data Streams
Pour créer un flux exporté vers Kinesis Data Streams, vos [composants Greengrass créent un flux avec une](work-with-streams.md#streammanagerclient-create-message-stream) définition d'exportation qui inclut un ou plusieurs objets. `KinesisConfig` Cet objet définit les paramètres d'exportation, tels que le flux de données cible, la taille du lot, l'intervalle entre les lots et la priorité.
Lorsque vos composants Greengrass reçoivent des données provenant d'appareils, ils [ajoutent des messages](work-with-streams.md#streammanagerclient-append-message) contenant une quantité importante de données au flux cible. Le gestionnaire de flux exporte ensuite les données en fonction des paramètres de lot et de la priorité définis dans les configurations d'exportation du flux.
Le gestionnaire de flux génère un UUID unique et aléatoire comme clé de partition pour chaque enregistrement chargé sur Amazon Kinesis.
## AWS IoT SiteWise propriétés des actifs
Le gestionnaire de flux prend en charge les exportations automatiques vers AWS IoT SiteWise. AWS IoT SiteWise vous permet de collecter, d'organiser et d'analyser les données des équipements industriels à grande échelle. Pour plus d'informations, voir [Qu'est-ce que c'est AWS IoT SiteWise ?](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/what-is-sitewise.html) dans le *guide de AWS IoT SiteWise l'utilisateur*.
Dans le SDK Stream Manager, vos composants Greengrass utilisent `IoTSiteWiseConfig` le pour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence du SDK pour votre langue cible :
+ [Io TSite WiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.IoTSiteWiseConfig) dans le SDK Python
+ [Io TSite WiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTSiteWiseConfig.html) dans le SDK Java
+ [Io TSite WiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTSiteWiseConfig.html) dans le SDK Node.js
**Note**
AWS fournit également des AWS IoT SiteWise composants, qui offrent une solution prédéfinie que vous pouvez utiliser pour diffuser des données à partir de sources OPC-UA. Pour de plus amples informations, veuillez consulter [Collecteur IoT SiteWise OPC UA](iotsitewise-opcua-collector-component.md).
### Prérequis
Cette destination d'exportation répond aux exigences suivantes :
+ Les propriétés de l'actif cible AWS IoT SiteWise doivent être identiques Compte AWS à Région AWS celles du périphérique principal Greengrass.
**Note**
Pour la liste de Région AWS s compatibles, voir AWS IoT SiteWise les [AWS IoT SiteWise points de terminaison et les quotas](https://docs.aws.amazon.com/general/latest/gr/iot-sitewise.html#iot-sitewise_region) dans la *référence AWS générale*.
+ Ils [Autoriser les appareils principaux à interagir avec les AWS services](device-service-role.md) doivent `iotsitewise:BatchPutAssetPropertyValue` autoriser le ciblage des propriétés des actifs. L'exemple de politique suivant utilise la clé de `iotsitewise:assetHierarchyPath` condition pour accorder l'accès à une ressource racine cible et à ses enfants. Vous pouvez le supprimer `Condition` de la politique pour autoriser l'accès à tous vos AWS IoT SiteWise actifs ou spécifier ARNs des actifs individuels.
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "iotsitewise:BatchPutAssetPropertyValue",
"Resource": "*",
"Condition": {
"StringLike": {
"iotsitewise:assetHierarchyPath": [
"/root node asset ID",
"/root node asset ID/*"
]
}
}
}
]
}
```
------
Vous pouvez accorder un accès granulaire ou conditionnel aux ressources, par exemple en utilisant un schéma de `*` dénomination générique. Pour plus d'informations, consultez la section [Ajout et suppression de politiques IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) dans le Guide de l'*utilisateur IAM*.
Pour obtenir des informations de sécurité importantes, consultez la section [ BatchPutAssetPropertyValue autorisation](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) dans le *guide de AWS IoT SiteWise l'utilisateur*.
### Exporter vers AWS IoT SiteWise
Pour créer un flux exporté vers AWS IoT SiteWise, vos composants Greengrass [créent un flux avec une](work-with-streams.md#streammanagerclient-create-message-stream) définition d'exportation qui inclut un ou plusieurs `IoTSiteWiseConfig` objets. Cet objet définit les paramètres d'exportation, tels que la taille du lot, l'intervalle entre les lots et la priorité.
Lorsque vos composants Greengrass reçoivent des données sur les propriétés des actifs depuis des appareils, ils ajoutent des messages contenant ces données au flux cible. Les messages sont des `PutAssetPropertyValueEntry` objets sérialisés en JSON qui contiennent des valeurs de propriété pour une ou plusieurs propriétés d'actifs. Pour plus d'informations, voir [Ajouter un message pour](work-with-streams.md#streammanagerclient-append-message-sitewise) les destinations AWS IoT SiteWise d'exportation.
**Note**
Lorsque vous envoyez des données à AWS IoT SiteWise, celles-ci doivent répondre aux exigences de l'`BatchPutAssetPropertyValue`action. Pour plus d’informations, consultez [BatchPutAssetPropertyValue](https://docs.aws.amazon.com/iot-sitewise/latest/APIReference/API_BatchPutAssetPropertyValue.html) dans la *Référence d’API AWS IoT SiteWise *.
Le gestionnaire de flux exporte ensuite les données en fonction des paramètres de lot et de la priorité définis dans les configurations d'exportation du flux.
Vous pouvez ajuster les paramètres de votre gestionnaire de flux et la logique des composants Greengrass pour concevoir votre stratégie d'exportation. Par exemple :
+ Pour les exportations en temps quasi réel, définissez des paramètres de taille de lot et d'intervalle faibles et ajoutez les données au flux dès leur réception.
+ Pour optimiser le traitement par lots, atténuer les contraintes de bande passante ou minimiser les coûts, vos composants Greengrass peuvent regrouper timestamp-quality-value les points de données (TQV) reçus pour une propriété d'actif unique avant d'ajouter les données au flux. L'une des stratégies consiste à regrouper les entrées pour un maximum de 10 combinaisons propriétés-actifs différentes, ou alias de propriété, dans un seul message au lieu d'envoyer plusieurs entrées pour la même propriété. Cela permet au gestionnaire de flux de rester dans les limites [AWS IoT SiteWise des quotas](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/quotas.html).
## Objets Amazon S3
Le gestionnaire de flux prend en charge les exportations automatiques vers Amazon S3. Vous pouvez utiliser Amazon S3 pour stocker et récupérer de grandes quantités de données. Pour plus d'informations, consultez [Qu'est-ce qu'Amazon S3 ?](https://docs.aws.amazon.com/AmazonS3/latest/dev/Welcome.html) dans le *guide du développeur d'Amazon Simple Storage Service*.
Dans le SDK Stream Manager, vos composants Greengrass utilisent `S3ExportTaskExecutorConfig` le pour définir la configuration d'exportation pour ce type de destination. Pour plus d'informations, consultez la référence du SDK pour votre langue cible :
+ [S3 ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.S3ExportTaskExecutorConfig) dans le SDK Python
+ [S3 ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/S3ExportTaskExecutorConfig.html) dans le SDK Java
+ [S3 ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.S3ExportTaskExecutorConfig.html) dans le SDK Node.js
### Exigences
Cette destination d'exportation répond aux exigences suivantes :
+ Les compartiments Amazon S3 cibles doivent se trouver dans le même emplacement Compte AWS que le périphérique principal de Greengrass.
+ Si une fonction Lambda exécutée en mode **conteneur Greengrass** écrit des fichiers d'entrée dans un répertoire de fichiers d'entrée, vous devez monter le répertoire en tant que volume dans le conteneur avec des autorisations d'écriture. Cela garantit que les fichiers sont écrits dans le système de fichiers racine et visibles par le composant du gestionnaire de flux, qui s'exécute en dehors du conteneur.
+ Si un composant de conteneur Docker écrit des fichiers d'entrée dans un répertoire de fichiers d'entrée, vous devez monter le répertoire en tant que volume dans le conteneur avec des autorisations d'écriture. Cela garantit que les fichiers sont écrits dans le système de fichiers racine et visibles par le composant du gestionnaire de flux, qui s'exécute en dehors du conteneur.
+ Ils [Autoriser les appareils principaux à interagir avec les AWS services](device-service-role.md) doivent accorder les autorisations suivantes aux compartiments cibles. Par exemple :
------
#### [ 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/*"
]
}
]
}
```
------
Vous pouvez accorder un accès granulaire ou conditionnel aux ressources, par exemple en utilisant un schéma de `*` dénomination générique. Pour plus d'informations, consultez la section [Ajout et suppression de politiques IAM](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html) dans le Guide de l'*utilisateur IAM*.
### Exportation vers Amazon S3
Pour créer un flux exporté vers Amazon S3, vos composants Greengrass utilisent l'`S3ExportTaskExecutorConfig`objet pour configurer la politique d'exportation. La politique définit les paramètres d'exportation, tels que le seuil et la priorité du téléchargement en plusieurs parties. Pour les exportations Amazon S3, le gestionnaire de flux télécharge les données qu'il lit à partir de fichiers locaux sur l'appareil principal. Pour lancer un téléchargement, vos composants Greengrass ajoutent une tâche d'exportation au flux cible. La tâche d'exportation contient des informations sur le fichier d'entrée et l'objet Amazon S3 cible. Le gestionnaire de flux exécute les tâches dans l'ordre dans lequel elles sont ajoutées au flux.
**Note**
Le compartiment cible doit déjà exister dans votre Compte AWS. Si aucun objet correspondant à la clé spécifiée n'existe, le gestionnaire de flux le crée pour vous.
Le gestionnaire de flux utilise la propriété de seuil de téléchargement en plusieurs parties, le paramètre de [taille de pièce minimale](configure-stream-manager.md#stream-manager-minimum-part-size) et la taille du fichier d'entrée pour déterminer le mode de téléchargement des données. Le seuil de téléchargement partitionné doit être supérieur ou égal à la taille de pièce minimale. Si vous souhaitez télécharger des données en parallèle, vous pouvez créer plusieurs flux.
Les clés qui spécifient vos objets Amazon S3 cibles peuvent inclure des DateTimeFormatter chaînes [Java](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html) valides dans les `!{timestamp:value}` espaces réservés. Vous pouvez utiliser ces espaces réservés d'horodatage pour partitionner les données dans Amazon S3 en fonction de l'heure à laquelle les données du fichier d'entrée ont été téléchargées. Par exemple, le nom de clé suivant correspond à une valeur telle que`my-key/2020/12/31/data.txt`.
```
my-key/!{timestamp:YYYY}/!{timestamp:MM}/!{timestamp:dd}/data.txt
```
**Note**
Si vous souhaitez surveiller l'état d'exportation d'un flux, créez d'abord un flux d'état, puis configurez le flux d'exportation pour l'utiliser. Pour de plus amples informations, veuillez consulter [Surveiller les tâches d'exportation](#monitor-export-status-s3).
#### Gérer les données d'entrée
Vous pouvez créer du code que les applications IoT utilisent pour gérer le cycle de vie des données d'entrée. L'exemple de flux de travail suivant montre comment vous pouvez utiliser les composants Greengrass pour gérer ces données.
1. Un processus local reçoit des données provenant d'appareils ou de périphériques, puis écrit les données dans des fichiers situés dans un répertoire du périphérique principal. Il s'agit des fichiers d'entrée pour le gestionnaire de flux.
1. Un composant Greengrass analyse le répertoire et [ajoute une tâche d'exportation](work-with-streams.md#streammanagerclient-append-message-export-task) au flux cible lorsqu'un nouveau fichier est créé. La tâche est un `S3ExportTaskDefinition` objet sérialisé en JSON qui spécifie l'URL du fichier d'entrée, le compartiment et la clé Amazon S3 cibles, ainsi que les métadonnées utilisateur facultatives.
1. Le gestionnaire de flux lit le fichier d'entrée et exporte les données vers Amazon S3 dans l'ordre des tâches ajoutées. Le compartiment cible doit déjà exister dans votre Compte AWS. Si aucun objet correspondant à la clé spécifiée n'existe, le gestionnaire de flux le crée pour vous.
1. Le composant Greengrass [lit les messages](work-with-streams.md#streammanagerclient-read-messages) d'un flux d'état pour surveiller le statut de l'exportation. Une fois les tâches d'exportation terminées, le composant Greengrass peut supprimer les fichiers d'entrée correspondants. Pour de plus amples informations, veuillez consulter [Surveiller les tâches d'exportation](#monitor-export-status-s3).
### Surveiller les tâches d'exportation
Vous pouvez créer du code que les applications IoT utilisent pour surveiller le statut de vos exportations Amazon S3. Vos composants Greengrass doivent créer un flux d'état, puis configurer le flux d'exportation pour écrire des mises à jour de statut dans le flux d'état. Un seul flux de statut peut recevoir des mises à jour de statut provenant de plusieurs flux exportés vers Amazon S3.
[Créez d'abord un flux](work-with-streams.md#streammanagerclient-create-message-stream) à utiliser comme flux d'état. Vous pouvez configurer la taille et les politiques de rétention du flux afin de contrôler la durée de vie des messages d'état. Par exemple :
+ Définissez `Persistence` cette `Memory` option si vous ne souhaitez pas enregistrer les messages d'état.
+ Réglé `StrategyOnFull` sur pour `OverwriteOldestData` que les nouveaux messages d'état ne soient pas perdus.
Créez ou mettez à jour le flux d'exportation pour utiliser le flux d'état. Spécifiquement, définissez la propriété de configuration d'état de la configuration `S3ExportTaskExecutorConfig` d'exportation du flux. Ce paramètre indique au gestionnaire de flux d'écrire des messages d'état concernant les tâches d'exportation dans le flux d'état. Dans l'`StatusConfig`objet, spécifiez le nom du flux d'état et le niveau de verbosité. Les valeurs prises en charge suivantes vont de la moins détaillée (`ERROR`) à la plus détaillée (). `TRACE` La valeur par défaut est `INFO`.
+ `ERROR`
+ `WARN`
+ `INFO`
+ `DEBUG`
+ `TRACE`
L'exemple de flux de travail suivant montre comment les composants Greengrass peuvent utiliser un flux d'état pour surveiller le statut des exportations.
1. Comme décrit dans le flux de travail précédent, un composant [Greengrass ajoute une tâche d'exportation](work-with-streams.md#streammanagerclient-append-message-export-task) à un flux configuré pour écrire des messages d'état concernant les tâches d'exportation dans un flux de statut. L'opération d'ajout renvoie un numéro de séquence qui représente l'ID de tâche.
1. Un composant Greengrass [lit les messages](work-with-streams.md#streammanagerclient-read-messages) de manière séquentielle à partir du flux d'état, puis filtre les messages en fonction du nom du flux et de l'ID de tâche ou en fonction d'une propriété de tâche d'exportation du contexte du message. Par exemple, le composant Greengrass peut filtrer en fonction de l'URL du fichier d'entrée de la tâche d'exportation, qui est représentée par l'`S3ExportTaskDefinition`objet dans le contexte du message.
Les codes d'état suivants indiquent qu'une tâche d'exportation est terminée :
+ `Success`. Le téléchargement a été effectué avec succès.
+ `Failure`. Le gestionnaire de flux a rencontré une erreur. Par exemple, le bucket spécifié n'existe pas. Une fois le problème résolu, vous pouvez à nouveau ajouter la tâche d'exportation au flux.
+ `Canceled`. La tâche a été arrêtée car le flux ou la définition d'exportation a été supprimé ou parce que la période time-to-live (TTL) de la tâche a expiré.
**Note**
La tâche peut également avoir le statut `InProgress` ou`Warning`. Le gestionnaire de flux émet des avertissements lorsqu'un événement renvoie une erreur qui n'affecte pas l'exécution de la tâche. Par exemple, l'échec du nettoyage d'un téléchargement partiel renvoie un avertissement.
1. Une fois les tâches d'exportation terminées, le composant Greengrass peut supprimer les fichiers d'entrée correspondants.
L'exemple suivant montre comment un composant Greengrass peut lire et traiter les messages d'état.
------
#### [ 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.
```
Référence du SDK 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.
}
```
Référence du SDK Java : [ReadMessages \$1 [StatusMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/StatusMessage.html)](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#readMessages-java.lang.String-com.amazonaws.greengrass.streammanager.model.ReadMessagesOptions-)
------
#### [ Node.js ]
```
const {
StreamManagerClient, ReadMessagesOptions,
Status, StatusConfig, StatusLevel, StatusMessage,
util,
} = require(*'aws-greengrass-stream-manager-sdk'*);
const client = new StreamManagerClient();
client.onConnected(async () => {
try {
let isS3UploadComplete = false;
while (!isS3UploadComplete) {
try {
// Read the statuses from the export status stream
const messages = await c.readMessages("StatusStreamName",
new ReadMessagesOptions()
.withMinMessageCount(1)
.withReadTimeoutMillis(1000));
messages.forEach((message) => {
// Deserialize the status message first.
const statusMessage = util.deserializeJsonBytesToObj(message.payload, StatusMessage);
// Check the status of the status message. If the status is 'Success', the file was successfully uploaded to S3.
// If the status was either 'Failure' or 'Cancelled', the server was unable to upload the file to S3.
// We will print the message for why the upload to S3 failed from the status message.
// If the status was "InProgress", the status indicates that the server has started uploading the S3 task.
if (statusMessage.status === Status.Success) {
console.log(`Successfully uploaded file at path ${FILE_URL} to S3.`);
isS3UploadComplete = true;
} else if (statusMessage.status === Status.Failure || statusMessage.status === Status.Canceled) {
console.log(`Unable to upload file at path ${FILE_URL} to S3. Message: ${statusMessage.message}`);
isS3UploadComplete = true;
}
});
// Sleep for sometime for the S3 upload task to complete before trying to read the status message.
await new Promise((r) => setTimeout(r, 5000));
} catch (e) {
// Ignored
}
} catch (e) {
// Properly handle errors.
}
});
client.onError((err) => {
// Properly handle connection errors.
// This is called only when the connection to the StreamManager server fails.
});
```
Référence du SDK 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)
------