为什么需要清洗管道
eventd v2 产生的事件数据可能：
- 重复发送：网络超时后重试导致同一条事件发送多次
- 格式不一致：不同接口产生的事件字段差异大
- 脏数据：金额负数、时间戳异常、必填字段缺失
- 数据丢失：部分字段在不同路径下未被填充
清洗管道在 v2 → v3 的数据流中间，解决上述问题。
eventd-cleansing 工程结构
eventd-v3/eventd-cleansing/
├── pom.xml
├── src/main/java/com/plume/eventd/
│ ├── CleansingTopology.java # 拓扑定义
│ ├── EventDeserializer.java # JSON → Event
│ ├── EventSerializer.java # Event → JSON
│ ├── DeduplicateProcessor.java # 去重逻辑
│ ├── NormalizeTransformer.java # 标准化
│ ├── FilterTransformer.java # 异常过滤
│ └── CleansingApplication.java # 启动类
└── src/test/java/com/plume/eventd/
└── CleansingTopologyTest.java # TopologyTestDriver 测试
完整拓扑
public class CleansingTopology {
public Topology build() {
StreamsBuilder builder = new StreamsBuilder();
// 状态存储：去重用。value 存「第一次见到的墙钟时间」，用来做 TTL
StoreBuilder<KeyValueStore<String, Long>> dedupStoreBuilder =
Stores.keyValueStoreBuilder(
Stores.persistentKeyValueStore("dedup-store"),
Serdes.String(),
Serdes.Long()
);
builder.addStateStore(dedupStoreBuilder);
// Source
KStream<String, String> rawStream = builder.stream(
"eventd-raw",
Consumed.with(Serdes.String(), Serdes.String())
);
// 先解析，后续两条分支都基于事件对象
KStream<String, Event> events = rawStream
.mapValues(this::deserialize)
.filter((key, maybeEvent) -> maybeEvent.isPresent())
.mapValues(Optional::get);
// 死信 1：JSON 解析失败（连事件对象都构造不出来）
// DeadLetter 不是 Event，不能复用 serialize(Event)，直接 toJson
rawStream
.filter((key, json) -> deserialize(json).isEmpty())
.mapValues((key, json) -> JsonMapper.toJson(new DeadLetter(
json, "INVALID_JSON", System.currentTimeMillis(), "parse error")))
.to("eventd-dead-letter", Produced.with(Serdes.String(), Serdes.String()));
// 死信 2：能解析但字段不合法。只 filter 不放行会让这类数据静默消失
events
.filter((key, event) -> !event.isValid())
.mapValues((key, event) -> JsonMapper.toJson(new DeadLetter(
JsonMapper.toJson(event), "VALIDATION_FAILED",
System.currentTimeMillis(), "required field missing")))
.to("eventd-dead-letter", Produced.with(Serdes.String(), Serdes.String()));
// 清洗管道
KStream<String, String> cleanedStream = events
.filter((key, event) -> event.isValid())
.processValues(
() -> new DeduplicateProcessor("dedup-store", Duration.ofHours(168)),
"dedup-store"
)
.mapValues(this::normalizeEvent)
.mapValues(this::serialize);
// 输出
cleanedStream.to("eventd-cleaned", Produced.with(Serdes.String(), Serdes.String()));
return builder.build();
}
private Optional<Event> deserialize(String json) {
JsonNode node;
try {
node = JsonMapper.parse(json);
} catch (Exception e) {
// 只有「连 JSON 都解析不了」才返回 empty，对应 INVALID_JSON
return Optional.empty();
}
// 字段缺失不在这里判：交给 isValid()，否则会误记成 INVALID_JSON
// v2 用 timestamp，v3 用 eventTime；两种都要能读
long eventTime = node.hasNonNull("eventTime")
? node.get("eventTime").asLong()
: node.path("timestamp").asLong(0);
return Optional.of(Event.builder()
.eventId(node.path("eventId").asText(""))
.type(node.path("type").asText(""))
.userId(node.path("userId").asText(""))
.amount(node.path("amount").asDouble(0))
.symbol(node.hasNonNull("symbol") ? node.get("symbol").asText() : null)
.eventTime(eventTime)
.version(node.path("version").asInt(1))
.source(node.path("source").asText("eventd-v2"))
.build());
}
private Event normalizeEvent(Event event) {
return Event.builder()
.eventId(event.eventId())
.type(event.type().toUpperCase().trim())
.version(event.version())
.source("eventd-v3")
.eventTime(event.eventTime())
.userId(event.userId().trim().toLowerCase())
.amount(Math.max(0, event.amount())) // 金额非负
.symbol(event.symbol() != null ? event.symbol().trim() : "UNKNOWN")
.status(event.status())
.ipAddress(event.ipAddress())
.device(event.device())
.build();
}
private String serialize(Event event) {
return JsonMapper.toJson(event);
}
}
JsonMapper 是示例对 Jackson ObjectMapper 的薄封装：parse(String) 返回 JsonNode、parse(String, Class<T>) 反序列化、toJson(Object) 序列化；换回标准 Jackson 时对应 readTree、readValue、writeValueAsString。
去重 Processor
Kafka Streams 3.8 里 KStream.transform / transformValues（基于 Transformer / ValueTransformerWithKey）已标记废弃，替代是 process / processValues + FixedKeyProcessor。这里用后者：
public class DeduplicateProcessor
implements FixedKeyProcessor<String, Event, Event> {
private final String storeName;
private final long ttlMillis;
private FixedKeyProcessorContext<String, Event> context;
private KeyValueStore<String, Long> dedupStore;
public DeduplicateProcessor(String storeName, Duration ttl) {
this.storeName = storeName;
this.ttlMillis = ttl.toMillis();
}
@Override
public void init(FixedKeyProcessorContext<String, Event> context) {
this.context = context;
this.dedupStore = context.getStateStore(storeName);
// 每 10 分钟清一次过期键；只靠 process() 里的惰性判断，
// 没再出现的 eventId 会永久占着 RocksDB。
context.schedule(Duration.ofMinutes(10), PunctuationType.WALL_CLOCK_TIME,
now -> {
List<String> expired = new ArrayList<>();
try (KeyValueIterator<String, Long> it = dedupStore.all()) {
while (it.hasNext()) {
var kv = it.next();
if (now - kv.value >= ttlMillis) {
expired.add(kv.key());
}
}
}
for (String key : expired) {
dedupStore.delete(key);
}
});
}
@Override
public void process(FixedKeyRecord<String, Event> record) {
String dedupKey = record.value().eventId();
long now = context.currentSystemTimeMs();
Long seenAt = dedupStore.get(dedupKey);
if (seenAt != null && now - seenAt < ttlMillis) {
// 重复事件：不 forward 就等于丢弃
return;
}
dedupStore.put(dedupKey, now);
context.forward(record);
}
@Override
public void close() {}
}
FixedKeyProcessor 不改变 key，正合适去重这种「要么原样放行、要么丢掉」的场景；需要改 key 时用 Processor 并 forward(record.withKey(...))。去重状态存储在 RocksDB，重启后从 changelog Topic 恢复。
TTL 必须大于「消息可能被重放的最大时间窗口」，否则同一个 eventId 在 TTL 之后重放会被当成新事件。示例的 168 小时是按上游最长重放周期定的，不是通用值。清理用 WALL_CLOCK_TIME 每 10 分钟全量扫一次，键多时这趟扫描不是免费的；更省的做法是按时间分桶，或换用支持 TTL 的外部存储。
清洗规则
| 步骤 | 规则 | 处理方式 |
|---|---|---|
| 反序列化 | JSON 格式错误 | 丢弃，写入死信队列 |
| 字段验证 | eventId/type/userId 缺失 | 丢弃，写入死信队列 |
| 去重 | eventId 已存在 | 丢弃 |
| 标准化 | type 大小写不统一 | 转大写 |
| 标准化 | userId 大小写不统一 | 转小写 |
| 标准化 | amount 负数 | 设为零 |
| 标准化 | symbol 为 null | 设为 “UNKNOWN” |
死信队列
// 死信队列中的数据
public record DeadLetter(
String originalJson,
String reason, // "INVALID_JSON", "VALIDATION_FAILED"
long timestamp,
String errorMessage
) {}
两个死信分支都写在拓扑里（见上）：解析失败保留原始字符串，校验失败保留能解析出的事件 JSON。合成一个分支会让「格式错误」和「字段缺失」混在一起，排查时无法区分。
死信队列的用途：
- 排查数据质量：哪些接口产生了不合规的数据？
- 人工修复：对于重要但格式错误的数据，手动修复后重放
- 监控告警：死信队列消息堆积表示数据质量恶化
启动配置
public class CleansingApplication {
public static void main(String[] args) {
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "eventd-cleansing");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10000);
Topology topology = new CleansingTopology().build();
KafkaStreams streams = new KafkaStreams(topology, props);
streams.start();
Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
}
}
单元测试
@Test
void testCleanPipeline() {
Topology topology = new CleansingTopology().build();
Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-cleansing");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
try (TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {
TestInputTopic<String, String> input = driver.createInputTopic(
"eventd-raw",
new StringSerializer(),
new StringSerializer()
);
TestOutputTopic<String, String> output = driver.createOutputTopic(
"eventd-cleaned",
new StringDeserializer(),
new StringDeserializer()
);
// 输入：有效事件，用 v2 的 timestamp 字段名
input.pipeInput("key1", "{\"eventId\":\"e1\",\"type\":\"order\",\"userId\":\"alice\",\"amount\":100,\"timestamp\":1700000000000}");
// 输入：重复事件（应被去重）
input.pipeInput("key1", "{\"eventId\":\"e1\",\"type\":\"order\",\"userId\":\"alice\",\"amount\":100,\"timestamp\":1700000000000}");
// 输入：无效 JSON（应写入死信）
input.pipeInput("key2", "not-json");
// 断言
List<KeyValue<String, String>> results = output.readKeyValuesToList();
assertEquals(1, results.size()); // 只有第一条被输出
// 验证输出已标准化
Event cleaned = JsonMapper.parse(results.get(0).value, Event.class);
assertEquals("ORDER", cleaned.type()); // 转大写
assertEquals("alice", cleaned.userId()); // 转小写
assertEquals(1700000000000L, cleaned.eventTime()); // timestamp → eventTime
}
}
监控
// 清洗指标
KafkaStreams streams = new KafkaStreams(topology, props);
// 暴露自定义指标（每 30 秒打印）
ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
executor.scheduleAtFixedRate(() -> {
Map<MetricName, ? extends Metric> metrics = streams.metrics();
metrics.forEach((name, metric) -> {
// 线程级指标；名字以 Kafka Streams 实际注册的为准
if (name.name().equals("dropped-records-total")
|| name.name().equals("record-lateness-max")
|| name.name().equals("process-rate")) {
System.out.println(name + " = " + metric.metricValue());
}
});
}, 30, 30, TimeUnit.SECONDS);
小结
| 清洗步骤 | 实现 | 状态 |
|---|---|---|
| 解析 | mapValues(deserialize) | 无 |
| 验证 | filter(isValid) | 无 |
| 去重 | processValues(DeduplicateProcessor) | KeyValueStore |
| 标准化 | mapValues(normalize) | 无 |
| 串行化 | mapValues(serialize) | 无 |
| 死信 | 旁路分支 to(dead-letter) | 无 |
下一节 9.3 Flink 窗口聚合 ：实时指标计算的 Flink DataStream 实现。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。