某些 x 条消息中的 KSQL 平均值

Question

我有一个正在运行的 kafka-confluent 实例和一个名为“mqtt-source-1”的主题，它读取这个结构化的 json 消息

{
    "device1": {
        "t": "timestamp",
        "id": "deviceid",
        "type": "presence",
        "m": [
            {
                "t": "nowTimestamp()",
                "tz": "now()",
                "k": "device_temperature",
                "v": "double(0, 40)",
                "u": "C"
            },
            {
                "t": "nowTimestamp()",
                "tz": "now()",
                "k": "battery_level",
                "v": "double(3.26, 3.56)",
                "u": "V"
            }
        ]
    },
    "device2": {
        "t": "timestamp",
        "id": "deviceid",
        "type": "presence",
        "m": [
            {
                "t": "nowTimestamp()",
                "tz": "now()",
                "k": "device_temperature",
                "v": "double(0, 40)",
                "u": "C"
            },
            {
                "t": "nowTimestamp()",
                "tz": "now()",
                "k": "battery_level",
                "v": "double(3.26, 3.56)",
                "u": "V"
            }
        ]
    }
}

是否可以得到按 m->k 分组的 v 的平均值，以及对特定 x 条消息计算的 deviceid 的平均值？

Answer 1

如果还没有完成，您需要创建一个能够正确解析传入消息的主题。这篇博文描述了如何在 JSON 中处理 JSON 和数组：https://rmoff.net/2020/05/26/working-with-json-nested-arrays-in-ksqldb-example/

现在，如果我正确理解了您的最后一句话，您希望按 {m->k, deviceid} 聚合。您可以使用CREATE TABLE <<name>> AS SELECT <<your desired columns>> FROM <<input stream that contains the parsed JSONs>> 语句执行此操作。使用纯 SQL 语法按 {m->k, deviceid} 分组并创建应用 AVG(m->v) 的列。应该这样做。