如何使用KSQL从变更事件流重建实体 支持删除后重建场景

原有latest_by_offset直接按id分组的方案缺陷在于没有识别delete事件的状态截断作用，删除后重建的同id实体会复用旧实体的历史字段值，要覆盖创建、更新、删除、删后重建全场景，可按以下步骤实现：

1. 标记实体生命周期边界 #

delete事件标志着对应id的当前实体生命周期终结，后续同id的create事件属于全新的实体实例。首先创建衍生流，为每个事件绑定所属的实体版本号，区分同id下的不同生命周期实体：

CREATE STREAM customer_lifecycle AS
SELECT
  event,
  content,
  content['id'] AS customer_id,
  -- 统计当前id历史累计delete次数，同版本号的事件属于同一个实体实例
  COUNT(CASE WHEN event = 'delete' THEN 1 END) OVER (
    PARTITION BY content['id']
    ORDER BY ROWTIME
    ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
  ) - CASE WHEN event = 'delete' THEN 1 ELSE 0 END AS entity_version
FROM customerstream
EMIT CHANGES;

逻辑说明：以删除后重建的事件序列为例，id=1的create、update事件累计delete计数为0，entity_version为0；delete事件之后的新create事件累计delete计数为1，entity_version为1，从分组层面就把删除前后两个同id实体做了隔离。

2. 聚合生成实体最新状态表 #

使用customer_id + entity_version作为联合分组键，聚合时过滤delete事件，只保留有效实体的最新字段值，创建持久化的状态表：

CREATE TABLE customer_current_state WITH (
  KEY_FORMAT = 'JSON',
  VALUE_FORMAT = 'JSON'
) AS
SELECT
  customer_id,
  entity_version,
  LATEST_BY_OFFSET(content['name'], true) AS name,
  LATEST_BY_OFFSET(content['location'], true) AS location
FROM customer_lifecycle
WHERE event != 'delete'
GROUP BY customer_id, entity_version
EMIT CHANGES;

LATEST_BY_OFFSET第二个参数传入true表示自动忽略null值，update事件仅传部分字段时，会自动保留该字段上一次的有效值，和事件溯源的合并逻辑一致。

3. 场景验证 #

• 常规创建+更新场景：同entity_version下自动取各字段最新偏移量的值，可正确输出预期的{name:'bob_new', location:'BER', id:1}结果
• 删除场景：delete事件仅用于切分版本，不会写入最终状态表，对应旧实体会从结果中移除
• 删除后重建场景：新create事件归属更高版本的entity_version，作为全新实体聚合，不会继承删除前的旧字段值，删后重建场景最终输出{name:'new_person', id:1}，符合预期。

如果需要输出和原始流一致的MAP格式content字段，可直接查询状态表做结构转换：

SELECT
  customer_id AS id,
  MAP(
    ARRAY['name', 'location'],
    ARRAY[name, location]
  ) AS content
FROM customer_current_state
EMIT CHANGES;

内容的提问来源于stack exchange，提问作者user3579222