Apache Beam中如何丢弃不符合条件的消息？

核心问题解答 #

是的，你完全可以通过**跳过调用c.outputWithTimestamp(...)**来丢弃不符合条件的消息，这是Beam中实现数据过滤的原生且高效的方式——只要不输出元素，该消息就不会进入后续的Pipeline步骤，相当于被自动丢弃。

具体实现示例 #

你只需要在Convert to DeviceData的DoFn中，解析完DeviceData后添加过滤判断逻辑，仅对符合条件的消息执行输出操作即可。以下是修改后的代码（示例中假设我们需要保留sensorId不为空且数据标记为有效的消息，你可以根据实际需求替换判断条件）：

pipeline.apply("Read PubSub messages",
                PubsubIO.readStrings().fromSubscription(pubsubSub))

        .apply("Convert to DeviceData",
                ParDo.of(new DoFn<String, KV<String, DeviceData>>() {
                    @Override
                    public Duration getAllowedTimestampSkew() {
                        return new Duration(Long.MAX_VALUE);
                    }

                    @ProcessElement
                    public void processElement(ProcessContext c) {
                        String message = c.element();
                        DeviceData data = new Gson().fromJson(message, DeviceData.class);
                        
                        // 添加过滤逻辑：仅保留符合特定属性的消息
                        if (data.getSensorId() != null && data.isValid()) {
                            String sourceId = data.getSensorId() != null ? data.getSensorId() : data.getFormulaId();

                            // 使用payload中的时间戳
                            Long timeInNanoSeconds = data.getTimeInNanoSeconds();
                            Instant timestamp = ClockUtil.fromNanos(timeInNanoSeconds);
                            long millis = timestamp.toEpochMilli();

                            c.outputWithTimestamp(KV.of(sourceId, data), new org.joda.time.Instant(millis));
                        }
                        // 不符合条件的消息直接跳过，不执行输出，自然被丢弃
                    }
                }))

        .apply("Apply fixed window", window)
        .apply("Group by inputId", GroupByKey.create())
        .apply("Collect created buckets", ParDo.of(new GatherBuckets(options.getWindowSize())))
        .apply("Send to Pub/sub", PubsubIO.writeStrings().to(topic));

可选方案：单独使用Filter转换 #

如果更倾向于将过滤逻辑与转换逻辑分离，也可以在Convert to DeviceData步骤之后添加一个独立的Filter操作，代码职责更清晰：

// 在Convert to DeviceData步骤后新增过滤逻辑
.apply("Filter valid messages", Filter.by(kv -> {
    DeviceData data = kv.getValue();
    // 替换为你的实际过滤条件
    return data.getSensorId() != null && data.isValid();
}))

这种方式会多一次Pipeline转换操作，性能上略逊于在同一个DoFn中合并处理的方式。

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