Export data streams to the Amazon Web Services Cloud (CLI)

Download the Amazon IoT Greengrass Core SDK for Python v1.5.0 or later.

Unzip the downloaded package to get the SDK. The SDK is the greengrasssdk folder.

Install package dependencies to include with the SDK in your Lambda function deployment package.

Save the following Python code function in a local file named 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())

Zip the following items into a file named transfer_stream_python.zip. This is your Lambda function deployment package.

When you create the zip file, include only these items, not the containing folder.

Create an IAM role so you can pass in the role ARN when you create the function.

Copy the Arn from the output.

Use the Amazon Lambda API to create the TransferStream function. The following command assumes that the zip file is in the current directory.

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

Publish a version of the function.

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

Create an alias for the published version.

Greengrass groups can reference a Lambda function by alias (recommended) or by version. Using an alias makes it easier to manage code updates because you don't have to change your subscription table or group definition when the function code is updated. Instead, you just point the alias to the new function version.

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

Copy the AliasArn from the output. You use this value when you configure the function for Amazon IoT Greengrass.

Create a function definition with an initial version that contains the system and user-defined Lambda functions.

The following definition version enables stream manager with default parameter settings. To configure custom settings, you must define environment variables for corresponding stream manager parameters. For an example, see To enable, disable, or configure stream manager (CLI). Amazon IoT Greengrass uses default settings for parameters that are omitted. MemorySize should be at least 128000. Pinned must be set to true.

Copy the LatestVersionArn from the output. You use this value to add the function definition version to the group version that you deploy to the core.

Create a logger definition that includes an initial version.

Copy the LatestVersionArn of the logger definition from the output. You use this value to add the logger definition version to the group version that you deploy to the core.

Get the IDs of the target Greengrass group and group version. This procedure assumes that this is the latest group and group version. The following query returns the most recently created group.

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

Or, you can query by name. Group names are not required to be unique, so multiple groups might be returned.

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

Copy the Id of the target group from the output. You use this to get the core definition version and when you deploy the group.

Copy the LatestVersion from the output, which is the ID of the last version added to the group. You use this to get the core definition version.

Get the ARN of the core definition version:

1. Get the group version.

   • Replace group-id with the Id that you copied for the group.

   • Replace group-version-id with the LatestVersion that you copied for the group.

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

2. Copy the CoreDefinitionVersionArn from the output. You use this value to add the core definition version to the group version that you deploy to the core.

Create a group version.

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

Copy the Version from the output. This is the ID of the new group version.

Make sure that the Amazon IoT Greengrass core is running. Run the following commands in your Raspberry Pi terminal, as needed.

1. To check whether the daemon is running:

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

   If the output contains a root entry for /greengrass/ggc/packages/ggc-version/bin/daemon, then the daemon is running.

   Note

   The version in the path depends on the Amazon IoT Greengrass Core software version that's installed on your core device.

2. To start the daemon:

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

Create a deployment.

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

Copy the DeploymentId from the output.

Get the deployment status.

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

If the status is Success, the deployment was successful. For troubleshooting help, see Troubleshooting Amazon IoT Greengrass.

In the Amazon Kinesis console, under Kinesis data streams, choose MyKinesisStream.

On the MyKinesisStream page, choose Monitoring. If the test is successful, you should see data in the Put Records charts. Depending on your connection, it might take a minute before the data is displayed.

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

Remove the TransferStream Lambda function from the core.

1. Follow Step 6: Create a group version to create a new group version. but remove the --function-definition-version-arn option in the create-group-version command. Or, create a function definition version that doesn't include the TransferStream Lambda function.

   Note

   By omitting the system GGStreamManager Lambda function from the deployed group version, you disable stream management on the core.

2. Follow Step 7: Create a deployment to deploy the new group version.