Exportar secuencias de datos a la Nube de AWS (CLI)

Descargue el SDK de AWS IoT Greengrass Core para Python versión 1.5.0 o posterior.

Descomprima el paquete descargado para obtener el SDK. El SDK es la carpeta greengrasssdk.

Instale dependencias de paquetes para incluirlas con el SDK en su paquete de implementación de funciones de Lambda.

Guarde la siguiente función de código de Python en un archivo local llamado "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())

Comprima en un archivo ZIP los siguientes elementos en un archivo denominado "transfer_stream_python.zip". Este es el paquete de implementación de la función de Lambda.

Al crear el archivo de zip, incluya solo estos elementos, no la carpeta que los contiene.

Cree un rol de IAM para poder pasar el ARN del rol cuando cree la función.

Copie la Arn del resultado.

Utilice la API de AWS Lambda para crear la función de TransferStream. El siguiente comando da por hecho que el archivo ZIP está en el directorio actual.

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

Publique una versión de la función.

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

Cree un alias a la versión publicada.

Los grupos de Greengrass pueden hacer referencia a una función de Lambda por versión o alias (recomendado). El uso de un alias facilita la gestión de las actualizaciones del código porque no tiene que cambiar la tabla de suscripción o la definición del grupo cuando se actualiza el código de la función. En su lugar, basta con apuntar el alias a la nueva versión de la función.

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

Copie la AliasArn del resultado. Este valor se usa al configurar la función para AWS IoT Greengrass.

Cree una definición de función con una versión inicial que contenga las funciones de Lambda del sistema y definidas por el usuario.

Con la siguiente versión de definición se habilita al administrador de secuencias con la configuración de parámetros predeterminada. Para configurar parámetros personalizados, debe definir variables de entorno para los parámetros correspondientes del administrador de secuencias. Para ver un ejemplo, consulte Para habilitar, deshabilitar o configurar el administrador de secuencias(CLI). AWS IoT Greengrass utiliza características predeterminadas para los parámetros que se omiten. MemorySize debería ser al menos 128000. Pinned debe establecerse en true.

Copie la LatestVersionArn del resultado. Este valor se usa para añadir la versión de la definición de función a la versión de grupo que implementó en el núcleo.

Cree una definición del registro que incluya una versión inicial.

Copie el LatestVersionArn de la definición del registro del resultado. Este valor se usa para añadir la versión de definición del registro a la versión del grupo que implementa en el núcleo.

Obtenga los ID de la versión de grupo y grupo de Greengrass de destino. En este procedimiento, suponemos que estos son el último grupo y la última versión de grupo. La siguiente consulta devuelve el grupo creado más recientemente.

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

También puede hacer la consulta por nombre. No es necesario que los nombres de grupo sean únicos, por lo que podrían devolverse varios grupos.

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

Copie el Id del grupo de destino de la salida. Puede utilizar esto para obtener la versión de la definición de núcleo y al implementar el grupo.

Copie el LatestVersion del resultado, que es el ID de la última versión añadida al grupo. Puede utilizar esto para obtener la versión de la definición de núcleo.

Obtenga el ARN de la versión de la definición principal:

1. Obtenga la versión de grupo.

   • Reemplace id-grupo con el Id que ha copiado para el grupo.

   • Reemplace group-version-id por el parámetro LatestVersion que ha copiado para el grupo.

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

2. Copie la CoreDefinitionVersionArn del resultado. Este valor se usa para añadir la versión de la definición del núcleo a la versión de grupo que implementó en el núcleo.

Cree una versión de grupo.

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

Copie la Version del resultado. Este es el ID de la nueva versión del grupo.

Asegúrese de que el núcleo de AWS IoT Greengrass se está ejecutando. Ejecute los siguientes comandos en el terminal de Raspberry Pi según sea necesario.

1. Para comprobar si el demonio está en ejecución:

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

   Si la salida contiene una entrada root para /greengrass/ggc/packages/ggc-version/bin/daemon, el demonio está en ejecución.

   nota

   La versión que figura en la ruta depende de la versión del software AWS IoT Greengrass Core que esté instalada en el dispositivo del núcleo.

2. Iniciar el daemon:

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

Cree una implementación de .

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

Copie la DeploymentId del resultado.

Obtenga el estado de las implementaciones.

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

Si el estado es Success, la implementación fue correcta. Para obtener ayuda sobre la resolución de problemas, consulte Solución de problemas de AWS IoT Greengrass.

En la consola de Amazon Kinesis , bajo las Secuencias de datos de Kinesis, seleccione MyKinesisStream.

En la página MyKinesisStream, seleccione Monitoring (Supervisión). Si la prueba se realiza correctamente, debería ver los datos en los gráficos Put Records. En función de la conexión, es posible que tarde un minuto en mostrar los datos.

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

Elimine la función de Lambda TransferStream del núcleo.

1. Siga Paso 6: Crear una versión del grupo para crear una nueva versión de grupo. pero elimine la opción --function-definition-version-arn en el comando create-group-version. O bien, cree una versión de definición de función que no incluya la función de Lambda TransferStream.

   nota

   Al omitir la función de Lambda de GGStreamManager del sistema de la versión de grupo implementada, se deshabilita la administración de secuencias en el núcleo.

2. Siga Paso 7: Crear una implementación para implementar la nueva versión de grupo.