Apache Beam 中的窗口数据每小时（顺时针）基础

Posted 2023-03-31

【中文标题】Apache Beam 中的窗口数据每小时（顺时针）基础

【英文标题】：window Data hourly(clockwise) basis in Apache beam

【发布时间】：2018-11-15 19:22:30

【问题描述】：

我正在尝试在 DataFlow/Apache Beam 作业中汇总每小时（如 12:00 到 12:59 和 01:00 到 01:59）的流数据。

以下是我的用例

数据从 pubsub 流式传输，它有一个时间戳（订单日期）。我想计算每小时收到的订单数量，我还想允许延迟 5 小时。以下是我正在使用的示例代码

    LOG.info("Start Running Pipeline");
    DataflowPipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation().as(DataflowPipelineOptions.class);

    Pipeline pipeline = Pipeline.create(options);
    PCollection<String>  directShipmentFeedData = pipeline.apply("Get Direct Shipment Feed Data", PubsubIO.readStrings().fromSubscription(directShipmentFeedSubscription));
    PCollection<String>  tibcoRetailOrderConfirmationFeedData = pipeline.apply("Get Tibco Retail Order Confirmation Feed Data", PubsubIO.readStrings().fromSubscription(tibcoRetailOrderConfirmationFeedSubscription));

    PCollection<String> flattenData = PCollectionList.of(directShipmentFeedData).and(tibcoRetailOrderConfirmationFeedData)
            .apply("Flatten Data from PubSub", Flatten.<String>pCollections());

    flattenData
        .apply(ParDo.of(new DataParse())).setCoder(SerializableCoder.of(SalesAndUnits.class))

        // Adding Window

        .apply(
                Window.<SalesAndUnits>into(
                            SlidingWindows.of(Duration.standardMinutes(15))
                            .every(Duration.standardMinutes(1)))
                            )

        // Data Enrich with Dimensions
        .apply(ParDo.of(new DataEnrichWithDimentions()))

        // Group And Hourly Sum
        .apply(new GroupAndSumSales())

        .apply(ParDo.of(new SQLWrite())).setCoder(SerializableCoder.of(SalesAndUnits.class));
    pipeline.run();
    LOG.info("Finish Running Pipeline");

【参考方案1】：

我会使用符合您要求的窗口。类似于

Window.into(
  FixedWindows.of(Duration.standardHours(1))
).withAllowedLateness(Duration.standardHours(5)))

可能后跟count，因为我理解你需要这个。

希望对你有帮助