You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@beam.apache.org by Teodor Spæren <te...@riseup.net> on 2021/05/06 07:28:59 UTC
Getting null pointer exception in a basic setup, don't know why
Hey!
I'm having problems with a program that I used to be able to run just
fine with flink, but now I'm getting a null pointer exception.
The beam program in question looks like this:
> package no.spaeren.thesis.benchmarks.beam;
>
> import no.spaeren.thesis.benchmarks.beam.helpers.CountSource;
> import no.spaeren.thesis.benchmarks.beam.helpers.Printer;
> import org.apache.beam.runners.flink.FlinkPipelineOptions;
> import org.apache.beam.runners.flink.FlinkRunner;
> import org.apache.beam.sdk.Pipeline;
> import org.apache.beam.sdk.io.Read;
> import org.apache.beam.sdk.options.PipelineOptionsFactory;
> import org.apache.beam.sdk.transforms.MapElements;
> import org.apache.beam.sdk.transforms.ParDo;
> import org.apache.beam.sdk.values.PCollection;
> import org.apache.beam.sdk.values.TypeDescriptors;
> import picocli.CommandLine;
>
> import java.util.concurrent.Callable;
>
> @CommandLine.Command(name = "BeamSimple", mixinStandardHelpOptions = true, description = "A simple beam job")
> public class BeamSimple implements Callable<Void> {
>
> @CommandLine.Option(names = {"--from"}, defaultValue = "0")
> final Long from = 0L;
>
> @CommandLine.Option(names = {"--to"}, defaultValue = "1000")
> final Long to = 1000L;
>
>
> @Override
> public Void call() {
> FlinkPipelineOptions options = PipelineOptionsFactory.create().as(FlinkPipelineOptions.class);
> options.setDisableMetrics(true);
> options.setRunner(FlinkRunner.class);
> // options.setShutdownSourcesAfterIdleMs(100L);
> // options.setParallelism(2);
> Pipeline p = Pipeline.create(options);
>
>
> //final PCollection<Long> ds = p.apply(GenerateSequence.from(this.from).to(this.to));
> final PCollection<Long> ds = p.apply(Read.from(new CountSource(this.from, this.to)));
> final PCollection<Long> db = ds.apply(MapElements.into(TypeDescriptors.longs()).via((Long x) -> x * 2));
>
> db.apply(ParDo.of(new Printer<>("BeamSimple: %d\n")));
>
>
> p.run().waitUntilFinish();
>
> return null;
> }
> }
I'm using Beam version 2.27.0, but it also happens on 2.29.0. The Flink
version is 1.12.1.
And the error message I'm getting is:
> dragon % JAVA_HOME="/usr/lib/jvm/java-8-openjdk" ~/madsci/thesis/world/tools/flink/bin/flink info -p 1 ~/madsci/thesis/codes/mezurilo/target/mezurilo-bundled-1.0-SNAPSHOT.jar BeamSimple
> Picked up _JAVA_OPTIONS: -Dawt.useSystemAAFontSettings=on -Dswing.aatext=true
>
> ------------------------------------------------------------
> The program finished with the following exception:
>
> org.apache.flink.client.program.ProgramInvocationException: The program plan could not be fetched - the program aborted pre-maturely.
>
> Classpath: [file:/home/rhermes/madsci/thesis/codes/mezurilo/target/mezurilo-bundled-1.0-SNAPSHOT.jar]
>
> System.out: (none)
>
> System.err: java.lang.NullPointerException
> at org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull(Preconditions.java:877)
> at org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.SingletonImmutableList.<init>(SingletonImmutableList.java:38)
> at org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList.of(ImmutableList.java:94)
> at org.apache.beam.sdk.coders.NullableCoder.getCoderArguments(NullableCoder.java:102)
> at org.apache.beam.sdk.coders.StructuredCoder.getComponents(StructuredCoder.java:49)
> at org.apache.beam.sdk.coders.StructuredCoder.hashCode(StructuredCoder.java:69)
> at java.util.AbstractList.hashCode(AbstractList.java:541)
> at org.apache.beam.sdk.coders.StructuredCoder.hashCode(StructuredCoder.java:69)
> at java.util.AbstractList.hashCode(AbstractList.java:541)
> at org.apache.beam.sdk.coders.StructuredCoder.hashCode(StructuredCoder.java:69)
> at org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Hashing.smearedHash(Hashing.java:54)
> at org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.HashBiMap.get(HashBiMap.java:254)
> at org.apache.beam.runners.core.construction.SdkComponents.registerCoder(SdkComponents.java:265)
> at org.apache.beam.runners.core.construction.ParDoTranslation.translateParDo(ParDoTranslation.java:231)
> at org.apache.beam.runners.core.construction.ParDoTranslation.translateParDo(ParDoTranslation.java:210)
> at org.apache.beam.runners.core.construction.ParDoTranslation$ParDoTranslator.translate(ParDoTranslation.java:176)
> at org.apache.beam.runners.core.construction.PTransformTranslation.toProto(PTransformTranslation.java:239)
> at org.apache.beam.runners.core.construction.ParDoTranslation.getParDoPayload(ParDoTranslation.java:743)
> at org.apache.beam.runners.core.construction.ParDoTranslation.isSplittable(ParDoTranslation.java:758)
> at org.apache.beam.runners.core.construction.PTransformMatchers$6.matches(PTransformMatchers.java:274)
> at org.apache.beam.sdk.Pipeline$2.visitPrimitiveTransform(Pipeline.java:289)
> at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:587)
> at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:579)
> at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:579)
> at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:579)
> at org.apache.beam.sdk.runners.TransformHierarchy$Node.access$500(TransformHierarchy.java:239)
> at org.apache.beam.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:213)
> at org.apache.beam.sdk.Pipeline.traverseTopologically(Pipeline.java:468)
> at org.apache.beam.sdk.Pipeline.replace(Pipeline.java:267)
> at org.apache.beam.sdk.Pipeline.replaceAll(Pipeline.java:217)
> at org.apache.beam.runners.flink.FlinkPipelineExecutionEnvironment.translate(FlinkPipelineExecutionEnvironment.java:118)
> at org.apache.beam.runners.flink.FlinkRunner.run(FlinkRunner.java:92)
> at org.apache.beam.sdk.Pipeline.run(Pipeline.java:322)
> at org.apache.beam.sdk.Pipeline.run(Pipeline.java:308)
> at no.spaeren.thesis.benchmarks.beam.BeamSimple.call(BeamSimple.java:45)
> at no.spaeren.thesis.benchmarks.beam.BeamSimple.call(BeamSimple.java:18)
> at picocli.CommandLine.executeUserObject(CommandLine.java:1853)
> at picocli.CommandLine.access$1100(CommandLine.java:145)
> at picocli.CommandLine$RunLast.executeUserObjectOfLastSubcommandWithSameParent(CommandLine.java:2243)
> at picocli.CommandLine$RunLast.handle(CommandLine.java:2237)
> at picocli.CommandLine$RunLast.handle(CommandLine.java:2201)
> at picocli.CommandLine$AbstractParseResultHandler.execute(CommandLine.java:2068)
> at picocli.CommandLine.execute(CommandLine.java:1978)
> at no.spaeren.thesis.App.main(App.java:24)
> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:498)
> at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:343)
> at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:213)
> at org.apache.flink.client.program.PackagedProgramUtils.getPipelineFromProgram(PackagedProgramUtils.java:158)
> at org.apache.flink.client.cli.CliFrontend.info(CliFrontend.java:366)
> at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1067)
> at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1136)
> at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
> at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1136)
>
> at org.apache.flink.client.program.PackagedProgramUtils.generateException(PackagedProgramUtils.java:264)
> at org.apache.flink.client.program.PackagedProgramUtils.getPipelineFromProgram(PackagedProgramUtils.java:184)
> at org.apache.flink.client.cli.CliFrontend.info(CliFrontend.java:366)
> at org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1067)
> at org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1136)
> at org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
> at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1136)
Any help with this would be greatly appriciated.
Best regards,
Teodor Spæren
Re: Getting null pointer exception in a basic setup, don't know why
Posted by Jan Lukavský <je...@seznam.cz>.
Hi Teodor,
can you share (maybe github link, if you have it in public repo) the
implementation of CountSource and Printer? What changed in Beam 2.25.0
(if I recall correctly) is how Read transform is translated. It uses SDF
now, so there might be something that was broken before, but the change
of the translation revealed the problem. You can check if your problem
disappears if you add use_deprecated_read into experiments of the
PipelineOptions. See Highlights in [1].
Jan
[1] https://beam.apache.org/blog/beam-2.25.0/
On 5/6/21 9:28 AM, Teodor Spæren wrote:
> Hey!
>
> I'm having problems with a program that I used to be able to run just
> fine with flink, but now I'm getting a null pointer exception.
>
> The beam program in question looks like this:
>
>> package no.spaeren.thesis.benchmarks.beam;
>>
>> import no.spaeren.thesis.benchmarks.beam.helpers.CountSource;
>> import no.spaeren.thesis.benchmarks.beam.helpers.Printer;
>> import org.apache.beam.runners.flink.FlinkPipelineOptions;
>> import org.apache.beam.runners.flink.FlinkRunner;
>> import org.apache.beam.sdk.Pipeline;
>> import org.apache.beam.sdk.io.Read;
>> import org.apache.beam.sdk.options.PipelineOptionsFactory;
>> import org.apache.beam.sdk.transforms.MapElements;
>> import org.apache.beam.sdk.transforms.ParDo;
>> import org.apache.beam.sdk.values.PCollection;
>> import org.apache.beam.sdk.values.TypeDescriptors;
>> import picocli.CommandLine;
>>
>> import java.util.concurrent.Callable;
>>
>> @CommandLine.Command(name = "BeamSimple", mixinStandardHelpOptions =
>> true, description = "A simple beam job")
>> public class BeamSimple implements Callable<Void> {
>>
>> @CommandLine.Option(names = {"--from"}, defaultValue = "0")
>> final Long from = 0L;
>>
>> @CommandLine.Option(names = {"--to"}, defaultValue = "1000")
>> final Long to = 1000L;
>>
>>
>> @Override
>> public Void call() {
>> FlinkPipelineOptions options =
>> PipelineOptionsFactory.create().as(FlinkPipelineOptions.class);
>> options.setDisableMetrics(true);
>> options.setRunner(FlinkRunner.class);
>> // options.setShutdownSourcesAfterIdleMs(100L);
>> // options.setParallelism(2);
>> Pipeline p = Pipeline.create(options);
>>
>>
>> //final PCollection<Long> ds =
>> p.apply(GenerateSequence.from(this.from).to(this.to));
>> final PCollection<Long> ds = p.apply(Read.from(new
>> CountSource(this.from, this.to)));
>> final PCollection<Long> db =
>> ds.apply(MapElements.into(TypeDescriptors.longs()).via((Long x) -> x
>> * 2));
>>
>> db.apply(ParDo.of(new Printer<>("BeamSimple: %d\n")));
>>
>>
>> p.run().waitUntilFinish();
>>
>> return null;
>> }
>> }
>
> I'm using Beam version 2.27.0, but it also happens on 2.29.0. The
> Flink version is 1.12.1.
>
> And the error message I'm getting is:
>
>
>> dragon % JAVA_HOME="/usr/lib/jvm/java-8-openjdk"
>> ~/madsci/thesis/world/tools/flink/bin/flink info -p 1
>> ~/madsci/thesis/codes/mezurilo/target/mezurilo-bundled-1.0-SNAPSHOT.jar
>> BeamSimple
>> Picked up _JAVA_OPTIONS: -Dawt.useSystemAAFontSettings=on
>> -Dswing.aatext=true
>>
>> ------------------------------------------------------------
>> The program finished with the following exception:
>>
>> org.apache.flink.client.program.ProgramInvocationException: The
>> program plan could not be fetched - the program aborted pre-maturely.
>>
>> Classpath:
>> [file:/home/rhermes/madsci/thesis/codes/mezurilo/target/mezurilo-bundled-1.0-SNAPSHOT.jar]
>>
>> System.out: (none)
>>
>> System.err: java.lang.NullPointerException
>> at
>> org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkNotNull(Preconditions.java:877)
>> at
>> org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.SingletonImmutableList.<init>(SingletonImmutableList.java:38)
>> at
>> org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList.of(ImmutableList.java:94)
>> at
>> org.apache.beam.sdk.coders.NullableCoder.getCoderArguments(NullableCoder.java:102)
>> at
>> org.apache.beam.sdk.coders.StructuredCoder.getComponents(StructuredCoder.java:49)
>> at
>> org.apache.beam.sdk.coders.StructuredCoder.hashCode(StructuredCoder.java:69)
>> at java.util.AbstractList.hashCode(AbstractList.java:541)
>> at
>> org.apache.beam.sdk.coders.StructuredCoder.hashCode(StructuredCoder.java:69)
>> at java.util.AbstractList.hashCode(AbstractList.java:541)
>> at
>> org.apache.beam.sdk.coders.StructuredCoder.hashCode(StructuredCoder.java:69)
>> at
>> org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Hashing.smearedHash(Hashing.java:54)
>> at
>> org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.HashBiMap.get(HashBiMap.java:254)
>> at
>> org.apache.beam.runners.core.construction.SdkComponents.registerCoder(SdkComponents.java:265)
>> at
>> org.apache.beam.runners.core.construction.ParDoTranslation.translateParDo(ParDoTranslation.java:231)
>> at
>> org.apache.beam.runners.core.construction.ParDoTranslation.translateParDo(ParDoTranslation.java:210)
>> at
>> org.apache.beam.runners.core.construction.ParDoTranslation$ParDoTranslator.translate(ParDoTranslation.java:176)
>> at
>> org.apache.beam.runners.core.construction.PTransformTranslation.toProto(PTransformTranslation.java:239)
>> at
>> org.apache.beam.runners.core.construction.ParDoTranslation.getParDoPayload(ParDoTranslation.java:743)
>> at
>> org.apache.beam.runners.core.construction.ParDoTranslation.isSplittable(ParDoTranslation.java:758)
>> at
>> org.apache.beam.runners.core.construction.PTransformMatchers$6.matches(PTransformMatchers.java:274)
>> at
>> org.apache.beam.sdk.Pipeline$2.visitPrimitiveTransform(Pipeline.java:289)
>> at
>> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:587)
>> at
>> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:579)
>> at
>> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:579)
>> at
>> org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:579)
>> at
>> org.apache.beam.sdk.runners.TransformHierarchy$Node.access$500(TransformHierarchy.java:239)
>> at
>> org.apache.beam.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:213)
>> at
>> org.apache.beam.sdk.Pipeline.traverseTopologically(Pipeline.java:468)
>> at org.apache.beam.sdk.Pipeline.replace(Pipeline.java:267)
>> at org.apache.beam.sdk.Pipeline.replaceAll(Pipeline.java:217)
>> at
>> org.apache.beam.runners.flink.FlinkPipelineExecutionEnvironment.translate(FlinkPipelineExecutionEnvironment.java:118)
>> at
>> org.apache.beam.runners.flink.FlinkRunner.run(FlinkRunner.java:92)
>> at org.apache.beam.sdk.Pipeline.run(Pipeline.java:322)
>> at org.apache.beam.sdk.Pipeline.run(Pipeline.java:308)
>> at
>> no.spaeren.thesis.benchmarks.beam.BeamSimple.call(BeamSimple.java:45)
>> at
>> no.spaeren.thesis.benchmarks.beam.BeamSimple.call(BeamSimple.java:18)
>> at picocli.CommandLine.executeUserObject(CommandLine.java:1853)
>> at picocli.CommandLine.access$1100(CommandLine.java:145)
>> at
>> picocli.CommandLine$RunLast.executeUserObjectOfLastSubcommandWithSameParent(CommandLine.java:2243)
>> at picocli.CommandLine$RunLast.handle(CommandLine.java:2237)
>> at picocli.CommandLine$RunLast.handle(CommandLine.java:2201)
>> at
>> picocli.CommandLine$AbstractParseResultHandler.execute(CommandLine.java:2068)
>> at picocli.CommandLine.execute(CommandLine.java:1978)
>> at no.spaeren.thesis.App.main(App.java:24)
>> at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>> at
>> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
>> at
>> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>> at java.lang.reflect.Method.invoke(Method.java:498)
>> at
>> org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:343)
>> at
>> org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:213)
>> at
>> org.apache.flink.client.program.PackagedProgramUtils.getPipelineFromProgram(PackagedProgramUtils.java:158)
>> at
>> org.apache.flink.client.cli.CliFrontend.info(CliFrontend.java:366)
>> at
>> org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1067)
>> at
>> org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1136)
>> at
>> org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
>> at
>> org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1136)
>>
>> at
>> org.apache.flink.client.program.PackagedProgramUtils.generateException(PackagedProgramUtils.java:264)
>> at
>> org.apache.flink.client.program.PackagedProgramUtils.getPipelineFromProgram(PackagedProgramUtils.java:184)
>> at
>> org.apache.flink.client.cli.CliFrontend.info(CliFrontend.java:366)
>> at
>> org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1067)
>> at
>> org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1136)
>> at
>> org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
>> at
>> org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1136)
>
> Any help with this would be greatly appriciated.
>
> Best regards,
> Teodor Spæren