Kafka将流作为表补丁日志，而不是完整发布

Question

期望的功能：对于给定的密钥key123，许多服务并行运行，并将其结果报告给单个位置，一旦为key123收集了所有结果，就会将其传递给新的下游消费者。

最初的想法：使用AWS DynamoDB保存给定条目的所有结果。每次结果准备就绪时，微服务都会对key123上的数据库执行PATCH操作。输出流检查每个UPDATE以查看条目是否完整，如果是，则将其转发到下游。

新想法：使用Kafka Streams和KSQL实现相同的目标。所有服务都将其输出写入结果主题，该主题形成一个更改日志Kstream，我们可以通过KSQL查询完整的条目。比如：

CREATE STREAM competed_results FROM results_stream SELECT * WHERE (all results != NULL). 

我不确定该怎么做的部分是流上的补丁操作。要让输出流显示key123的所有消息的累积，而不是仅显示最近的消息？

KSQL用户，这有意义吗？我是否接近于一个以前有人做过的解决方案？

Answer 1

如果您可以使用密钥集为同一主题生成所有事件，那么您可以使用ksqlDB中的聚合来收集特定密钥的所有事件，例如：

CREATE STREAM source (
    KEY INT KEY,  -- example key to group by
    EVENT STRING  -- example event to collect
  ) WITH (
   kafka_topic='source', -- or whatever your source topic is called.
   value_format='json' -- or whatever value format you need.
);

CREATE TABLE agg AS
  SELECT 
    key, 
    COLLECT_LIST(event) as events
  FROM source
  GROUP BY key;

默认情况下，这将创建一个名为AGG的变更日志主题。当收到源主题上特定键的新事件时，ksqlDB将向AGG主题生成消息，该键设置为key，该值包含该键看到的所有事件的列表。

然后，您可以将此变更日志作为流导入：

CREATE STREAM agg_stream (
   KEY INT KEY,
   EVENTS ARRAY<STRING> 
) WITH (
   kafka_topic='AGG',
   value_format='json'
);

然后，您可以应用一些标准来过滤流，只包括最终结果：

STREAM competed_results AS 
  SELECT 
    * 
  FROM agg_stream 
  WHERE ARRAY_LEN(EVENTS) = 5; -- example 'complete' criteria.

您甚至可能希望使用用户定义的函数来定义您的完整标准：

STREAM competed_results AS 
  SELECT 
    * 
  FROM agg_stream 
  WHERE IS_COMPLETE(EVENTS);