使用 kafka-streams 创建一个包含多个聚合的新 KStream

Question

我正在发送 JSON 消息，其中包含有关 Web 服务请求和对 Kafka 主题的响应的详细信息。我想使用 Kafka 流处理到达 Kafka 的每条消息，并将结果作为持续更新的摘要（JSON 消息）发送到客户端连接的 websocket。

然后客户端将解析 JSON 并在网页上显示各种计数/摘要。

输入消息示例如下

{
  "reqrespid":"048df165-71c2-429c-9466-365ad057eacd",
  "reqDate":"30-Aug-2017",
  "dId":"B198693",
  "resp_UID":"N",
  "resp_errorcode":"T0001",
  "resp_errormsg":"Unable to retrieve id details. DB Procedure error",
  "timeTaken":11,
  "timeTakenStr":"[0 minutes], [0 seconds], [11 milli-seconds]",
  "invocation_result":"T"
}

{
  "reqrespid":"f449af2d-1f8e-46bd-bfda-1fe0feea7140",
  "reqDate":"30-Aug-2017",
  "dId":"G335887",
  "resp_UID":"Y",
  "resp_errorcode":"N/A",
  "resp_errormsg":"N/A",
  "timeTaken":23,
  "timeTakenStr":"[0 minutes], [0 seconds], [23 milli-seconds]",
  "invocation_result":"S"
}

{
  "reqrespid":"e71b802d-e78b-4dcd-b100-fb5f542ea2e2",
  "reqDate":"30-Aug-2017",
  "dId":"X205014",
  "resp_UID":"Y",
  "resp_errorcode":"N/A",
  "resp_errormsg":"N/A",
  "timeTaken":18,
  "timeTakenStr":"[0 minutes], [0 seconds], [18 milli-seconds]",
  "invocation_result":"S"
}

随着消息流进入 Kafka，我希望能够即时计算

**

• 请求总数，即所有请求的计数
• invocation_result 等于“S”的请求总数
• invocation_result 不等于“S”的请求总数
• invocation_result 等于“S”和 UID 的请求总数 等于“Y”
• invocation_result 等于“S”和 UID 的请求总数 等于“Y”
• 最短时间，即 min(timeTaken)
• 所用的最大时间，即 max(timeTaken)
• 平均花费时间，即 avg(timeTaken)

**

并将它们写到一个 KStream 中，新键设置为 reqdate 值和新值一个 JSON 消息，其中包含计算值，如下所示，使用前面显示的 3 条消息

{
  "total_cnt":3, "num_succ":2, "num_fail":1, "num_succ_data":2, 
  "num_succ_nodata":0, "num_fail_biz":0, "num_fail_tech":1,
  "min_timeTaken":11, "max_timeTaken":23, "avg_timeTaken":17.3
}

我是 Kafka 流的新手。我如何进行多次计数并通过不同的列全部在一个或作为一系列不同的步骤？ Apache flink 或 calcite 是否更合适，因为我对 KTable 的理解表明您只能拥有一个密钥，例如30-AUG-2017，然后是单个列值，例如计数说 3。我需要一个带有一个键和多个计数值的结果表结构。

非常感谢所有帮助。

Answer 1

您可以执行一个复杂的聚合步骤，一次计算所有这些。我只是在勾勒这个想法：

class AggResult {
    long total_cnt = 0;
    long num_succ = 0;
    // and many more
}

stream.groupBy(...).aggregate(
    new Initializer<AggResult>() {
        public AggResult apply() {
            return new AggResult();
        }
    },
    new Aggregator<KeyType, JSON, AggResult> {
        AggResult apply(KeyType key, JSON value, AggResult aggregate) {
            ++aggregate.total_cnt;
            if (value.get("success").equals("true")) {
                ++aggregate.num_succ;
            }
            // add more conditions to get all the other aggregate results
            return aggregate;
        }
    },
    // other parameters omitted for brevity
)
.to("result-topic");