Hi Rajath,
I think you need to guarantee that the PCollection will have a single
element, so you can use .asSingleton().
A possible approach is getting the latest for the window prior to
.asSingleton():
PCollectionView<Map<String, String>> targetCountryByJourneyId =
p.apply("Generate a sequence for extracting resource_id->targetCountry
map", GenerateSequence.from(0).withRate(1,
Duration.standardSeconds(Long.parseLong(options.getRulesRefreshDuration().get()))))
.apply("Get all rules from Redis for resourceId->targetCountry
map", ParDo.of(new
GetAllRulesFromRedis(Integer.parseInt(options.getRedisPort().get()),
options.getRedisHost().get(), statsdHost, statsdPort)))
.apply("Transform all Rules into journeyId->targetCountry map",
ParDo.of(new TransformRulesAsJourneyIdToMetadataMap(statsdHost, statsdPort,
Rule.TARGET_COUNTRY)))
.apply("Window journeyId->targetCountry map", Window.<Map<String,
String>>into(new GlobalWindows())
.triggering(Repeatedly.forever(AfterPane.elementCountAtLeast(1)))
.discardingFiredPanes())
* .apply("Combine to Latest", Latest.globally())*
.apply("singleton view of journeyId -> targetCountry map",
View.asSingleton());
Best,
Bruno
On Fri, Oct 20, 2023 at 9:05 AM Rajath BK <rajath.u...@gmail.com> wrote:
>
> Hi ,
> We are using a few side Inputs access data that would be used in
> filtering out elements in the pipelines. The source of the side input data
> is redis instance in Google cloud memory store.
> The jobs are deployed using the generated pipeline templates via the
> Dataflow service APIs.
>
> The side inputs are generated as below
> PCollectionView<Map<String, String>> targetCountryByJourneyId =
> p.apply("Generate a sequence for extracting resource_id->targetCountry
> map", GenerateSequence.from(0).withRate(1,
>
> Duration.standardSeconds(Long.parseLong(options.getRulesRefreshDuration().get()))))
> .apply("Get all rules from Redis for resourceId->targetCountry
> map", ParDo.of(new
> GetAllRulesFromRedis(Integer.parseInt(options.getRedisPort().get()),
> options.getRedisHost().get(), statsdHost, statsdPort)))
> .apply("Transform all Rules into journeyId->targetCountry map",
> ParDo.of(new TransformRulesAsJourneyIdToMetadataMap(statsdHost, statsdPort,
> Rule.TARGET_COUNTRY)))
> .apply("Window journeyId->targetCountry map", Window.<Map<String,
> String>>into(new GlobalWindows())
>
> .triggering(Repeatedly.forever(AfterPane.elementCountAtLeast(1)))
> .discardingFiredPanes())
> .apply("singleton view of journeyId -> targetCountry map",
> View.asSingleton());
>
> and accessed as below
>
> PCollection<UserData> qualifiedUser = userValidatedEntry.
> apply("Eliminate user by target country",
> ParDo.of(new EliminateUserByTargetCountry(
> options.getProjectNameForBT().get(),
> options.getInstanceNameForBT().get(),
> bigTableAppProfileId,
> targetCountryByJourneyId,
> statsdHost,
> statsdPort)
> )
> .withSideInputs(targetCountryByJourneyId));
>
> The pipelines are being run on Google cloud Dataflow and
> We have been noticing 2 kinds of issues
> 1. Despite closing the window on the side input collection when the first
> element occurs, we are seeing this error. Strange thing is we don't see
> these errors occurring across all the workers in the job and we have also
> noticed this issue to get resolved during new deployments and only to
> resurface later.
>
> java.lang.IllegalArgumentException: PCollection with more than one element
> accessed as a singleton view. Consider using
> Combine.globally().asSingleton() to combine the PCollection into a single
> value at
> org.apache.beam.sdk.transforms.View$SingletonCombineFn.apply(View.java:434)
> at
> org.apache.beam.sdk.transforms.Combine$BinaryCombineFn.addInput(Combine.java:524)
> at
> org.apache.beam.sdk.transforms.Combine$BinaryCombineFn.addInput(Combine.java:493)
> at
> org.apache.beam.runners.dataflow.worker.WindmillStateInternals$WindmillCombiningState.add(WindmillStateInternals.java:2051)
> at
> org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SystemReduceFn.processValue(SystemReduceFn.java:119)
> at
> org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.ReduceFnRunner.processElement(ReduceFnRunner.java:613)
> at
> org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.ReduceFnRunner.processElements(ReduceFnRunner.java:360)
> at
> org.apache.beam.runners.dataflow.worker.StreamingGroupAlsoByWindowViaWindowSetFn.processElement(StreamingGroupAlsoByWindowViaWindowSetFn.java:96)
> at
> org.apache.beam.runners.dataflow.worker.StreamingGroupAlsoByWindowViaWindowSetFn.processElement(StreamingGroupAlsoByWindowViaWindowSetFn.java:43)
> at
> org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner.invokeProcessElement(GroupAlsoByWindowFnRunner.java:121)
> at
> org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowFnRunner.processElement(GroupAlsoByWindowFnRunner.java:73)
> at
> org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.LateDataDroppingDoFnRunner.processElement(LateDataDroppingDoFnRunner.java:80)
> at
> org.apache.beam.runners.dataflow.worker.GroupAlsoByWindowsParDoFn.processElement(GroupAlsoByWindowsParDoFn.java:137)
> at
> org.apache.beam.runners.dataflow.worker.util.common.worker.ParDoOperation.process(ParDoOperation.java:44)
> at
> org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:49)
> at
> org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.runReadLoop(ReadOperation.java:218)
> at
> org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.start(ReadOperation.java:169)
> at
> org.apache.beam.runners.dataflow.worker.util.common.worker.MapTaskExecutor.execute(MapTaskExecutor.java:83)
> at
> org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.process(StreamingDataflowWorker.java:1450)
> at
> org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.access$1100(StreamingDataflowWorker.java:165)
> at
> org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker$7.run(StreamingDataflowWorker.java:1125)
> at
> org.apache.beam.runners.dataflow.worker.util.BoundedQueueExecutor.lambda$executeLockHeld$0(BoundedQueueExecutor.java:133)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> at java.lang.Thread.run(Thread.java:748)
>
> 2. Secondly, we have noticed the side input to not contain data even when
> the underlying data exists. This side input view being a map, when this
> occurs, we see empty keys and breaks our code while accessing it.
>
> Has anyone faced similar issues ? I would greatly help if someone can shed
> some light on this seemingly non-deterministic behavior of the side inputs
> when created as Singletons.
>
> -Thanks in advance
> Rajath
>
