在 IoT Analytics 中解析 JSON 数组输入

Question

我一次从频道中的物联网设备接收多个数据记录作为 JSON 数组。收到的消息如下所示：

[
{
"Field1": "Value1",
"Field2": "Value2",
"Field3": "Value3"
},
{
"Field1": "AnotherValue1",
"Field2": "AnotherValue2",
"Field3": "AnotherValue3"
}
]

我使用以下SQL查询创建一个数据集：

SELECT * FROM mydatastore

当我运行数据集时，返回的结果是：

array                                              __dt 
-----                                              -----
[{field1=Value1, field2=Value2, field3=Value3}]    2019-02-21 00:00:00.000

我想要的结果是：

Field1           Field2           Field3
------           ------           ------
Value1           Value2           Value3
AnotherValue1    AnotherValue2    AnotherValue3

如何让 IoT Analytics 在数据存储中为收到的 JSON 数组中的每个元素创建一个新行？

Answer 1

如何让 IoT Analytics 在数据存储中为收到的 JSON 数组中的每个元素创建一个新行？

最简单的方法应该是利用管道上的 Lambda 活动，并让它将单个 JSON 有效负载解析为所需的结构。这在一定程度上取决于发送到通道的消息的"原始"结构。

因此，例如，我们可以通过 CLI 批处理放置消息将数据发送到通道，如下所示：

aws iotanalytics batch-put-message --channel-name sample_channel --messages '[{"messageId": "message1", "payload": "{"array": [{"Field1": "Value1", "Field2": "Value2", "Field3": "Value3"},{"Field1": "AnotherValue1", "Field2": "AnotherValue2", "Field3": "AnotherValue3"}]}"}]'

然后，通道将具有一条结构如下的消息：

{
"messageId": "message1",
"payload": {
"array": [
{
"Field1": "Value1",
"Field2": "Value2",
"Field3": "Value3"
},
{
"Field1": "AnotherValue1",
"Field2": "AnotherValue2",
"Field3": "AnotherValue3"
}
]
}
}

如果您的管道具有 Lambda 活动，则来自通道的消息将在event参数中传递到您的 Lambda 函数。

我使用 AWS Lambda控制台内联编辑器创建了一个简单的 Lambda 函数(使用 Python 3.7)，并将其命名为sample_lambda：

import json
import sys
import logging
# Configure logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
streamHandler = logging.StreamHandler(stream=sys.stdout)
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
streamHandler.setFormatter(formatter)
logger.addHandler(streamHandler)

def lambda_handler(event, context):
# This can be handy to see the raw structure of the incoming event
# will log to the matching CloudWatch log:
# /aws/lambda/<name_of_the_lambda>
# logger.info("raw event: {}".format(event))
parsed_rows = []
# Depending on the batchSize setting of the Lambda Pipeline Activity,
# you may receive multiple messages in a single event
for message_payload in event:
if 'array' in message_payload:
for row in message_payload['array']:
parsed = {}
for key, value in row.items():
parsed[key] = value
parsed_rows.append(parsed)
return parsed_rows

我添加了适当的权限，以便物联网分析可以通过CLI调用lambda函数：

aws lambda add-permission --function-name sample_lambda --statement-id statm01 --principal iotanalytics.amazonaws.com --action lambda:InvokeFunction

重新处理管道，解析的行被放置在数据存储中;执行数据集，我得到以下最终结果：

"array","field1","field2","field3","__dt"
,"Value1","Value2","Value3","2019-04-26 00:00:00.000"
,"AnotherValue1","AnotherValue2","AnotherValue3","2019-04-26 00:00:00.000"

Answer 2

迟到的回复：

只是想想，物联网规则 - select语句有很多内置的子句/函数，如"as"能够转换输入有效载荷。 请检查，可能会解决。

示例 - https://docs.aws.amazon.com/iot/latest/developerguide/iot-sql-select.html

在主题"主题/子主题"上发布的传入有效负载：{"颜色"：{"红色"：255，"绿色"：0，"蓝色"：0}，"温度"：50}

.SQL：

SELECT color.red as red_value FROM 'topic/subtopic' Outgoing payload: {"red_value":255}