小编Man*_*rez的帖子

TL; DR:

如何使用CalendarWindows设置相同的窗口策略CoGroupByKey一组PCollections？

长版

我正在编写一个从两个不同的pub/subs读取的数据流管道,其中一个PCollections被拆分为PCollectionTuple,最后我尝试将它们连接到CoGroupByKey,然后将其保存在BigQuery中.

在测试管道期间,我的PCollections的窗口策略是:

private static PCollection<KV<String, Long>> applyWindowsAndCount(final PCollection<KV<String, Long>> summary, final String OperationName){
    return summary
            .apply("Apply Windows " + OperationName, Window
                    .<KV<String, Long>>into(FixedWindows.of(Duration.standardMinutes(1))) 
                    .discardingFiredPanes()
                    .withAllowedLateness(Duration.standardDays(1)))
            .apply("Count " + OperationName, Count.perKey());
}

我用1分钟长度的FixedWindow设置它们以便快速获得结果.

我的分组如下:

private static PCollection<KV<String, CoGbkResult>> MergeSummary(PCollection<KV<String, Long>> Avail, PCollection<KV<String, Long>> ValuationOK, PCollection<KV<String, Long>> ValuationKO){
    return KeyedPCollectionTuple.of(Util.AVAIL, Avail)
                                .and(Util.VALUATION_OK, ValuationOK)
                                .and(Util.VALUATION_KO, ValuationKO)
                                .apply("Merge Summary", CoGroupByKey.create());
}

当我在本地和云上进行测试时,它运行顺畅,但是,当我使用实际生产值设置窗口时,CalendarWindows的长度为1天,如下所示:

private static PCollection<KV<String, Long>> applyWindowsAndCount(final PCollection<KV<String, Long>> summary, …