Beam中如何用PCollection<PubsubMessage>做侧输入并处理死信队列

问题根源 #

你之前用View.asList()处理无界Pubsub消息的PCollection时报错，核心原因是：无界流在全局窗口中执行GroupByKey（View.asList()底层依赖该操作）必须配置触发器，但默认全局窗口没有触发器，导致无法完成聚合。更关键的是，这种全局sideInput的方式根本无法对应到Transformation2中失败的那条消息的原始内容——全局视图会把所有消息存成一个列表，你没法找到当前失败消息对应的原始数据。

可行解决方案 #

方案1：携带原始消息贯穿所有转换步骤

最直接的方式是在每一步转换中，把原始PubsubMessage和转换结果绑定在一起（用KV或者自定义类），这样无论哪一步失败，都能直接拿到对应的原始消息发送到死信队列。

示例代码：

// 定义死信队列的TupleTag
TupleTag<PubsubMessage> deadLetterTag = new TupleTag<>() {};
TupleTag<KV<PubsubMessage, IntermediateResult>> t1SuccessTag = new TupleTag<>() {};

// 第一步转换：绑定原始消息，失败则输出到死信
PCollectionTuple t1Result = message.apply(
    "Transformation1",
    ParDo.of(new DoFn<PubsubMessage, KV<PubsubMessage, IntermediateResult>>() {
        @ProcessElement
        public void processElement(ProcessContext c) {
            PubsubMessage originalMsg = c.element();
            try {
                // 执行Transformation1的业务逻辑
                IntermediateResult t1Output = yourTransformation1Logic(originalMsg);
                c.output(KV.of(originalMsg, t1Output));
            } catch (Exception e) {
                // Transformation1失败，发送原始消息到死信队列
                c.sideOutput(deadLetterTag, originalMsg);
            }
        }
    }).withOutputTags(t1SuccessTag, TupleTagList.of(deadLetterTag))
);

// 提取Transformation1的成功结果，执行第二步转换
TupleTag<KV<PubsubMessage, TableRow>> t2SuccessTag = new TupleTag<>() {};
PCollectionTuple t2Result = t1Result.get(t1SuccessTag).apply(
    "Transformation2",
    ParDo.of(new DoFn<KV<PubsubMessage, IntermediateResult>, KV<PubsubMessage, TableRow>>() {
        @ProcessElement
        public void processElement(ProcessContext c) {
            KV<PubsubMessage, IntermediateResult> input = c.element();
            PubsubMessage originalMsg = input.getKey();
            IntermediateResult t1Data = input.getValue();
            try {
                // 执行Transformation2的业务逻辑
                TableRow t2Output = yourTransformation2Logic(t1Data);
                c.output(KV.of(originalMsg, t2Output));
            } catch (Exception e) {
                // Transformation2失败，发送原始消息到死信队列
                c.sideOutput(deadLetterTag, originalMsg);
            }
        }
    }).withOutputTags(t2SuccessTag, TupleTagList.of(deadLetterTag))
);

// 合并两步转换产生的死信消息
PCollection<PubsubMessage> allDeadLetterMessages = PCollectionList.of(
    t1Result.get(deadLetterTag),
    t2Result.get(deadLetterTag)
).apply(Flatten.pCollections());

// 将死信消息写入Pubsub死信队列
allDeadLetterMessages.apply(
    "Write to Dead Letter Queue",
    PubsubIO.writeMessages().to("projects/your-project/topics/dead-letter-topic")
);

// 提取成功结果进行保存
PCollection<TableRow> validTableRows = t2Result.get(t2SuccessTag)
    .apply("Extract TableRow", ParDo.of(new DoFn<KV<PubsubMessage, TableRow>, TableRow>() {
        @ProcessElement
        public void processElement(ProcessContext c) {
            c.output(c.element().getValue());
        }
    }));
// 执行保存逻辑
validTableRows.apply("Save to Target", yourSaveLogic());

方案2：使用Try类型包装结果（更优雅的错误区分）

可以用Beam的Try类（或者自定义类似的包装类）来区分成功/失败的结果，同时携带原始消息，让流程更清晰：

// 自定义包装类，携带原始消息和转换结果
class ProcessResult<T> {
    private final PubsubMessage originalMsg;
    private final Try<T> result;

    public ProcessResult(PubsubMessage originalMsg, Try<T> result) {
        this.originalMsg = originalMsg;
        this.result = result;
    }

    // getter方法
    public PubsubMessage getOriginalMsg() { return originalMsg; }
    public Try<T> getResult() { return result; }
}

// Transformation1处理
PCollection<ProcessResult<IntermediateResult>> t1ProcessResults = message.apply(
    "Transformation1",
    ParDo.of(new DoFn<PubsubMessage, ProcessResult<IntermediateResult>>() {
        @ProcessElement
        public void processElement(ProcessContext c) {
            PubsubMessage original = c.element();
            Try<IntermediateResult> tryResult = Try.of(() -> yourTransformation1Logic(original));
            c.output(new ProcessResult<>(original, tryResult));
        }
    })
);

// 过滤Transformation1的失败消息，送入死信
PCollection<PubsubMessage> dlqFromT1 = t1ProcessResults.apply(
    "Filter T1 Failures",
    ParDo.of(new DoFn<ProcessResult<IntermediateResult>, PubsubMessage>() {
        @ProcessElement
        public void processElement(ProcessContext c) {
            ProcessResult<IntermediateResult> res = c.element();
            res.getResult().ifFailure(failure -> c.output(res.getOriginalMsg()));
        }
    })
);

// 处理Transformation1的成功结果，进入Transformation2
PCollection<ProcessResult<TableRow>> t2ProcessResults = t1ProcessResults.apply(
    "Filter T1 Successes",
    ParDo.of(new DoFn<ProcessResult<IntermediateResult>, ProcessResult<TableRow>>() {
        @ProcessElement
        public void processElement(ProcessContext c) {
            ProcessResult<IntermediateResult> input = c.element();
            Try<TableRow> tryResult = Try.of(() -> yourTransformation2Logic(input.getResult().get()));
            c.output(new ProcessResult<>(input.getOriginalMsg(), tryResult));
        }
    })
);

// 过滤Transformation2的失败消息，送入死信
PCollection<PubsubMessage> dlqFromT2 = t2ProcessResults.apply(
    "Filter T2 Failures",
    ParDo.of(new DoFn<ProcessResult<TableRow>, PubsubMessage>() {
        @ProcessElement
        public void processElement(ProcessContext c) {
            ProcessResult<TableRow> res = c.element();
            res.getResult().ifFailure(failure -> c.output(res.getOriginalMsg()));
        }
    })
);

// 合并死信消息并写入
PCollection<PubsubMessage> allDeadLetterMessages = PCollectionList.of(dlqFromT1, dlqFromT2)
    .apply(Flatten.pCollections());
allDeadLetterMessages.apply("Write DLQ", PubsubIO.writeMessages().to("your-dlq-topic"));

为什么之前的SideInput方法不可行 #

无界PCollection使用View.asList()时，Beam会在全局窗口中执行GroupByKey操作来聚合所有消息，但无界流的全局窗口默认没有触发器——Beam不知道什么时候应该停止聚合并生成视图，因此报错。就算你配置了触发器，这种全局视图也无法对应到Transformation2中具体失败的那条消息的原始内容，完全不符合你的需求。

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