当使用CalendarWindows的CoGroupByKey时,Flatten的输入具有不兼容的窗口窗口

Question

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, final String OperationName){
        return summary
                .apply("Apply Windows " + OperationName, Window
                                .<KV<String, Long>>into(CalendarWindows.days(1).withTimeZone(DateTimeZone.UTC).withStartingDay(2016,9,20)) //Per day windowing.                                    
                                .discardingFiredPanes()
                                .withAllowedLateness(Duration.standardDays(1))) //Accepts X days late data.
                .apply("Count " + OperationName, Count.perKey());
    }

然后我甚至无法编译代码,因为我收到如下消息:

Exception in thread "main" java.lang.IllegalStateException: Inputs to Flatten had incompatible window windowFns: com.google.cloud.dataflow.sdk.transforms.windowing.CalendarWindows$DaysWindows@6af9fcb2, com.google.cloud.dataflow.sdk.transforms.windowing.CalendarWindows$DaysWindows@6cce16f4
at com.google.cloud.dataflow.sdk.transforms.Flatten$FlattenPCollectionList.apply(Flatten.java:121)
at com.google.cloud.dataflow.sdk.transforms.Flatten$FlattenPCollectionList.apply(Flatten.java:105)
at com.google.cloud.dataflow.sdk.runners.PipelineRunner.apply(PipelineRunner.java:74)
at com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner.apply(DataflowPipelineRunner.java:413)
at com.google.cloud.dataflow.sdk.Pipeline.applyInternal(Pipeline.java:367)
at com.google.cloud.dataflow.sdk.Pipeline.applyTransform(Pipeline.java:274)
at com.google.cloud.dataflow.sdk.values.PCollectionList.apply(PCollectionList.java:175)
at com.google.cloud.dataflow.sdk.transforms.join.CoGroupByKey.apply(CoGroupByKey.java:124)
at com.google.cloud.dataflow.sdk.transforms.join.CoGroupByKey.apply(CoGroupByKey.java:74)
at com.google.cloud.dataflow.sdk.runners.PipelineRunner.apply(PipelineRunner.java:74)
at com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner.apply(DataflowPipelineRunner.java:413)
at com.google.cloud.dataflow.sdk.Pipeline.applyInternal(Pipeline.java:367)
at com.google.cloud.dataflow.sdk.Pipeline.applyTransform(Pipeline.java:290)
at com.google.cloud.dataflow.sdk.transforms.join.KeyedPCollectionTuple.apply(KeyedPCollectionTuple.java:116)

阅读文档后我发现了这个:

使用CoGroupByKey对应用了窗口策略的PCollection进行分组时,要分组的所有PCollections必须使用相同的窗口策略和窗口大小调整.例如,您正在合并的所有集合必须使用(假设)相同的5分钟固定窗口或每30秒开始的4分钟滑动窗口.

如果您的管道尝试使用CoGroupByKey将PCollections与不兼容的窗口合并,则在构建管道时,Dataflow将生成IllegalStateException错误.

很明显,数据流认为我的PCollections具有不兼容的窗口,但是,所有这些都是使用我之前复制的函数应用的.所以,我怎么能CoGroupByKey一组PCollections的与CalendarWindows设置相同的窗口战略是什么？

Answer 1

看起来这是 CalendarWindows 中的一个错误；要解决此问题，您可以创建一个 CalendarWindows 对象，并将其用作每个 PCollection 的 WindowFn，而不是为每个 PCollection 创建单独的 CalendarWindows 对象。