如何在 AWS IoT TwinMaker 中使用您自己的数据源 官方博客

在 AWS IoT TwinMaker 中使用自有数据源的指南

文作者：Ali Benfattoum，发表于 2022 年 10 月 18 日，链接：AWS IoT TwinMaker，技术操作指南永久链接 分享

关键要点

引言

AWS IoT TwinMaker 为开发者提供了便捷的工具，用于创建实际系统的数字双胞胎，如建筑和工厂，且能够利用来自多种来源的现有数据。它采用了连接器架构，用户可以将自己的数据源连接到 AWS IoT TwinMaker，而无需将数据重新获取或迁移到其他位置。AWS IoT TwinMaker 为 AWS 服务提供了内置的数据连接器，例如 AWS IoT SiteWise 和 Amazon Kinesis Video Streams。同时，用户还可以创建自定义数据连接器，以便与其他 AWS 或第三方数据源如 Amazon Timestream、Amazon DynamoDB、Snowflake 和 Siemens Mindsphere一起使用。

在这篇博客中，您将学习如何使用 AWS IoT TwinMaker 数据连接器接口将自己的数据源集成进来。

概述

数据源与 AWS IoT TwinMaker 之间的连接在 组件 中有所描述。组件通过 Lambda 连接器访问外部数据源。Lambda 连接器是您在组件定义中指定的 Lambda 函数。

以下是使用 Schema initializer connector 创建 Amazon DynamoDB 数据连接器的步骤，该连接器从底层数据源获取属性，并使用 DataReader connector 获取这些属性的时间序列值。一旦数据连接器创建完成，您将获得有关如何为该数据连接器创建组件并将其附加到实体的指导。

在本文中，我们将使用 Amazon DynamoDB 作为数据源，但这里描述的概念适用于其他任何数据源。

前提条件

在本博客中设置和执行步骤之前，您需要：

操作步骤

在本操作中，您将执行 6 个步骤，将 Amazon DynamoDB 数据源连接到 AWS IoT TwinMaker：

第一步：创建 DynamoDB 表

为了本帖子，您将创建一个名为 TwinMakerTable 的 DynamoDB 表，其分区键 thingName 的类型为 String，排序键 timestamp 的类型为 Number。有关如何创建 DynamoDB 表的更多信息，请参见 如何创建 DynamoDB 表。

您创建的表将存储传感器的空气质量测量值。为了简单起见，此帖子中将创建对应于传感器通过其名称存储作为分区键 thingName测量的 5 个项目。每个测量项将拥有以下类型为 Number 的属性：timestamp存储为排序键 timestamp，即测量的 Unix 时间戳 在 毫秒 中；temperature、humidity 和 co2。

让我们在表中插入 5 个项目，对应于传感器 airTwin 的 5 个测量值。您可以从 网站 获取当前时间戳以毫秒为单位，然后通过每次减少 10000 得到 5 个时间戳。接着您可以为属性：temperature、humidity 和 co2 输入随机值。有关更多信息，请参见 通过控制台向表写入数据。

现在您已经创建了包含数据的表，接下来将创建两个 Lambda 函数：一个用于 Schema initializer connector，另一个用于 DataReader connector。

第二步：创建 Schema initializer connector

Schema initializer connector 是在组件类型或实体生命周期中使用的 Lambda 函数，用于从底层数据源获取组件类型或组件属性。您将创建一个 Lambda 函数，以返回 TwinMakerTable 的模式。

您可以使用 Lambda 控制台创建 Nodejs Lambda 函数。

javascriptexportshandler = async (event) =gt { let result = { properties { temperature { definition { dataType { type DOUBLE } isTimeSeries true } } humidity { definition { dataType { type DOUBLE } isTimeSeries true } } co2 { definition { dataType { type DOUBLE } isTimeSeries true } }

      }    }return result

}

这个函数返回我们表中每个属性的定义，并指定其类型。在这种情况下，所有属性的类型都是 DOUBLE，且是时间序列数据。您可以在 文档 中查看有效类型。

注意：这里的属性在函数中硬编码。您可以设计一个函数，例如从条目中自动检索属性及其类型。

现在，让我们创建 DataReader connector。

第三步：创建 DataReader connector

DataReader 是一种数据平面连接器，用于获取单个组件中属性的时间序列值。

您可以使用 Lambda 控制台创建 Nodejs Lambda 函数。

javascriptconst TABLE = TwinMakerTableconst aws = require(awssdk)const dynamo = new awsDynamoDBDocumentClient()

