将数据流导出到 Amazon Web Services 云（控制台）

下载适用于 Python 的 Amazon IoT Greengrass Core 软件开发工具包 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 文件时，仅包含这些项目，而不是包含文件夹。

首先，创建 Lambda 函数。

1. 在Amazon Web Services Management Console中，选择服务，然后打开 Amazon Lambda 控制台。

2. 选择创建函数，然后选择从头开始创作。

3. 在基本信息部分中，使用以下值：

   • 对于函数名称，请输入 TransferStream。

   • 对于运行时系统，选择 Python 3.7。

   • 对于权限，请保留默认设置。这将创建一个授予基本 Lambda 权限的执行角色。此角色不由 Amazon IoT Greengrass 使用。

4. 在页面底部，选择创建函数。

接下来，注册处理程序并上传您的 Lambda 函数部署包。

1. 在代码选项卡上的代码源下，选择上传自。从下拉列表中选择 .zip 文件。

   

2. 选择上传，然后选择您的 transfer_stream_python.zip 部署包。然后，选择保存。

3. 在函数的代码选项卡中，在运行时设置下选择编辑，然后输入以下值。

   • 对于运行时系统，选择 Python 3.7。

   • 对于处理程序，输入 transfer_stream.function_handler。

4. 选择 Save（保存）。

   注意

   Amazon Lambda 控制台上的测试按钮不适用于此功能。Amazon IoT Greengrass Core 软件开发工具包不包含在 Amazon Lambda 控制台中独立运行 Greengrass Lambda 函数所需的模块。这些模块（例如 greengrass_common）是在函数部署到您的 Greengrass 核心之后提供给它们的。

现在，发布 Lambda 函数的第一个版本并创建此版本的别名。

1. 在操作菜单上，选择发布新版本。

2. 对于版本描述，输入 First version，然后选择发布。

3. 在 TransferStream: 1 (TransferStream：1) 配置页面上，从 Actions (操作) 菜单中选择 Create alias (创建别名)。

4. 在创建新别名页面上，使用以下值：

   • 对于名称，输入 GG_TransferStream。

   • 对于版本，选择 1。

   注意

   Amazon IoT Greengrass 不支持 $LATEST 版本的 Lambda 别名。

5. 选择 Create（创建）。

在 Amazon IoT 控制台导航窗格的管理下，展开 Greengrass 设备，然后选择组 (V1)。

选择目标组。

在组配置页面上，选择Lambda 函数选项卡。

在我的 Lambda 函数部分下，选择添加。

在添加 Lambda 函数页面上，为您的 Lambda 函数选择 Lambda 函数。

在 Lambda 版本中，选择 Alias:GG_TransferStream。

现在，配置用于确定 Greengrass 组中 Lambda 函数的行为的属性。

在 Lambda 函数配置部分中，进行以下更改。

选择添加 Lambda 函数。

在组配置页面上，选择Lambda 函数选项卡。

在系统 Lambda 函数下，选择流管理器，然后检查状态。如果禁用，请选择 Edit (编辑)。然后，选择 Enable (启用) 和 Save (保存)。您可以对本教程使用默认参数设置。有关更多信息，请参阅配置 Amazon IoT Greengrass 流管理器。

在 Local logs configuration (本地日志配置) 下，检查是否配置了本地日志记录。

如果未为 Greengrass 系统组件或用户定义的 Lambda 函数配置日志，请选择编辑。

选择用户 Lambda 函数日志级别和 Greengrass 系统日志级别。

保留日志记录级别和磁盘空间限制的默认值，然后选择 Save (保存)。

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

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

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

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

   注意

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

2. 启动进程守护程序：

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

在组配置页面上，选择部署。

1. 在 Lambda 函数选项卡的系统 Lambda 函数部分下，选择 IP 检测器，再选择编辑。

2. 在编辑 IP 检测器设置对话框中，选择自动检测和覆盖 MQTT 代理端点。

3. 选择 Save（保存）。

   这使得设备可以自动获取核心的连接信息，例如 IP 地址、DNS 和端口号。建议使用自动检测，不过 Amazon IoT Greengrass 也支持手动指定的端点。只有在首次部署组时，系统才会提示您选择发现方法。

   注意

   如果出现提示，请授予权限，以创建 Greengrass 服务角色并将其关联至当前 Amazon Web Services 区域中的 Amazon Web Services 账户。该角色将允许 Amazon IoT Greengrass 访问您在 Amazon 服务中的资源。

   部署页面显示了部署时间戳、版本 ID 和状态。完成后，部署的状态应显示为成功完成。

   有关问题排查帮助，请参阅故障排除 Amazon IoT Greengrass。

在 Amazon Kinesis 控制台的 Kinesis data streams 下，选择 MyKinesisStream。

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

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

从核心中删除 TransferStream Lambda 函数。

1. 在 Amazon IoT 控制台导航窗格的管理下，展开 Greengrass 设备，然后选择组 (V1)。

2. 在 Greengrass 组下，选择您的组。

3. 在 Lambdas 页面上，为 TransferStream 函数选择省略号 (…)，然后选择删除函数。

4. 从操作中，选择部署。