beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Davor Bonaci (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (BEAM-944) Spark runner causes an exception when creating pipeline options
Date Tue, 08 Nov 2016 18:37:59 GMT

    [ https://issues.apache.org/jira/browse/BEAM-944?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15648395#comment-15648395
] 

Davor Bonaci commented on BEAM-944:
-----------------------------------

I believe PipelineOptions will scan *registered* pipeline options only, plus those requested
by the user when calling {code}something.as(SomeOptions.class){code}. It scans Spark ones
because SparkRunner uses AutoService to register its options, https://github.com/apache/incubator-beam/blob/master/runners/spark/pom.xml#L165.
I believe this behavior is roughly correct; but perhaps it could be less eager in some cases.

It would be good to move {code}JavaSparkContext{code} elsewhere, in addition to {code}JavaStreamingListener{code}.
Basically, these aren't real pipeline options. They seem to be there because pipeline options
make it easy to pass things around. To be honest, there are many such examples everywhere,
this being just one of them: https://github.com/apache/incubator-beam/blob/465c7d1f1e4f77e3de218365005f06e8e4ec6167/sdks/java/core/src/main/java/org/apache/beam/sdk/options/GcsOptions.java#L46
-- just the impact is smaller because it doesn't normally causes an exception.

Is there a better way to pass these around?

> Spark runner causes an exception when creating pipeline options
> ---------------------------------------------------------------
>
>                 Key: BEAM-944
>                 URL: https://issues.apache.org/jira/browse/BEAM-944
>             Project: Beam
>          Issue Type: Bug
>          Components: runner-spark
>    Affects Versions: 0.3.0-incubating
>            Reporter: Davor Bonaci
>            Assignee: Amit Sela
>             Fix For: 0.4.0-incubating
>
>
> Running any example with *any* runner throws an exception as soon as it tries to construct
PipelineOptions as long as SparkRunner is on the class path:
> {code}
> mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.complete.game.LeaderBoard
-Dexec.args="--runner=DirectRunner"
> [INFO] Scanning for projects...
> [INFO]                                                                         
> [INFO] ------------------------------------------------------------------------
> [INFO] Building Tutorial 0.0.1-SNAPSHOT
> [INFO] ------------------------------------------------------------------------
> [INFO] 
> [INFO] --- maven-resources-plugin:2.6:resources (default-resources) @ Tutorial ---
> [WARNING] Using platform encoding (UTF-8 actually) to copy filtered resources, i.e. build
is platform dependent!
> [INFO] skip non existing resourceDirectory /Users/fjp/dev/mobile/src/main/resources
> [INFO] 
> [INFO] --- maven-compiler-plugin:3.3:compile (default-compile) @ Tutorial ---
> [INFO] No sources to compile
> [INFO] 
> [INFO] --- exec-maven-plugin:1.5.0:java (default-cli) @ Tutorial ---
> SLF4J: Class path contains multiple SLF4J bindings.
> SLF4J: Found binding in [jar:file:/Users/fjp/.m2/repository/org/slf4j/slf4j-log4j12/1.7.10/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: Found binding in [jar:file:/Users/fjp/.m2/repository/org/slf4j/slf4j-jdk14/1.7.5/slf4j-jdk14-1.7.5.jar!/org/slf4j/impl/StaticLoggerBinder.class]
> SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
> SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]
> [WARNING] 
> java.lang.reflect.InvocationTargetException
> 	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:497)
> 	at org.codehaus.mojo.exec.ExecJavaMojo$1.run(ExecJavaMojo.java:294)
> 	at java.lang.Thread.run(Thread.java:745)
> Caused by: java.lang.ExceptionInInitializerError
> 	at org.apache.beam.examples.complete.game.LeaderBoard.main(LeaderBoard.java:179)
> 	... 6 more
> Caused by: java.lang.TypeNotPresentException: Type org.apache.spark.streaming.api.java.JavaStreamingListener
not present
> 	at sun.reflect.generics.factory.CoreReflectionFactory.makeNamedType(CoreReflectionFactory.java:117)
> 	at sun.reflect.generics.visitor.Reifier.visitClassTypeSignature(Reifier.java:125)
> 	at sun.reflect.generics.tree.ClassTypeSignature.accept(ClassTypeSignature.java:49)
> 	at sun.reflect.generics.visitor.Reifier.reifyTypeArguments(Reifier.java:68)
> 	at sun.reflect.generics.visitor.Reifier.visitClassTypeSignature(Reifier.java:138)
> 	at sun.reflect.generics.tree.ClassTypeSignature.accept(ClassTypeSignature.java:49)
> 	at sun.reflect.generics.repository.ConstructorRepository.getParameterTypes(ConstructorRepository.java:94)
> 	at java.lang.reflect.Executable.getGenericParameterTypes(Executable.java:284)
> 	at java.lang.reflect.Method.getGenericParameterTypes(Method.java:282)
> 	at java.beans.FeatureDescriptor.getParameterTypes(FeatureDescriptor.java:387)
> 	at java.beans.PropertyDescriptor.findPropertyType(PropertyDescriptor.java:653)
> 	at java.beans.PropertyDescriptor.setWriteMethod(PropertyDescriptor.java:326)
> 	at java.beans.PropertyDescriptor.<init>(PropertyDescriptor.java:138)
> 	at org.apache.beam.sdk.options.PipelineOptionsFactory.getPropertyDescriptors(PipelineOptionsFactory.java:896)
> 	at org.apache.beam.sdk.options.PipelineOptionsFactory.validateClass(PipelineOptionsFactory.java:992)
> 	at org.apache.beam.sdk.options.PipelineOptionsFactory.validateWellFormed(PipelineOptionsFactory.java:627)
> 	at org.apache.beam.sdk.options.PipelineOptionsFactory.register(PipelineOptionsFactory.java:561)
> 	at org.apache.beam.sdk.options.PipelineOptionsFactory.initializeRegistry(PipelineOptionsFactory.java:587)
> 	at org.apache.beam.sdk.options.PipelineOptionsFactory.<clinit>(PipelineOptionsFactory.java:536)
> 	... 7 more
> Caused by: java.lang.ClassNotFoundException: org.apache.spark.streaming.api.java.JavaStreamingListener
> 	at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
> 	at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
> 	at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
> 	at java.lang.Class.forName0(Native Method)
> 	at java.lang.Class.forName(Class.java:348)
> 	at sun.reflect.generics.factory.CoreReflectionFactory.makeNamedType(CoreReflectionFactory.java:114)
> 	... 25 more
> {code}
> It is introduced by https://github.com/apache/incubator-beam/pull/1072 that was added
in mid-October, so it reproduces starting with 0.3.0-incubating, but not earlier.
> A possible workaround for users is to add this explicit dependency, which is marked as
provided in the Spark runner:
> {code}
> <dependency>
>   <groupId>org.apache.spark</groupId>
>   <artifactId>spark-streaming_2.10</artifactId>
>   <version>${spark.version}</version>
> </dependency>
> {code}
> On a quick glance, JavaStreamingListener is not a real pipeline option. Perhaps it can
be just removed from pipeline options, and passed around differently.
> Separately, we should perhaps re-think how this works with respect to dependencies in
general.
> Marking blocking for 0.4.0-incubating since it affects Spark quickstart.
> [~amitsela], what do you think?



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Mime
View raw message