Exportieren von Daten-Streams inAWS Cloud(CLI)

Herunterladen desAWS IoT GreengrassCore-SDK für Pythonv1.5.0 oder höher.

Entpacken Sie das heruntergeladene Paket, um das SDK zu erhalten. Das SDK ist der greengrasssdk-Ordner.

Python installieren Sie Paketabhängigkeiten, die mit dem SDK in Ihr Bereitstellungspaket der Lambda-Funktion

Speichern Sie die folgende Pythoncode-Funktion in einer lokalen Datei namens 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())

Packen Sie die folgenden Elemente in einer ZIP-Datei mit dem Namen transfer_stream_python.zip. Dies ist das Bereitstellungspaket Ihrer Lambda-Funktion.

Wenn Sie die zip-Datei erstellen, schließen Sie nur diese Elemente ein, nicht den enthaltenen Ordner.

Erstellen Sie eine IAM-Rolle, damit Sie beim Anlegen der -Funktion den Rollen-ARN übergeben können.

Kopieren Sie die Arn aus der Ausgabe.

Verwenden Sie die AWS Lambda-API, um die TransferStream-Funktion zu erstellen. Der folgende Befehl geht davon aus, dass sich die Zip-Datei im aktuellen Verzeichnis befindet.

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

Veröffentlichen einer Version der Funktion.

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

Erstellen Sie einen Alias für die veröffentlichte Version.

Greengrass-Gruppen können eine Lambda-Funktion per Alias (empfohlen) oder nach Version referenzieren. Die Verwendung eines Alias erleichtert die Verwaltung von Code-Updates, da Sie Ihre Abonnementtabelle oder Gruppendefinition nicht ändern müssen, sobald der Funktionscode aktualisiert wird. Stattdessen verweisen Sie einfach den Alias auf die neue Funktionsversion.

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

Kopieren Sie die AliasArn aus der Ausgabe. Sie verwenden diesen Wert, wenn Sie die Funktion für AWS IoT Greengrass konfigurieren.

Erstellen Sie eine Funktionsdefinition, die die System- und benutzerdefinierten Lambda-Funktionen enthält.

Die folgende Definitionsversionparameter-Einstellungenaus. Um benutzerdefinierte Einstellungen zu konfigurieren, müssen Sie Umgebungsvariablen für entsprechende Stream-Managers. Ein Beispiel finden Sie unterSo aktivieren, deaktivieren oder konfigurieren Sie den Stream-Manager (CLI)aus.AWS IoT Greengrassverwendet Standardeinstellungen.MemorySizesollte mindestens erforderlich sein128000aus.Pinnedmuss auf festgelegt seintrueaus.

Kopieren Sie die LatestVersionArn aus der Ausgabe. Mit diesem Wert fügen Sie die Version der Funktionsdefinition der Gruppenversion hinzu, die Sie im Core bereitstellen.

Erstellen Sie eine Logger-Definition, die eine Erstversion enthält.

Kopieren Sie den LatestVersionArn der Logger-Definition aus der Ausgabe. Mit diesem Wert fügen Sie die Version der Logger-Definition der Gruppenversion hinzu, die Sie im Core bereitstellen.

Rufen Sie die IDs der Greengrass-Zielgruppen und die Gruppenversion ab. Dieses Verfahren setzt voraus, dass es sich um die neueste Gruppe und Gruppenversion handelt. Die folgende Abfrage gibt die zuletzt erstellte Gruppe zurück.

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

Sie können auch nach Namen abfragen. Gruppennamen müssen nicht eindeutig sein, sodass mehrere Gruppen zurückgegeben werden können.

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

Kopieren Sie die Id der Zielgruppe aus der Ausgabe. Sie verwenden sie, um die Core-Definitionsversion zu erhalten, und wenn Sie die Gruppe bereitstellen.

Kopieren Sie die LatestVersion aus der Ausgabe, d. h. die ID der letzten Version, die der Gruppe hinzugefügt wurde. Damit erhalten Sie die Core-Definitionsversion.

Abrufen des ARN der Core-Definitionsversion:

1. Abrufen der Gruppenversion.

   • Ersetzen Sie group-id durch die kopierte Id für die Gruppe.

   • Ersetzengroup-version-idmit demLatestVersiondie du für die Gruppe kopiert hast.

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

2. Kopieren Sie die CoreDefinitionVersionArn aus der Ausgabe. Mit diesem Wert fügen Sie die Core-Definitionsversion der Gruppenversion hinzu, die Sie im Core bereitstellen.

Erstellen einer Gruppenversion

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

Kopieren Sie die Version aus der Ausgabe. Dies ist die ID der neuen Gruppenversion.

Stellen Sie sicher, dass dieAWS IoT Greengrasscore läuft. Führen Sie im Raspberry Pi-Terminal die folgenden Befehle aus, falls nötig.

1. So prüfen Sie, ob der Daemon ausgeführt wird:

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

   Wenn die Ausgabe einen root-Eintrag für /greengrass/ggc/packages/ggc-version/bin/daemon enthält, dann wird der Daemon ausgeführt.

   Anmerkung

   Die Version in dem Pfad hängt von der AWS IoT Greengrass-Core-Softwareversion ab, die auf Ihrem Core-Gerät installiert ist.

2. So starten Sie den Daemon:

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

Erstellen einer Bereitstellung.

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

Kopieren Sie die DeploymentId aus der Ausgabe.

Abrufen des Bereitstellungsstatus.

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

Wenn der Status lautetSuccesswar die Bereitstellung erfolgreich. Hilfe zur Problembehebung finden Sie unter Fehlerbehebung für AWS IoT Greengrass.

In der Amazon Kinesis Kinesis-Konsole unterKinesis Data Streams, wählenMyKinesisStreamaus.

Auf derMyKinesisStream-Seite, wählen SieÜberwachungaus. Wenn der Test erfolgreich ist, sollten Sie die Daten in den Put Records (Datensätze übergeben)-Diagrammen sehen. Je nach Verbindung kann es eine Minute dauern, bis die Daten angezeigt werden.

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

Entfernen Sie dasTransferStreamLambda-Funktion aus dem Kern.

1. Folgen Sie Schritt 6: Erstellen einer Gruppenversion, um eine neue Gruppenversion zu erstellen. Entfernen Sie jedoch die --function-definition-version-arn-Option im create-group-version-Befehl. Sie können auch eine Funktionsdefinitionsversion erstellen, dieTransferStreamLambda-Funktion.

   Anmerkung

   Durch das Weglassen des SystemsGGStreamManagerLambda—Funktion aus der bereitgestellten Gruppenversion, deaktivieren Sie die Stream-Managers.

2. Folgen Sie Schritt 7: Eine Bereitstellung auswählen, um die neue Gruppenversion bereitzustellen.