ZEPPELIN-44 Interpreter for Apache Flink

[Leemoonsoo]

Interpreter for Apache Flink.

Flink people helped a lot to write the interpreter. Thanks so much! Some codes are copied from Flink's development branch. Once Flink releases 0.9, copied code and snapshot repository configuration will be removed.

Build

if there're no options, by default it is building against flink 0.9.0-milestone-1.
With combination of Zeppelin, it is good idea to use 0.9-SNAPSHOT, because of it support .collect() that helps really a lot to get results data and display it on Zeppelin.

So, you might want to build in this way,

mvn package -Dflink.version=0.9-SNAPSHOT -DskipTests

Screenshot

[jongyoul]

@Leemoonsoo Wow, Does it work?

[felixcheung]

'jar'?

[Leemoonsoo]

@jongyoul Yes it works!

[rmetzger]

I'm very excited to see Flink support in Zeppelin 👍

[jongyoul]

@Leemoonsoo Great!!

[Leemoonsoo]

This PR has some test using FlinkMiniCluster.
Tests are fine when i run them on my IDE.
But if i run them using maven, it fails. (CI is failing, too)

I think everything is ready, except for this test failure. Any advice is very appreciated.
here's failure message

-------------------------------------------------------
T E S T S
-------------------------------------------------------
Running org.apache.zeppelin.flink.FlinkInterpreterTest
SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/home/travis/.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:/home/travis/build/apache/incubator-zeppelin/zeppelin-interpreter/target/lib/slf4j-log4j12-1.7.10.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]
Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 106.131 sec <<< FAILURE! - in org.apache.zeppelin.flink.FlinkInterpreterTest
org.apache.zeppelin.flink.FlinkInterpreterTest Time elapsed: 106.131 sec <<< ERROR!
java.util.concurrent.TimeoutException: Futures timed out after [100000 milliseconds]
at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:219)
at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:223)
at scala.concurrent.Await$$anonfun$result$1.apply(package.scala:107)
at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn(BlockContext.scala:53)
at scala.concurrent.Await$.result(package.scala:107)
at akka.remote.Remoting.start(Remoting.scala:180)
at akka.remote.RemoteActorRefProvider.init(RemoteActorRefProvider.scala:184)
at akka.actor.ActorSystemImpl.liftedTree2$1(ActorSystem.scala:619)
at akka.actor.ActorSystemImpl._start$lzycompute(ActorSystem.scala:616)
at akka.actor.ActorSystemImpl._start(ActorSystem.scala:616)
at akka.actor.ActorSystemImpl.start(ActorSystem.scala:633)
at akka.actor.ActorSystem$.apply(ActorSystem.scala:142)
at akka.actor.ActorSystem$.apply(ActorSystem.scala:119)
at akka.actor.ActorSystem$.create(ActorSystem.scala:67)
at org.apache.flink.runtime.akka.AkkaUtils$.createActorSystem(AkkaUtils.scala:82)
at org.apache.flink.runtime.minicluster.FlinkMiniCluster.startJobManagerActorSystem(FlinkMiniCluster.scala:105)
at org.apache.flink.runtime.minicluster.FlinkMiniCluster.<init>(FlinkMiniCluster.scala:61)
at org.apache.flink.runtime.minicluster.LocalFlinkMiniCluster.<init>(LocalFlinkMiniCluster.scala:44)
at org.apache.zeppelin.flink.FlinkInterpreter.startFlinkMiniCluster(FlinkInterpreter.java:297)
at org.apache.zeppelin.flink.FlinkInterpreter.initializeFlinkEnv(FlinkInterpreter.java:152)
at org.apache.zeppelin.flink.FlinkInterpreter.open(FlinkInterpreter.java:115)
at org.apache.zeppelin.flink.FlinkInterpreterTest.setUp(FlinkInterpreterTest.java:40)
Results :
Tests in error: FlinkInterpreterTest.setUp:40 » Timeout Futures timed out after [100000 millis...

[fhueske]

I forwarded the error to the Flink dev mailing list. Hopefully somebody from the Flink community can help with this.

[asfbot]

Fabian Hueske on dev@flink.apache.org replies:
Hi Flink folks,

the Flink interpreter PR for Apache Zeppelin is blocked by a failing test
case (see below).
Does anybody have an idea what is going on and can maybe help to resolve
the problem?

Thanks, Fabian

[asfbot]

Till Rohrmann on dev@flink.apache.org replies:
I'll try to reproduce this problem locally on my machine.
r
84
j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
er/target/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.c=
lass]
31
0
cala:53)
)
a:82)
rSystem(FlinkMiniCluster.scala:105)
ster.scala:61)
inkMiniCluster.scala:44)
erpreter.java:297)
reter.java:152)
)
.java:40)
t after
ur
e
se

