Exportation de flux de données vers leAWS Cloud(CLI)

Télécharger leAWS IoT GreengrassKit SDK pour Pythonv1.5.0 ou version ultérieure.

Décompressez le package téléchargé pour obtenir le kit SDK. Le kit SDK est représenté par le dossier greengrasssdk.

Installer Installer Installer les dépendances de package Installer les dépendances de package à inclure dans le kit de déploiement de votre fonction Lambda.

Enregistrez la fonction de code Python suivante dans un fichier local nommé transfer_stream.py.

import asyncio
import logging
import random
import time

from greengrasssdk.stream_manager import (
    ExportDefinition,
    KinesisConfig,
    MessageStreamDefinition,
    ReadMessagesOptions,
    ResourceNotFoundException,
    StrategyOnFull,
    StreamManagerClient,
)


# This example creates a local stream named "SomeStream".
# It starts writing data into that stream and then stream manager automatically exports  
# the data to a customer-created Kinesis data stream named "MyKinesisStream". 
# This example runs forever until the program is stopped.

# The size of the local stream on disk will not exceed the default (which is 256 MB).
# Any data appended after the stream reaches the size limit continues to be appended, and
# stream manager deletes the oldest data until the total stream size is back under 256 MB.
# The Kinesis data stream in the cloud has no such bound, so all the data from this script is
# uploaded to Kinesis and you will be charged for that usage.


def main(logger):
    try:
        stream_name = "SomeStream"
        kinesis_stream_name = "MyKinesisStream"

        # Create a client for the StreamManager
        client = StreamManagerClient()

        # Try deleting the stream (if it exists) so that we have a fresh start
        try:
            client.delete_message_stream(stream_name=stream_name)
        except ResourceNotFoundException:
            pass

        exports = ExportDefinition(
            kinesis=[KinesisConfig(identifier="KinesisExport" + stream_name, kinesis_stream_name=kinesis_stream_name)]
        )
        client.create_message_stream(
            MessageStreamDefinition(
                name=stream_name, strategy_on_full=StrategyOnFull.OverwriteOldestData, export_definition=exports
            )
        )

        # Append two messages and print their sequence numbers
        logger.info(
            "Successfully appended message to stream with sequence number %d",
            client.append_message(stream_name, "ABCDEFGHIJKLMNO".encode("utf-8")),
        )
        logger.info(
            "Successfully appended message to stream with sequence number %d",
            client.append_message(stream_name, "PQRSTUVWXYZ".encode("utf-8")),
        )

        # Try reading the two messages we just appended and print them out
        logger.info(
            "Successfully read 2 messages: %s",
            client.read_messages(stream_name, ReadMessagesOptions(min_message_count=2, read_timeout_millis=1000)),
        )

        logger.info("Now going to start writing random integers between 0 and 1000 to the stream")
        # Now start putting in random data between 0 and 1000 to emulate device sensor input
        while True:
            logger.debug("Appending new random integer to stream")
            client.append_message(stream_name, random.randint(0, 1000).to_bytes(length=4, signed=True, byteorder="big"))
            time.sleep(1)

    except asyncio.TimeoutError:
        logger.exception("Timed out while executing")
    except Exception:
        logger.exception("Exception while running")


def function_handler(event, context):
    return


logging.basicConfig(level=logging.INFO)
# Start up this sample code
main(logger=logging.getLogger())

Compressez les éléments suivants dans un fichier nommé transfer_stream_python.zip. Il s'agit du package de déploiement de votre fonction Lambda.

Lorsque vous créez le fichier zip, incluez uniquement ces éléments, et non le dossier conteneur.

Créez un rôle IAM afin de pouvoir transmettre l'ARN du rôle lorsque vous créez un rôle IAM pour vous permettre de transmettre l'ARN du rôle

Copiez la Arn à partir de la sortie.

Utilisez l'API AWS Lambda pour créer la fonction TransferStream. La commande suivante suppose que le fichier ZIP se trouve dans le répertoire actuel.

aws lambda create-function \
--function-name TransferStream \
--zip-file fileb://transfer_stream_python.zip \
--role role-arn \
--handler transfer_stream.function_handler \
--runtime python3.7

Publiez une version de la fonction.

aws lambda publish-version --function-name TransferStream --description 'First version'

Créez un alias pour la version publiée.

Les groupes Greengrass peuvent référencer une fonction Lambda par alias (recommandé) ou par version. L'utilisation d'un alias facilite la gestion des mises à jour de code, car vous n'avez pas besoin de changer votre table d'abonnement ou la définition de groupe lorsque le code de fonction est mis à jour. À la place, il vous suffit de pointer l'alias vers la nouvelle version de fonction.

aws lambda create-alias --function-name TransferStream --name GG_TransferStream --function-version 1

Copiez la AliasArn à partir de la sortie. Vous utilisez cette valeur lorsque vous configurez la fonction pour AWS IoT Greengrass.

Créez une définition de fonction avec une version initiale qui contient les fonctions système et les fonctions Lambda définies par l'utilisateur.

