(Forgot to add, this was with the transform coder wrapped with
NullableCoder so it didn’t fail outright.)
- Claire
On Tue, Mar 22, 2022 at 11:26 PM Claire McGinty <claire.d.mcgi...@gmail.com>
wrote:
> I was wondering if there’s any difference in how Dataflow v1 vs v2 loads
> CoGbkResult iterables - in one of our internal pipielines that started
> failing, I added some logging to CoGbkResult and found that the
> TagIterable’s iterator would return hasNext() == true, but next() == null.
>
> - Claire
>
> On Tue, Mar 22, 2022 at 6:13 PM Robert Bradshaw <rober...@google.com>
> wrote:
>
>> BEAM-13541 is as far as I'm aware the only major change that has
>> happened in this code recently, but this should be totally agnostic of
>> Dataflow v1 vs. v2 vs. DirectRunner. Have there been any changes to
>> the shuffle/reiterable code?
>>
>> On Tue, Mar 22, 2022 at 2:37 PM Steve Niemitz <sniem...@apache.org>
>> wrote:
>> >
>> > oh I just realized I responded to a different thread, feel free to
>> ignore me.
>> >
>> > On Tue, Mar 22, 2022 at 5:34 PM Niel Markwick <ni...@google.com> wrote:
>> >>
>> >> yeah there does seem to be some heisenbug attributes...
>> >> It failed on 4 out of 6 runs with this reproducer, and always
>> succeeded with DirectRunner or 2.35.0...
>> >> It does seem to depend on the number of elements in the group.
>> >>
>> >>
>> >> On Tue, 22 Mar 2022 at 22:27, Steve Niemitz <sniem...@apache.org>
>> wrote:
>> >>>
>> >>> Your email was actually what made me notice this! :D
>> >>>
>> >>> I haven't been able to reproduce the NPE you found (also on 2.37) but
>> that certainly doesn't mean it's not a bug.
>> >>>
>> >>>
>> >>>
>> >>> On Tue, Mar 22, 2022 at 5:23 PM Niel Markwick <ni...@google.com>
>> wrote:
>> >>>>
>> >>>> I have also seen this with Java beam 2.36.0 and 2.37.0, again with
>> large groups...
>> >>>>
>> >>>> The behaviour is that some elements become null, which then causes
>> either coder problems for the primitive types (as seen in Claire's stack
>> trace), or downstream issues for custom types which are not expecting null
>> elements...
>> >>>>
>> >>>> I have a relatively simple Java reproducer creating 200,000 KVs and
>> doing a CoGroupByKey with a single KV. This runs fine on DirectRunner, and
>> on Dataflow with 2.35.0, but fails on Dataflow on 2.36.0 and 2.47.0 with
>> the same error: "org.apache.beam.sdk.coders.CoderException: cannot encode a
>> null String" in the CoGroupByKey step
>> >>>>
>> >>>> Pipeline p =
>> Pipeline.create(PipelineOptionsFactory.fromArgs(args).create());
>> >>>> String[] bigarray = new String[200000];
>> >>>> Arrays.fill(bigarray, "Hello World");
>> >>>> PCollection<KV<Integer, String>> right =
>> >>>> p.apply(Create.of(Arrays.asList(bigarray)))
>> >>>> .apply(
>> >>>> MapElements.into(new TypeDescriptor<KV<Integer,
>> String>>() {})
>> >>>> .via(input -> KV.of(100, input)));
>> >>>> PCollection<KV<Integer, String>> left =
>> p.apply(Create.of(KV.of(100, "goodbye world")));
>> >>>>
>> >>>> TupleTag<String> leftTag = new TupleTag<String>();
>> >>>> TupleTag<String> rightTag = new TupleTag<String>();
>> >>>>
>> >>>> PCollection<KV<Integer, CoGbkResult>> joinedResult =
>> >>>> KeyedPCollectionTuple.of(leftTag, left)
>> >>>> .and(rightTag, right)
>> >>>> .apply("Join By Key", CoGroupByKey.<Integer>create());
>> >>>>
>> >>>> joinedResult.apply(
>> >>>> "Report",
>> >>>> MapElements.into(TypeDescriptor.of(Void.class))
>> >>>> .via(
>> >>>> (KV<Integer, CoGbkResult> element) -> {
>> >>>> Iterable<String> leftValues =
>> element.getValue().getAll(leftTag);
>> >>>> Iterable<String> rightValues =
>> element.getValue().getAll(rightTag);
>> >>>>
>> >>>> System.out.println("Key = " + element.getKey());
>> >>>> System.out.println("left size= " +
>> Iterables.size(leftValues));
>> >>>> System.out.println("right size= " +
>> Iterables.size(rightValues));
>> >>>> return (Void) null;
>> >>>> }));
>> >>>>
>> >>>> p.run().waitUntilFinish();
>> >>>>
>> >>>>
>> >>>> On Fri, 18 Mar 2022, 16:01 Reuven Lax, <re...@google.com> wrote:
>> >>>>>
>> >>>>>
>> >>>>> On Fri, Mar 18, 2022, 8:28 AM Claire McGinty <
>> claire.d.mcgi...@gmail.com> wrote:
>> >>>>>>
>> >>>>>> To add more investigative context, we discovered that these jobs
>> succeed when run with Dataflow Prime; it's just Beam 2.36.0 + Dataflow V1
>> that produces this regression. Unfortunately it's not feasible to upgrade
>> our whole fleet to Dataflow Prime right now, so we've created an internal
>> Google support ticket for this, too.
>> >>>>>>
>> >>>>>> - Claire
>> >>>>>>
>> >>>>>> On Tue, Mar 15, 2022 at 12:22 PM Claire McGinty <
>> claire.d.mcgi...@gmail.com> wrote:
>> >>>>>>>
>> >>>>>>> Hi! No, there aren't null strings in the data--it would have
>> thrown a NPE much earlier if that were the case, and likely failed when run
>> with Beam 2.35.0 also, unless null-handling changed too?
>> >>>>>>>
>> >>>>>>> - Claire
>> >>>>>>>
>> >>>>>>> On Tue, Mar 15, 2022 at 12:19 PM Reuven Lax <re...@google.com>
>> wrote:
>> >>>>>>>>
>> >>>>>>>> Is it expected to have null strings in your data?
>> >>>>>>>>
>> >>>>>>>> On Tue, Mar 15, 2022 at 9:06 AM Claire McGinty <
>> claire.d.mcgi...@gmail.com> wrote:
>> >>>>>>>>>
>> >>>>>>>>> Hi Beam devs!
>> >>>>>>>>>
>> >>>>>>>>> I wanted to surface a possible regression with CoGroupByKeys in
>> this transform. Our organization (which primarily uses Scio) has had
>> several pipelines start failing after upgrading from Beam 2.35.0 to Beam
>> 2.36.0, specifically during cogroup operations with large (>10,000) key
>> groups.
>> >>>>>>>>>
>> >>>>>>>>> We initially saw this problem using Scio join operations, but I
>> was able to reproduce using public datasets & Beam PTransforms directly
>> (job code link; the repo README includes instructions to run it if anyone
>> is interested). This job throws the following error when run in Dataflow
>> with Beam 2.36.0:
>> >>>>>>>>>
>> >>>>>>>>> Caused by: java.lang.IllegalArgumentException: Unable to encode
>> element
>> '[org.apache.beam.sdk.transforms.join.CoGbkResult$TagIterable@33dd2cbb,
>> org.apache.beam.sdk.transforms.join.CoGbkResult$TagIterable@26edfb82]'
>> with coder
>> 'org.apache.beam.sdk.transforms.join.CoGbkResult$CoGbkResultCoder@346927'.
>> org.apache.beam.sdk.coders.Coder.getEncodedElementByteSize(Coder.java:300)
>> >>>>>>>>>
>> org.apache.beam.sdk.coders.Coder.registerByteSizeObserver(Coder.java:291)
>> >>>>>>>>>
>> org.apache.beam.sdk.coders.KvCoder.registerByteSizeObserver(KvCoder.java:130)
>> >>>>>>>>>
>> org.apache.beam.sdk.coders.KvCoder.registerByteSizeObserver(KvCoder.java:37)
>> >>>>>>>>>
>> org.apache.beam.sdk.util.WindowedValue$FullWindowedValueCoder.registerByteSizeObserver(WindowedValue.java:642)
>> >>>>>>>>>
>> org.apache.beam.sdk.util.WindowedValue$FullWindowedValueCoder.registerByteSizeObserver(WindowedValue.java:558)
>> >>>>>>>>>
>> org.apache.beam.runners.dataflow.worker.IntrinsicMapTaskExecutorFactory$ElementByteSizeObservableCoder.registerByteSizeObserver(IntrinsicMapTaskExecutorFactory.java:403)
>> >>>>>>>>>
>> org.apache.beam.runners.dataflow.worker.util.common.worker.OutputObjectAndByteCounter.update(OutputObjectAndByteCounter.java:128)
>> >>>>>>>>>
>> org.apache.beam.runners.dataflow.worker.DataflowOutputCounter.update(DataflowOutputCounter.java:67)
>> >>>>>>>>>
>> org.apache.beam.runners.dataflow.worker.util.common.worker.OutputReceiver.process(OutputReceiver.java:43)
>> >>>>>>>>>
>> org.apache.beam.runners.dataflow.worker.SimpleParDoFn$1.output(SimpleParDoFn.java:285)
>> >>>>>>>>>
>> org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.outputWindowedValue(SimpleDoFnRunner.java:268)
>> >>>>>>>>>
>> org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner.access$900(SimpleDoFnRunner.java:84)
>> >>>>>>>>>
>> org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output(SimpleDoFnRunner.java:416)
>> >>>>>>>>>
>> org.apache.beam.runners.dataflow.worker.repackaged.org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.output(SimpleDoFnRunner.java:404)
>> >>>>>>>>>
>> org.apache.beam.sdk.transforms.join.CoGroupByKey$ConstructCoGbkResultFn.processElement(CoGroupByKey.java:192)
>> >>>>>>>>> Caused by: org.apache.beam.sdk.coders.CoderException: cannot
>> encode a null String
>> org.apache.beam.sdk.coders.StringUtf8Coder.encode(StringUtf8Coder.java:74)
>> >>>>>>>>>
>> >>>>>>>>> To troubleshoot, I first tried downgrading to Beam 2.35.0; the
>> job then ran successfully. Next, I tried running the job using Beam 2.36.0,
>> but with 2.35.0's version of CoGbkResult on the classpath; it ran
>> successfully then, too. I'm wondering if it's related to the lazy-iterator
>> improvements introduced in BEAM-13541? I wasn't able to reproduce it with a
>> unit test yet, unfortunately; maybe it's unique to the way Dataflow
>> lazily-loads large CoGbkResults.
>> >>>>>>>>>
>> >>>>>>>>> I'm continuing to try to reproduce this on a smaller scale, but
>> wanted to flag it here for now; it's preventing our pipeline fleet from
>> upgrading to Beam 2.36.0. Any insight would be appreciated!
>> >>>>>>>>>
>> >>>>>>>>> Thanks,
>> >>>>>>>>> Claire
>> >>>>>>>>>
>> >>>>>>>>>
>> >>>>>>>>>
>> >>>>>>>>>
>>
>