[asfbot]

Stephan Ewen on dev@flink.apache.org replies:
The actor system fails to start. Here are the suspicious log lines:

SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/home/travis/.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:/home/travis/build/
apache/incubator-zeppelin/zeppelin-interpreter/target/
lib/slf4j-log4j12-1.7.10.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]

Seems that SLF4J has more than one binding. I remember that Akka crashed
because of that before.

As a simple fix, can you try and exclude the SLF4J jar from your build
somehow? Or set it to "provided" in the Flink POM?
:
st
e
84
ce
j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
er/target/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.c=
lass]
]
31
cala:53)
4)
a:82)
rSystem(FlinkMiniCluster.scala:105)
ster.scala:61)
inkMiniCluster.scala:44)
erpreter.java:297)
reter.java:152)
)
.java:40)
out after

[asfbot]

Till Rohrmann on dev@flink.apache.org replies:
Hmm that is interesting. I always thought that in the case of multiple
SLF4J bindings, slf4j will simply pick one of them (first, last, random).
84
j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
er/target/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.c=
lass]
)
cala:53)
a:82)
rSystem(FlinkMiniCluster.scala:105)
ster.scala:61)
inkMiniCluster.scala:44)
erpreter.java:297)
reter.java:152)
)
.java:40)
d out after
e

[asfbot]

Stephan Ewen on dev@flink.apache.org replies:
True, may still have another issue.
]
d
g
84
j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.class]
er/target/lib/slf4j-log4j12-1.7.10.jar!/org/slf4j/impl/StaticLoggerBinder.c=
lass]
r
t
9)
cala:53)
4)
a:82)
rSystem(FlinkMiniCluster.scala:105)
ster.scala:61)
inkMiniCluster.scala:44)
erpreter.java:297)
reter.java:152)
)
.java:40)
med out
,

[tillrohrmann]

I figured out the problem: It is a wrong protobuf-java version. flink-runtime depends on akka-remote_2.10:2.3.7 which depends on protobuf-java:2.5.0. However, in the Zeppelin parent pom the protobuf-java version is manually set to 2.4.1. There is a comment which declares this as a work-around to make Zeppelin build with Hadoop/YARN 2.2.

As a quick work-around you can set the protobuf-java version to 2.5.0 in the zeppelin-flink pom.xml. I'll also open a JIRA to shade the protobuf and akka dependency away in the flink project. This should solve this issue permanently.

[Leemoonsoo]

Really appreciate for investigating the problem.
I think this is wrong in Zeppelin side. That, Zeppelin used to have dependency configuration for spark in pom.xml not in spark/pom.xml. I think it should be moved into spark/pom.xml to not interfere the other modules.
I'll create issue for it in Zeppelin side.

[rmetzger]

Okay, so this pull request is not blocked by anything from the Flink side right now?

[Leemoonsoo]

@rmetzger Not blocked by anything from Flink side.
I'm testing the branch with 9 node Flink cluster environment. Works well but looks like need special build command. I'll update build configuration.

[Leemoonsoo]

After resolving Zeppelin side issue by #88, Tests are passing.
Ready to merge.

[Leemoonsoo]

Merging it if there is no more discussions :-)

[hsaputra]

W000t!