使用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（timeTake）
• 花费的最大时间，即max（time）
• 平均花费时间，即平均花费时间

**

并将它们写入一个KStream，其中new key设置为reqdate值，new value设置为一个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或方解石更合适吗？因为我对KTable的理解表明，您只能有一个键，例如2017年8月30日，然后是一个单列值，例如计数，比如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");