将数据流导出到Amazon Web Services 云(CLI)

下载Amazon IoT Greengrass适用于 Python 的核心v1.5.0 或更高版本。

解压缩下载的程序包以获取软件开发工具包。软件开发工具包是 greengrasssdk 文件夹。

安装程序包依赖项以将其包含在 Lambda 函数部署程序包的开发工具包中。

将以下 Python 代码函数保存在名为 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())

将以下项目压缩到名为 transfer_stream_python.zip 的文件中。此即您的 Lambda 函数部署软件包。

创建 zip 文件时，仅包含这些项目，而不是包含文件夹。

创建 IAM 角色，以便您可以在创建该函数时传入角色 ARN。

从输出中复制 Arn。

使用 Amazon Lambda API 创建 TransferStream 函数。以下命令假定该 zip 文件位于当前目录中。

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

发布该函数的版本。

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

为发布的版本创建别名。

Greengrass 组可以按别名（推荐）或版本引用 Lambda 函数。使用别名，您可以更轻松地管理代码更新，因为您无需在函数代码更新时更改订阅表或组定义。相反，您只需将别名指向新的函数版本。

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

从输出中复制 AliasArn。为 Amazon IoT Greengrass 配置函数时，可以使用此值。

使用包含系统和用户定义的 Lambda 函数的初始版本创建函数定义。

以下定义版本使用默认值启用流管理器parameter. 要配置自定义设置，必须为相应的流管理器参数定义环境变量。有关示例，请参阅 。启用、禁用或配置流管理器 (CLI).Amazon IoT Greengrass将为省略的参数使用默认设置。MemorySize应至少为128000.Pinned必须设置为true.

从输出中复制 LatestVersionArn。您将使用此值向部署到核心的组版本添加函数定义版本。

创建一个包含初始版本的记录器定义。

从输出中复制记录器定义的 LatestVersionArn。您将使用此值向部署到核心的组版本添加记录器定义版本。

获取目标 Greengrass 组和组版本的 ID。此过程假定这是最新的组和组版本。以下查询返回最近创建的组。

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

或者，您也可以按名称查询。系统不要求组名称是唯一的，所以可能会返回多个组。

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

从输出中复制目标组的 Id。您将使用此值获取核心定义版本及部署组的时间。

从输出中复制 LatestVersion，这是添加到组的最后一个版本的 ID。您将使用此值获取核心定义版本。

获取核心定义版本的 ARN：

1. 获取组版本。

   • 使用为组复制的 替换 group-idId。

   • Replacegroup-version-id用LatestVersion您为组复制的内容。

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

2. 从输出中复制 CoreDefinitionVersionArn。您将使用此值向部署到核心的组版本添加核心定义版本。

创建组版本。

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

从输出中复制 Version。这是新组版本的 ID。

确保Amazon IoT Greengrass核心正在运行。根据需要在您的 Raspberry Pi 终端中运行以下命令。

1. 要检查守护程序是否正在运行，请执行以下操作：

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

   如果输出包含 root 的 /greengrass/ggc/packages/ggc-version/bin/daemon 条目，则表示守护程序正在运行。

   注意

   路径中的版本取决于您的核心设备上安装的 Amazon IoT Greengrass 核心软件版本。

2. 启动守护程序：

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

创建部署。

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

从输出中复制 DeploymentId。

获取部署状态。

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

如果状态为Success，则部署成功。有关问题排查帮助，请参阅排除 Amazon IoT Greengrass 的故障。

在Amazon Kinesis 控制台中，在Kinesis 数据流，选择MyKinesisStream.

在存储库的MyKinesisStream页面上，选择监控. 如果测试成功，您应在 Put Records (放置记录) 图表中看到数据。根据您的连接，显示数据可能需要一分钟时间。

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

删除TransferStreamLambda 函数来自核心。

1. 按照第 6 步：创建组版本创建新的组版本，但删除 create-group-version 命令中的 --function-definition-version-arn 选项。或者，创建不包括TransferStreamLambda 函数。

   注意

   通过省略系统GGStreamManagerLambda 函数从部署的组版本开始，您将禁用核心上的流管理。

2. 按照步骤 7：创建部署以部署新的组版本。