SO Q&A 中也讨论了以下模式，这对于您的用例来说可能是一个很好的探索模式。可能成为问题的一项是压缩的慢速移动流的大小。希望它有用。
对于这种模式，我们可以使用 GenerateSequence 源转换定期发出一个值，例如每天一次。
通过在每个元素上激活的数据驱动触发器将此值传递到全局窗口。
在 DoFn 中，使用此过程作为触发器从有界源中提取数据
创建 SideInput 以用于下游转换。
请务必注意，由于此模式使用全局窗口 SideInput 触发处理时间，因此与事件时间处理的元素的匹配将是不确定的。例如，如果我们有一个在事件时间窗口化的主管道，这些窗口将看到的 SideInput 视图的版本将取决于在处理时间而不是事件时间中触发的最新触发器。
同样重要的是要注意，通常 SideInput 应该是适合内存的东西。
Java（SDK 2.9.0）：
在下面的示例中，sideinput 以非常短的间隔更新，这样可以很容易地看到效果。期望侧输入更新缓慢，例如每隔几个小时或一天一次。
在下面的示例代码中，我们使用了在 DoFn 中创建的 Map，它成为 View.asSingleton，这是此模式的推荐方法。
下面的示例说明了该模式，请注意 View.asSingleton 会在每次计数器更新时重建。
对于您的用例，您可以将 GenerateSequence 转换替换为 PubSubIO 转换。这有意义吗？
public static void main(String[] args) {
// Create pipeline
PipelineOptions options = PipelineOptionsFactory.fromArgs(args).withValidation()
.as(PipelineOptions.class);
// Using View.asSingleton, this pipeline uses a dummy external service as illustration.
// Run in debug mode to see the output
Pipeline p = Pipeline.create(options);
// Create slowly updating sideinput
PCollectionView<Map<String, String>> map = p
.apply(GenerateSequence.from(0).withRate(1, Duration.standardSeconds(5L)))
.apply(Window.<Long>into(new GlobalWindows())
.triggering(Repeatedly.forever(AfterProcessingTime.pastFirstElementInPane()))
.discardingFiredPanes())
.apply(ParDo.of(new DoFn<Long, Map<String, String>>() {
@ProcessElement public void process(@Element Long input,
OutputReceiver<Map<String, String>> o) {
// Do any external reads needed here...
// We will make use of our dummy external service.
// Every time this triggers, the complete map will be replaced with that read from
// the service.
o.output(DummyExternalService.readDummyData());
}
})).apply(View.asSingleton());
// ---- Consume slowly updating sideinput
// GenerateSequence is only used here to generate dummy data for this illustration.
// You would use your real source for example PubSubIO, KafkaIO etc...
p.apply(GenerateSequence.from(0).withRate(1, Duration.standardSeconds(1L)))
.apply(Window.into(FixedWindows.of(Duration.standardSeconds(1))))
.apply(Sum.longsGlobally().withoutDefaults())
.apply(ParDo.of(new DoFn<Long, KV<Long, Long>>() {
@ProcessElement public void process(ProcessContext c) {
Map<String, String> keyMap = c.sideInput(map);
c.outputWithTimestamp(KV.of(1L, c.element()), Instant.now());
LOG.debug("Value is {} key A is {} and key B is {}"
, c.element(), keyMap.get("Key_A"),keyMap.get("Key_B"));
}
}).withSideInputs(map));
p.run();
}
public static class DummyExternalService {
public static Map<String, String> readDummyData() {
Map<String, String> map = new HashMap<>();
Instant now = Instant.now();
DateTimeFormatter dtf = DateTimeFormat.forPattern("HH:MM:SS");
map.put("Key_A", now.minus(Duration.standardSeconds(30)).toString(dtf));
map.put("Key_B", now.minus(Duration.standardSeconds(30)).toString());
return map;
}
}