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

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.

First, create the Lambda function.

1. In the Amazon Web Services Management Console, choose Services, and open the Amazon Lambda console.

2. Choose Create function and then choose Author from scratch.

3. In the Basic information section, use the following values:

   • For Function name, enter TransferStream.

   • For Runtime, choose Python 3.7.

   • For Permissions, keep the default setting. This creates an execution role that grants basic Lambda permissions. This role isn't used by Amazon IoT Greengrass.

4. At the bottom of the page, choose Create function.

Next, register the handler and upload your Lambda function deployment package.

1. On the Code tab, under Code source, choose Upload from. From the dropdown, choose .zip file.

   

2. Choose Upload, and then choose your transfer_stream_python.zip deployment package. Then, choose Save.

3. On the Code tab for the function, under Runtime settings, choose Edit, and then enter the following values.

   • For Runtime, choose Python 3.7.

   • For Handler, enter transfer_stream.function_handler

4. Choose Save.

   Note

   The Test button on the Amazon Lambda console doesn't work with this function. The Amazon IoT Greengrass Core SDK doesn't contain modules that are required to run your Greengrass Lambda functions independently in the Amazon Lambda console. These modules (for example, greengrass_common) are supplied to the functions after they are deployed to your Greengrass core.

Now, publish the first version of your Lambda function and create an alias for the version.

1. From the Actions menu, choose Publish new version.

2. For Version description, enter First version, and then choose Publish.

3. On the TransferStream: 1 configuration page, from the Actions menu, choose Create alias.

4. On the Create a new alias page, use the following values:

   • For Name, enter GG_TransferStream.

   • For Version, choose 1.

   Note

   Amazon IoT Greengrass doesn't support Lambda aliases for $LATEST versions.

5. Choose Create.

In the Amazon IoT console navigation pane, under Manage, expand Greengrass devices, and then choose Groups (V1).

Choose the target group.

On the group configuration page, choose the Lambda functions tab.

Under My Lambda functions, choose Add.

On the Add Lambda function page, choose the Lambda function for your Lambda function.

For the Lambda version, choose Alias:GG_TransferStream.

Now, configure properties that determine the behavior of the Lambda function in the Greengrass group.

In the Lambda function configuration section, make the following changes:

Choose Add Lambda function.

On the group configuration page, choose the Lambda functions tab.

Under System Lambda functions, select Stream manager, and check the status. If disabled, choose Edit. Then, choose Enable and Save. You can use the default parameter settings for this tutorial. For more information, see Configure Amazon IoT Greengrass stream manager.

Under Local logs configuration, check if local logging is configured.

If logs aren't configured for Greengrass system components or user-defined Lambda functions, choose Edit.

Choose User Lambda functions log level and Greengrass system log level.

Keep the default values for logging level and disk space limit, and then choose Save.

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

On the group configuration page, choose Deploy.

1. In the Lambda functions tab, under the System Lambda functions section, select IP detector and choose Edit.

2. In the Edit IP detector settings dialog box, select Automatically detect and override MQTT broker endpoints.

3. Choose Save.

   This enables devices to automatically acquire connectivity information for the core, such as IP address, DNS, and port number. Automatic detection is recommended, but Amazon IoT Greengrass also supports manually specified endpoints. You're only prompted for the discovery method the first time that the group is deployed.

   Note

   If prompted, grant permission to create the Greengrass service role and associate it with your Amazon Web Services account in the current Amazon Web Services Region. This role allows Amazon IoT Greengrass to access your resources in Amazon services.

   The Deployments page shows the deployment timestamp, version ID, and status. When completed, the status displayed for the deployment should be Completed.

   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. In the Amazon IoT console navigation pane, under Manage, expand Greengrass devices, and then choose Groups (V1).

2. Under Greengrass groups, choose your group.

3. On the Lambdas page, choose the ellipses (…) for the TransferStream function, and then choose Remove function.

4. From Actions, choose Deploy.