exportshandler = async (event) =gt { try { let {workspaceId entityId componentName selectedProperties startTime endTime } = event

    // 查询数据库以获取所选的属性    const {Items} = await dynamoquery({        TableName TABLE        ProjectionExpression {selectedProperties} #tmsp        KeyConditionExpression thingName = hashKey AND #tmsp BETWEEN startTime AND endTime        ExpressionAttributeNames {            #tmsp timestamp        }        ExpressionAttributeValues {            hashKey entityId            startTime (new Date(startTime))getTime()             endTime (new Date(endTime))getTime()         }    })promise()    let results = { propertyValues [] }    let res = []    ItemsforEach(item =gt {        selectedPropertiesforEach(prop =gt {            if(!res[prop]){                res[prop] = {                    entityPropertyReference{                        propertyName prop                        componentName                        entityId evententityId                    }                    values []                }            }            res[prop]valuespush({                time (new Date(item[timestamp]))toISOString()                value {doubleValue item[prop]}            })        })    })    for (let key in res){        resultspropertyValuespush(res[key])    }    consolelog(results)    return results} catch (e) {    consolelog(e)}

}

TwinMaker 组件将使用此 DataReader connector 从 DynamoDB 表中获取数据。组件在请求中提供两个属性：startTime 和 endTimeISO8601 时间戳格式，这些属性可用于连接器仅获取该时间范围内的数据。您可以在文档的 数据连接器 部分中查看请求和响应界面。

在移动到下一步之前，您需要授予函数访问该表的权限。有关更多信息，请参见 允许 Lambda 函数访问 Amazon DynamoDB 表。

现在您可以进入创建 AWS IoT TwinMaker 工作区的步骤。

第四步：在 AWS IoT TwinMaker 中创建工作区

在 AWS IoT TwinMaker 控制台 中，创建一个名为 AirWorkspace 的工作区。您可以按照 AWS IoT TwinMaker 文档中的 创建工作区 部分中的说明进行操作。

工作区创建完成后，您应该有一个 Amazon 简单存储服务 (Amazon S3) 桶被创建。AWS IoT TwinMaker 将使用这个桶来存储与工作区相关的信息和资源。

您还应该有一个 IAM Identity Center 角色被创建。该角色允许工作区代表您访问其他服务中的资源。

在创建组件之前，您必须向工作区角色提供调用之前创建的两个 Lambda 函数的权限。请参见 允许连接器访问外部数据源的权限 以了解如何将权限授予服务角色以使用 Lambda 函数。

json{ Version 20121017 Statement [ { Sid VisualEditor0 Effect Allow Action lambdaInvokeFunction Resource [ arnawslambda{{AWSREGION}}{{ACCOUNTID}}functionTwinMakerDynamoDataReader arnawslambda{{AWSREGION}}{{ACCOUNTID}}functionTwinMakerDynamoSchemaInit ] } ]}

您现在可以创建组件了。

第五步：创建 AWS IoT TwinMaker 组件

选择您创建的工作区。在工作区中，选择 组件类型，然后选择 创建组件类型。

在 请求 部分中复制以下 JSON 文档，并将之前创建的 DataReader 和 Schema initializer 函数的 ARN 替换掉：

json{ componentTypeId comdynamodbairQuality description Connector for DynamoDB Use case Air Quality propertyDefinitions{ } functions { dataReader { implementedBy { lambda { arn arnawslambda{{AWSREGION}}{{ACCOUNTID}}functionTwinMakerDynamoDataReader } } } schemaInitializer { implementedBy { lambda { arn arnawslambda{{AWSREGION}}{{ACCOUNTID}}functionTwinMakerDynamoSchemaInit } } } }}

选择 创建组件类型。组件创建完成后，您可以创建一个实体以测试该组件。

第六步：创建实体并测试组件

您现在将创建一个实体并将您创建的组件附加到此实体。

此组件与 ID 为 airTwin 的实体关联。现在只剩下最后一步，即测试组件。当测试组件时或调用 GetPropertyValueHistory API 操作时，组件会向 DataReader Lambda connector 发送请求，包括实体 ID。Lambda connector 将使用该 ID 查询与此 ID 对应的传感器测量值。在这种情况下，它将获取传感器 airTwin 的测量值。

您应该能在 时间序列结果 部分看到传感器的测量值。

![TwinMaker 中的时间序列

在AWS上通过多播技术监控监控摄像头视频 网络与内容分发

2026-01-27 13:37:54

新的 AWS 云技术顾问专业证书现已在 Coursera 上推出 培训与认证博客

2026-01-27 14:22:27