使用Windows与Triggers的Apache Beam数据流管道异常排查

云上老周

2026-8-10

Apache Beam流处理管道数据不匹配与重复问题排查与修复

问题描述

搭建了一个Apache Beam流处理管道，从Kafka Topic读取JSON数据并转换为类对象，按ID分组后将输出JSON发送至另一个Kafka Topic。当同时向Kafka发送500条以上JSON时，输出数量与输入不符，且存在数据重复，当前使用DirectRunner运行。

测试JSON示例

{
    "transactionId": "1k1",
    "products":[
    {
        "negotiationId": 2002,
        "productGtin": 329794528329,
        "state": 1,
        "observation": "0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023",
        "retailerCode": 9087120
    },
    {
        "negotiationId": 2004,
        "productGtin": 219059901669,
        "state": 2,
        "observation": "0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023",
        "retailerCode": 1312691
    },
    {
        "negotiationId": 2008,
        "productGtin": 111579181771,
        "state": 2,
        "observation": "0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023",
        "retailerCode": 4494522
    },
    {
        "negotiationId": 2009,
        "productGtin": 622670393720,
        "state": 2,
        "observation": "0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023",
        "retailerCode": 5742150
    },
    {
        "negotiationId": 2005,
        "productGtin": 397999498452,
        "state": 1,
        "observation": "0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023",
        "retailerCode": 5075384
    },
    {
        "negotiationId": 2004,
        "productGtin": 161158769517,
        "state": 2,
        "observation": "0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023",
        "retailerCode": 4256911
    },
    {
        "negotiationId": 2009,
        "productGtin": 352265510055,
        "state": 1,
        "observation": "0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023",
        "retailerCode": 2751445
    },
    {
        "negotiationId": 2003,
        "productGtin": 681014619525,
        "state": 2,
        "observation": "0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023-0056-0055-0012-0034-0032-2056-0021-0058-0073-0020-0028-0023",
        "retailerCode": 5359730
    }
     ]
}

原实现代码

public PipelineResult run() {

    Pipeline pipeline = Pipeline.create(options);
    CoderRegistry coderRegistry = pipeline.getCoderRegistry();
    coderRegistry.registerCoderForType(coderPipeline.getCoder().getEncodedTypeDescriptor(), coderPipeline.getCoder());
    
    options.setProject("low-code-exito");
    options.setRegion("us-central1");
    options.setStagingLocation("gs://low-code-exito/binaries/");
    options.setGcpTempLocation("gs://low-code-exito/temp/");
    options.setNetwork("default");
    options.setSubnetwork("regions/us-central1/subnetworks/default");
    options.isStreaming();
    options.setKafkaReadTopics("test-topic");
    options.setReadBootstrapServers("localhost:9092");

    pipeline
            .apply(
                    "ReadFromKafka",
                    KafkaIO.<String, String>read()
                            .withTopic(options.getKafkaReadTopics())
                            .withBootstrapServers(options.getReadBootstrapServers())
                            .withKeyDeserializer(StringDeserializer.class)
                            .withValueDeserializer(StringDeserializer.class)
                            .withTimestampPolicyFactory((tp, previousWatermark) -> new CustomFieldTimePolicy(previousWatermark))
                            .withoutMetadata())
            .apply(Values.create())
            .apply(Window.<String>into(FixedWindows.of(Duration.millis(1)))
                    .triggering(Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.millis(1))))
                    .discardingFiredPanes()
                    .withAllowedLateness(Duration.ZERO))
            .apply("print", ParDo.of(new DoFn<String, KV<String, Product>>() {
                    @ProcessElement
                    public void processElement(@Element String input, ProcessContext context) {
                        Gson gson = builder.create();
                        NegotiationProduct negotiationProducts = gson.fromJson(input, NegotiationProduct.class);
                        fileName = negotiationProducts.getTransactionId();
                        Arrays.stream(negotiationProducts.getProducts())
                                .forEach(product -> context.output(KV.of(product.getNegotiationId(), product)));
                    }
                }))
            .apply(GroupByKey.create())
            .apply(WithKeys.of(0))
            .apply(GroupIntoBatches.ofSize(8))
            .apply(ParDo.of(new DoFn<KV<Integer, Iterable<KV<String, Iterable<Product>>>>, String>() {
                @ProcessElement
                public void processElement(ProcessContext c) {
                    GsonBuilder builder = new GsonBuilder();
                    builder.setPrettyPrinting();
                    Gson gson = builder.create();

                    List<Negotiation> negotiations = new ArrayList<>();
                    Objects.requireNonNull(c.element()).getValue().forEach(stringIterableKV -> {
                        Negotiation negotiation1 = new Negotiation(stringIterableKV.getKey(), IterableUtils.toList(stringIterableKV.getValue()));
                        negotiations.add(negotiation1);
                    });

                    Payload payload = new Payload(negotiations);
                    String products = gson.toJson(payload);
                    Output output = new Output(
                            new Header(
                                    "a30631d2-56a0-4bdf-b311-355676495af3",
                                    "sinco-notificacion-aprobacion-segmentacion-mq",
                                    LocalDateTime.now().atOffset(ZoneOffset.UTC).toString(),
                                    new FlexField(fileName)),
                            new DataInfo(
                                    "sinco-notificacion-aprobacion-segmentacion-mq",
                                    "application/json",
                                    products.replaceAll("\n|\t|\s", ""),
                                    LocalDateTime.now().atOffset(ZoneOffset.UTC).toString(),
                                    "a30631d2-56a0-4bdf-b311-355676495af3",
                                    "negotiation-load-approved-2096-1664469006272-a30631d2-56a0-4bdf-b311-355676495af3.json",
                                    "UTF-8"
                            ));
                    String results = gson.toJson(output);
                    c.output(results);
                }
            }))
            .apply("write to kafka", KafkaIO.<Void, String>write()
                    .withBootstrapServers("localhost:9092")
                    .withTopic("exito")
                    .withValueSerializer(StringSerializer.class)
                    .withProducerConfigUpdates(ImmutableMap.of(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 52428800))
                    .values());

    return pipeline.run();
}

