创建 PCollectionView<Map<String,String>> 时如何解决重复值异常
How to solve Duplicate values exception when I create PCollectionView<Map<String,String>>
我正在我的 Apache-Beam 管道中设置一个缓慢变化的查找映射。它不断更新查找映射。对于查找映射中的每个键，我使用累积模式检索全局 window 中的最新值。
但它总是遇到异常：
org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.IllegalArgumentException: Duplicate values for mykey
这段代码有什么问题吗？
如果我改用.discardingFiredPanes()，我会在最后一次发出时丢失信息。
pipeline
.apply(GenerateSequence.from(0).withRate(1, Duration.standardMinutes(1L)))
.apply(
Window.<Long>into(new GlobalWindows())
.triggering(Repeatedly.forever(
AfterProcessingTime.pastFirstElementInPane()))
.accumulatingFiredPanes())
.apply(new ReadSlowChangingTable())
.apply(Latest.perKey())
.apply(View.asMap());
示例输入触发器：
t1 : KV<k1,v1> KV< k2,v2>
t2 : KV<k1,v1>
accumulatingFiredPanes => t2 的预期结果 => KV(k1,v1), KV(k2,v2) 但由于重复异常而失败
discardingFiredPanes => t2 的预期结果 => KV(k1,v1) 成功
特别是关于 view.asMap 和评论中的累积窗格讨论：
如果您想使用 View.asMap 辅助输入（例如，当地图元素的源本身是分布式的时——通常是因为您正在从先前的输出创建辅助输入transform），还有一些其他因素需要考虑：View.asMap本身就是一个聚合，它会继承触发并累加它的输入。在此特定模式中，在此转换之前将管道设置为 accumulatingPanes 模式将导致重复键错误，即使在 View.asMap 转换之前使用 Latest.perKey 等转换也是如此。
鉴于读取会更新整个地图，那么我认为使用 View.asSingleton 是此用例的更好方法。
关于此模式的一些一般说明，希望对其他人也有用：
对于此模式，我们可以使用 GenerateSequence 源转换定期发出一个值，例如每天一次。通过在每个元素上激活的数据驱动触发器将此值传递到全局 window。在 DoFn 中，使用此过程作为触发器从您的有界源 Create 您的 SideInput 中提取数据以用于下游转换。
请务必注意，由于此模式使用全局-window 边输入触发处理时间，因此与事件时间中正在处理的元素的匹配将是不确定的。例如，如果我们有一个在事件时间 windowed 的主管道，那些 windows 将看到的 SideInput View 的版本将取决于在处理时间触发的最新触发器，而不是活动时间。
同样重要的是要注意，一般来说，侧输入应该是适合记忆的东西。
Java（SDK 2.9.0）：
在下面的示例中，侧输入以非常短的间隔更新，这样可以很容易地看到效果。期望侧输入更新缓慢，例如每隔几个小时或每天一次。
在下面的示例代码中，我们使用了在 DoFn 中创建的 Map，它变成了 View.asSingleton，这是此模式的推荐方法。
下面的示例说明了该模式，请注意 View.asSingleton 会在每次计数器更新时重建。
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;
}
}
我正在我的 Apache-Beam 管道中设置一个缓慢变化的查找映射。它不断更新查找映射。对于查找映射中的每个键，我使用累积模式检索全局 window 中的最新值。 但它总是遇到异常：
org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.IllegalArgumentException: Duplicate values for mykey
这段代码有什么问题吗？
如果我改用.discardingFiredPanes()，我会在最后一次发出时丢失信息。
pipeline
.apply(GenerateSequence.from(0).withRate(1, Duration.standardMinutes(1L)))
.apply(
Window.<Long>into(new GlobalWindows())
.triggering(Repeatedly.forever(
AfterProcessingTime.pastFirstElementInPane()))
.accumulatingFiredPanes())
.apply(new ReadSlowChangingTable())
.apply(Latest.perKey())
.apply(View.asMap());
示例输入触发器：
t1 : KV<k1,v1> KV< k2,v2>
t2 : KV<k1,v1>
accumulatingFiredPanes => t2 的预期结果 => KV(k1,v1), KV(k2,v2) 但由于重复异常而失败
discardingFiredPanes => t2 的预期结果 => KV(k1,v1) 成功
特别是关于 view.asMap 和评论中的累积窗格讨论：
如果您想使用 View.asMap 辅助输入（例如，当地图元素的源本身是分布式的时——通常是因为您正在从先前的输出创建辅助输入transform），还有一些其他因素需要考虑：View.asMap本身就是一个聚合，它会继承触发并累加它的输入。在此特定模式中，在此转换之前将管道设置为 accumulatingPanes 模式将导致重复键错误，即使在 View.asMap 转换之前使用 Latest.perKey 等转换也是如此。
鉴于读取会更新整个地图，那么我认为使用 View.asSingleton 是此用例的更好方法。
关于此模式的一些一般说明，希望对其他人也有用：
对于此模式，我们可以使用 GenerateSequence 源转换定期发出一个值，例如每天一次。通过在每个元素上激活的数据驱动触发器将此值传递到全局 window。在 DoFn 中，使用此过程作为触发器从您的有界源 Create 您的 SideInput 中提取数据以用于下游转换。
请务必注意，由于此模式使用全局-window 边输入触发处理时间，因此与事件时间中正在处理的元素的匹配将是不确定的。例如，如果我们有一个在事件时间 windowed 的主管道，那些 windows 将看到的 SideInput View 的版本将取决于在处理时间触发的最新触发器，而不是活动时间。
同样重要的是要注意，一般来说，侧输入应该是适合记忆的东西。
Java（SDK 2.9.0）：
在下面的示例中，侧输入以非常短的间隔更新，这样可以很容易地看到效果。期望侧输入更新缓慢，例如每隔几个小时或每天一次。
在下面的示例代码中，我们使用了在 DoFn 中创建的 Map，它变成了 View.asSingleton，这是此模式的推荐方法。
下面的示例说明了该模式，请注意 View.asSingleton 会在每次计数器更新时重建。
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;
}
}