La version de définition suivante active le gestionnaire de flux avec leréglages des paramètres. Pour configurer des paramètres personnalisés, vous devez définir des variables d'environnement pour les paramètres du gestionnaire de flux correspondants. Pour obtenir un exemple, veuillez consulter .Pour activer, désactiver ou configurer le gestionnaire de flux (interface de ligne de commande).AWS IoT Greengrassutilise les paramètres par défaut pour tous les paramètres qui sont omis.MemorySizedoit être au moins128000.Pinneddoit être défini surtrue.

Copiez la LatestVersionArn à partir de la sortie. Vous utilisez cette valeur pour ajouter la version de définition de fonction à la version de groupe que vous déployez pour le noyau.

Créez une définition d'enregistreur qui inclut une version initiale.

Copiez l'élément LatestVersionArn de la définition de l'enregistreur à partir de la sortie. Vous utilisez cette valeur pour ajouter la version de la définition de l'enregistreur à la version de groupe que vous déployez sur le noyau.

Obtenez les ID du groupe et de la version de groupe Greengrass cible. Cette procédure suppose qu'il s'agit du dernier groupe et de la dernière version de groupe les plus récents La requête suivante renvoie le groupe créé le plus récemment.

aws greengrass list-groups --query "reverse(sort_by(Groups, &CreationTimestamp))[0]"

Vous pouvez également procéder à une interrogation par nom. Les noms de groupe ne devant pas nécessairement être uniques, plusieurs groupes peuvent être renvoyés.

aws greengrass list-groups --query "Groups[?Name=='MyGroup']"

Copiez l'Id du groupe cible à partir de la sortie. Vous utilisez cela pour obtenir la version de définition du noyau et lorsque vous déployez le groupe.

Copiez l'élément LatestVersion à partir de la sortie (ID de la dernière version ajoutée au groupe). Vous utilisez cela pour obtenir la version de la définition du noyau.

Obtenir l'ARN de la version de la définition du noyau :

1. Obtenez la version de groupe.

   • Remplacez group-id par l'Id que vous avez copié pour le groupe.

   • Remplacezgroup-version-idavec leLatestVersionque vous avez copiés pour le groupe.

   aws greengrass get-group-version \
   --group-id group-id \
   --group-version-id group-version-id

2. Copiez la CoreDefinitionVersionArn à partir de la sortie. Vous utilisez cette valeur pour ajouter la version de définition du noyau à la version de groupe que vous déployez sur le noyau.

Créer une version de groupe.

aws greengrass create-group-version \
--group-id group-id \
--core-definition-version-arn core-definition-version-arn \
--function-definition-version-arn function-definition-version-arn \
--logger-definition-version-arn logger-definition-version-arn

Copiez la Version à partir de la sortie. Il s'agit de l'ID de la nouvelle version de groupe.

Vérifiez que leAWS IoT Greengrasscore est en cours d'exécution. Dans la fenêtre de terminal de votre Raspberry Pi, exécutez les commandes suivantes, si nécessaire.

1. Pour vérifier si le démon est en cours d'exécution :

   ps aux | grep -E 'greengrass.*daemon'

   Si la sortie contient une entrée root pour /greengrass/ggc/packages/ggc-version/bin/daemon, le démon est en cours d'exécution.

   Note

   La version du chemin d'accès dépend de la version du logiciel AWS IoT Greengrass Core installée sur votre appareil principal.

2. Pour démarrer le démon :

   cd /greengrass/ggc/core/
   sudo ./greengrassd start

Créer un déploiement.

aws greengrass create-deployment \
--deployment-type NewDeployment \
--group-id group-id \
--group-version-id group-version-id

Copiez la DeploymentId à partir de la sortie.

Obtenir le statut du déploiement.

aws greengrass get-deployment-status \
--group-id group-id \
--deployment-id deployment-id

Si le statut estSuccess, le déploiement a réussi. Pour bénéficier d'une aide à la résolution des problèmes, consultez Résolution des problèmes de AWS IoT Greengrass.

Dans la console Amazon Kinesis, sousKinesis data flux de données, choisissezMyKinesisStream.

Dans la pageMyKinesisStreampage, choisissezSurveillance. Si le test réussit, vous devriez voir des données dans les graphiques PutRecords . Selon votre connexion, l'affichage des données peut prendre une minute.

cd /greengrass/ggc/core/
sudo ./greengrassd stop

Supprimez leTransferStreamUne fonction Lambda depuis le cœur.

1. Suivez l'Étape 6 : Créer une version de groupe pour créer une nouvelle version de groupe, mais supprimez l'option --function-definition-version-arn dans la commande create-group-version. Vous pouvez également créer une version de définition de fonction qui n'inclut pas leTransferStreamfonction Lambda.

   Note

   En omettant le systèmeGGStreamManagerInstaller la fonction Lambda à partir de la version du groupe déployé, vous désactivez la gestion de flux sur le noyau.

2. Suivez l'Étape 7 : Créer un déploiement pour déployer la nouvelle version du groupe.