问题根源分析

窗口与触发器配置错误：使用1ms的FixedWindow搭配无限重复的处理时间触发器，会导致数据被多次触发处理；discardingFiredPanes会丢弃已触发的数据，容易造成数据丢失；DirectRunner的本地并行特性会加剧这种窗口触发的混乱。
全局变量fileName竞争：在ParDo中使用全局变量存储transactionId，多线程并行处理时会被不同数据覆盖，导致后续输出的Header中FlexField关联错误，甚至数据重复关联。
GroupIntoBatches使用不当：GroupByKey后用WithKeys.of(0)将所有分组合并为同一个key，再进行批量分组，会引发严重的数据倾斜，且批量逻辑与窗口触发冲突，导致部分数据被多次批量或遗漏。
自定义TimestampPolicy潜在问题：如果CustomFieldTimePolicy未正确推进水印，会导致窗口无法正常关闭或提前关闭，引发数据处理不完整或重复。

修复建议与修正代码

核心修复点

移除不必要的全局窗口配置，依赖GroupByKey默认的全局窗口，配合合理的触发器实现低延迟处理。
将transactionId与Product绑定传递，避免全局变量竞争。
调整批量逻辑，针对每个negotiationId的组进行批量，或根据实际业务需求调整批量策略。
优化触发器配置，使用水印触发配合早期触发，平衡延迟与数据完整性。

修正后的代码

public PipelineResult run() {
    Pipeline pipeline = Pipeline.create(options);
    CoderRegistry coderRegistry = pipeline.getCoderRegistry();
    coderRegistry.registerCoderForType(coderPipeline.getCoder().getEncodedTypeDescriptor(), coderPipeline.getCoder());
    
    options.setProject("low-code-exito");
    options.setRegion("us-central1");
    options.setStagingLocation("gs://low-code-exito/binaries/");
    options.setGcpTempLocation("gs://low-code-exito/temp/");
    options.setNetwork("default");
    options.setSubnetwork("regions/us-central1/subnetworks/default");
    options.setStreaming(true); // 修正：正确设置流模式
    options.setKafkaReadTopics("test-topic");
    options.setReadBootstrapServers("localhost:9092");

    // 定义辅助类：绑定Product与transactionId
    class ProductWithTransactionId implements Serializable {
        private final Product product;
        private final String transactionId;

        public ProductWithTransactionId(Product product, String transactionId) {
            this.product = product;
            this.transactionId = transactionId;
        }

        public Product getProduct() { return product; }
        public String getTransactionId() { return transactionId; }
    }

    pipeline
            .apply("ReadFromKafka", KafkaIO.<String, String>read()
                    .withTopic(options.getKafkaReadTopics())
                    .withBootstrapServers(options.getReadBootstrapServers())
                    .withKeyDeserializer(StringDeserializer.class)
                    .withValueDeserializer(StringDeserializer.class)
                    .withTimestampPolicyFactory((tp, previousWatermark) -> new CustomFieldTimePolicy(previousWatermark))
                    .withoutMetadata())
            .apply(Values.create())
            // 移除不合理的窗口配置，GroupByKey默认使用全局窗口
            .apply("ParseAndExtractProducts", ParDo.of(new DoFn<String, KV<String, ProductWithTransactionId>>() {
                private final Gson gson = new GsonBuilder().create(); // 提前初始化Gson，避免重复创建

                @ProcessElement
                public void processElement(@Element String input, ProcessContext context) {
                    NegotiationProduct negotiationProducts = gson.fromJson(input, NegotiationProduct.class);
                    String

使用Windows与Triggers的Apache Beam数据流管道异常排查

问题描述 #

测试JSON示例 #

原实现代码 #

问题根源分析 #

修复建议与修正代码 #

核心修复点 #

修正后的代码 #

问题描述

测试JSON示例

原实现代码

问题根源分析

修复建议与修正代码

核心修复点

修正后的代码