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

下载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 文件时，仅包含这些项目，而不是包含文件夹。

首先，创建 Lambda 函数。

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

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

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

   • 对于 Function name（函数名称），请输入 TransferStream。

   • 对于 Runtime (运行时)，选择 Python 3.7。

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

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

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

1. 在存储库的代码选项卡，位于源代码，选择从以下项上传. 从下拉列表中，选择.zip 格式文件.

   

2. 选择上传，然后选择您的transfer_stream_python.zip部署程序包。然后，选择 Save (保存)。

3. 在存储库的代码函数的选项卡，位于Runte 设置，选择编辑，然后输入以下值。

   • 对于 Runtime (运行时)，选择 Python 3.7。

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

4. 选择Save（保存）。

   注意

   这些区域有：测试按钮上的Amazon Lambda控制台不使用此函数。这些区域有：Amazon IoT Greengrass核心开发工具包不包含在Amazon Lambda控制台。这些模块（例如greengrass_common) 将在函数部署到你的 Greengrass 核心后提供给这些函数。

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

1. 在 Actions 菜单上，选择 Publish new version。

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

3. 在存储库的TransferStream：1配置页面，来自操作菜单中，选择创建别名.

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

   • 对于 Name (名称)，请输入 GG_TransferStream。

   • 对于 Version (版本)，选择 1。

   注意

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

5. 选择Create（创建）。

在Amazon IoT控制台导航窗格，在Manage，展开Greengrass 设备，然后选择组 (V1).

选择目标组。

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

UNDER我的 Lambda 函数，选择Add.

在存储库的添加 Lambda 函数页面上，选择Lambda 函数您的 Lambda 函数。

对于Lambda 版本，选择别名:gg_TransferStream.

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

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

选择添加 Lambda 函数.

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

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

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

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

选择用户 Lambda 函数日志级别和Greengrass Systm.

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

确保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

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

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服务。

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

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

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

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

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

删除TransferStreamLambda 函数来自核心。

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

2. UNDERGreengrass 组，选择您的组。

3. 在存储库的Lambdas页面上，选择省略号 (...) 用于TransferStream函数，然后选择移除函数.

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