流分析可以过滤数组属性的项目吗？

Question

您好，我想知道是否可以从流分析的 JSON 输入的数组属性中选择某些项目并将它们作为 JSON 输出的数组属性返回。

为了更清楚，我的示例 - 我发送了一个在具有 name、version 和 state 的设备上运行的 OSGI bundles 列表的捆绑。 （我省略了其余内容。）示例消息：

{"bundles":[{"name":"org.eclipse.osgi","version":"3.5.1.R35x_v20090827","state":32},{"name":"slf4j.log4j12","version":"1.6.1","state":4}]}

通过流分析，我想为活动包（状态 == 32）创建一个 JSON 输出（事件中心），并将其余的放在不同的输出中。这些事件中心的内容将在稍后处理。但在处理过程中，我还需要原始的设备 ID，所以我从 IoTHub 消息属性中获取它。

所以我的查询看起来像这样：

WITH Step1 AS
(
SELECT
    IoTHub.ConnectionDeviceId AS deviceId,
    bundles as bundles
FROM 
    iotHubMessages
)

SELECT
    messages.deviceId AS deviceId,
    bundle.ArrayValue.name AS name,
    bundle.ArrayValue.version AS version
INTO
    active
FROM 
    Step1 as messages
CROSS APPLY GetArrayElements(messages.bundles) AS bundle
WHERE 
    bundle.ArrayValue.state = 32

SELECT
    messages.deviceId AS deviceId,
    bundle.ArrayValue.name AS name,
    bundle.ArrayValue.version AS version
INTO
    other
FROM 
    Step1 as messages
CROSS APPLY GetArrayElements(messages.bundles) AS bundle
WHERE 
    bundle.ArrayValue.state != 32

这样，在 active 中包含 deviceId、name 和 version 属性的原始数组的每个项目都有一行 输出。因此 deviceId 属性被复制了多次，这意味着消息中的附加数据。我更喜欢具有一个 deviceId 属性和一个数组属性 bundles 的 JSON，类似于原始 JSON 输入。

喜欢活跃：

{"deviceid":"javadevice","bundles":[{"name":"org.eclipse.osgi","version":"3.5.1.R35x_v20090827"}]}

还有其他：

{"deviceid":"javadevice","bundles":[{"name":"slf4j.log4j12","version":"1.6.1"}]}

有什么方法可以实现吗？ - 过滤数组项并将其作为数组返回，格式与输入中的格式相同。 （在我的代码中，我更改了属性的数量，但这不是必需的。）

感谢您的任何想法！

Answer 1

我认为您可以使用 Collect() 聚合函数来实现这一点。 我看到的唯一问题是 deviceId 属性也将输出到 bundle 数组中。

WITH Step1 AS
(
SELECT
    IoTHub.ConnectionDeviceId AS deviceId,
    bundles as bundles
FROM 
    iotHubMessages
),
Step2 AS 
(
SELECT
    messages.deviceId AS deviceId,
    bundle.ArrayValue.name AS name,
    bundle.ArrayValue.version AS version
    bundle.ArrayValue.state  AS state
FROM 
    Step1 as messages
CROSS APPLY GetArrayElements(messages.bundles) AS bundle
)

SELECT deviceId, Collect() AS bundles
FROM Step2
GROUP BY deviceId, state, System.Timestamp
WHERE state = 32