flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Matthias J. Sax" <mj...@apache.org>
Subject Re: question on flink-storm-examples
Date Thu, 03 Sep 2015 08:11:05 GMT
One more remark that just came to my mind. There is a storm-hdfs module
available: https://github.com/apache/storm/tree/master/external/storm-hdfs

Maybe you can use it. It would be great if you could give feedback if
this works for you.

-Matthias

On 09/02/2015 10:52 AM, Matthias J. Sax wrote:
> Hi,
> StormFileSpout uses a simple FileReader internally an cannot deal with
> HDFS. It would be a nice extension to have. I just opened a JIRA for it:
> https://issues.apache.org/jira/browse/FLINK-2606
> 
> Jerry, feel to work in this feature and contribute code to Flink ;)
> 
> -Matthias
> 
> On 09/02/2015 07:52 AM, Aljoscha Krettek wrote:
>> Hi Jerry,
>> unfortunately, it seems that the StormFileSpout can only read files from
>> a local filesystem, not from HDFS. Maybe Matthias has something in the
>> works for that.
>>
>> Regards,
>> Aljoscha
>>
>> On Tue, 1 Sep 2015 at 23:33 Jerry Peng <jerry.boyang.peng@gmail.com
>> <mailto:jerry.boyang.peng@gmail.com>> wrote:
>>
>>     Ya that what I did and everything seems execute fine but when I try
>>     to run the WordCount-StormTopology with a file on hfs I get
>>     a java.io.FileNotFoundException :
>>
>>     java.lang.RuntimeException: java.io.FileNotFoundException:
>>     /home/jerrypeng/hadoop/hadoop_dir/data/data.txt (No such file or
>>     directory)
>>
>>     at
>>     org.apache.flink.stormcompatibility.util.StormFileSpout.open(StormFileSpout.java:50)
>>
>>     at
>>     org.apache.flink.stormcompatibility.wrappers.AbstractStormSpoutWrapper.run(AbstractStormSpoutWrapper.java:102)
>>
>>     at
>>     org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:57)
>>
>>     at
>>     org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:58)
>>
>>     at
>>     org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:172)
>>
>>     at org.apache.flink.runtime.taskmanager.Task.run(Task.java:581)
>>
>>     at java.lang.Thread.run(Thread.java:745)
>>
>>     Caused by: java.io.FileNotFoundException:
>>     /home/jerrypeng/hadoop/hadoop_dir/data/data.txt (No such file or
>>     directory)
>>
>>     at java.io.FileInputStream.open(Native Method)
>>
>>     at java.io.FileInputStream.<init>(FileInputStream.java:138)
>>
>>     at java.io.FileInputStream.<init>(FileInputStream.java:93)
>>
>>     at java.io.FileReader.<init>(FileReader.java:58)
>>
>>     at
>>     org.apache.flink.stormcompatibility.util.StormFileSpout.open(StormFileSpout.java:48)
>>
>>
>>
>>     However I have that file on my hdfs namespace:
>>
>>
>>     $ hadoop fs -ls -R /
>>
>>     15/09/01 21:25:11 WARN util.NativeCodeLoader: Unable to load
>>     native-hadoop library for your platform... using builtin-java
>>     classes where applicable
>>
>>     drwxr-xr-x   - jerrypeng supergroup          0 2015-08-21 14:40 /home
>>
>>     drwxr-xr-x   - jerrypeng supergroup          0 2015-08-21 14:40
>>     /home/jerrypeng
>>
>>     drwxr-xr-x   - jerrypeng supergroup          0 2015-08-21 14:41
>>     /home/jerrypeng/hadoop
>>
>>     drwxr-xr-x   - jerrypeng supergroup          0 2015-08-21 14:40
>>     /home/jerrypeng/hadoop/dir
>>
>>     drwxr-xr-x   - jerrypeng supergroup          0 2015-08-24 16:06
>>     /home/jerrypeng/hadoop/hadoop_dir
>>
>>     drwxr-xr-x   - jerrypeng supergroup          0 2015-09-01 20:48
>>     /home/jerrypeng/hadoop/hadoop_dir/data
>>
>>     -rw-r--r--   3 jerrypeng supergroup      18552 2015-09-01 19:18
>>     /home/jerrypeng/hadoop/hadoop_dir/data/data.txt
>>
>>     -rw-r--r--   3 jerrypeng supergroup          0 2015-09-01 20:48
>>     /home/jerrypeng/hadoop/hadoop_dir/data/result.txt
>>
>>     drwxr-xr-x   - jerrypeng supergroup          0 2015-08-21 14:41
>>     /home/jerrypeng/hadoop/hadoop_dir/dir1
>>
>>     drwxr-xr-x   - jerrypeng supergroup          0 2015-08-24 15:59
>>     /home/jerrypeng/hadoop/hadoop_dir/test
>>
>>     -rw-r--r--   3 jerrypeng supergroup         32 2015-08-24 15:59
>>     /home/jerrypeng/hadoop/hadoop_dir/test/filename.txt
>>
>>
>>     Any idea what's going on?
>>
>>
>>     On Tue, Sep 1, 2015 at 4:20 PM, Matthias J. Sax
>>     <mjsax@informatik.hu-berlin.de
>>     <mailto:mjsax@informatik.hu-berlin.de>> wrote:
>>
>>         You can use "bin/flink cancel JOBID" or JobManager WebUI to
>>         cancel the
>>         running job.
>>
>>         The exception you see, occurs in
>>         FlinkSubmitter.killTopology(...) which
>>         is not used by "bin/flink cancel" or JobMaanger WebUI.
>>
>>         If you compile the example you yourself, just remove the call to
>>         killTopology().
>>
>>         -Matthias
>>
>>         On 09/01/2015 11:16 PM, Matthias J. Sax wrote:
>>         > Oh yes. I forgot about this. I have already a fix for it in a
>>         pending
>>         > pull request... I hope that this PR is merged soon...
>>         >
>>         > If you want to observe the progress, look here:
>>         > https://issues.apache.org/jira/browse/FLINK-2111
>>         > and
>>         > https://issues.apache.org/jira/browse/FLINK-2338
>>         >
>>         > This PR, resolves both and fixed the problem you observed:
>>         > https://github.com/apache/flink/pull/750
>>         >
>>         > -Matthias
>>         >
>>         >
>>         > On 09/01/2015 11:09 PM, Jerry Peng wrote:
>>         >> Hello,
>>         >>
>>         >> I corrected the number of slots for each task manager but now
>>         when I try
>>         >> to run the WordCount-StormTopology, the job manager daemon on
>>         my master
>>         >> node crashes and I get this exception in the log:
>>         >>
>>         >> java.lang.Exception: Received a message
>>         >> CancelJob(6a4b9aa01ec87db20060210e5b36065e) without a leader
>>         session ID,
>>         >> even though the message requires a leader session ID.
>>         >>
>>         >> at
>>         >>
>>         org.apache.flink.runtime.LeaderSessionMessageFilter$$anonfun$receive$1.applyOrElse(LeaderSessionMessageFilter.scala:41)
>>         >>
>>         >> at
>>         >>
>>         scala.runtime.AbstractPartialFunction$mcVL$sp.apply$mcVL$sp(AbstractPartialFunction.scala:33)
>>         >>
>>         >> at
>>         >>
>>         scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:33)
>>         >>
>>         >> at
>>         >>
>>         scala.runtime.AbstractPartialFunction$mcVL$sp.apply(AbstractPartialFunction.scala:25)
>>         >>
>>         >> at
>>         org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:33)
>>         >>
>>         >> at
>>         org.apache.flink.runtime.LogMessages$$anon$1.apply(LogMessages.scala:28)
>>         >>
>>         >> at
>>         scala.PartialFunction$class.applyOrElse(PartialFunction.scala:118)
>>         >>
>>         >> at
>>         >>
>>         org.apache.flink.runtime.LogMessages$$anon$1.applyOrElse(LogMessages.scala:28)
>>         >>
>>         >> at akka.actor.Actor$class.aroundReceive(Actor.scala:465)
>>         >>
>>         >> at
>>         >>
>>         org.apache.flink.runtime.jobmanager.JobManager.aroundReceive(JobManager.scala:104)
>>         >>
>>         >> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516)
>>         >>
>>         >> at akka.actor.ActorCell.invoke(ActorCell.scala:487)
>>         >>
>>         >> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:254)
>>         >>
>>         >> at akka.dispatch.Mailbox.run(Mailbox.scala:221)
>>         >>
>>         >> at akka.dispatch.Mailbox.exec(Mailbox.scala:231)
>>         >>
>>         >> at
>>         scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
>>         >>
>>         >> at
>>         >>
>>         scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
>>         >>
>>         >> at
>>         scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
>>         >>
>>         >> at
>>         >>
>>         scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
>>         >>
>>         >>
>>         >> It seems to have something to do with canceling of the
>>         topology after
>>         >> the sleep.  Any ideas?
>>         >>
>>         >>
>>         >> Best,
>>         >>
>>         >>
>>         >> Jerry
>>         >>
>>         >>
>>         >> On Tue, Sep 1, 2015 at 3:33 PM, Matthias J. Sax
>>         >> <mjsax@informatik.hu-berlin.de
>>         <mailto:mjsax@informatik.hu-berlin.de>
>>         <mailto:mjsax@informatik.hu-berlin.de
>>         <mailto:mjsax@informatik.hu-berlin.de>>>
>>         >> wrote:
>>         >>
>>         >>     Yes. That is what I expected.
>>         >>
>>         >>     JobManager cannot start the job, due to less task slots.
>>         It logs the
>>         >>     exception NoResourceAvailableException (it is not shown
>>         in stdout; see
>>         >>     "log" folder). There is no feedback to Flink CLI that the
>>         job could not
>>         >>     be started.
>>         >>
>>         >>     Furthermore, WordCount-StormTopology sleeps for 5 seconds
>>         and tries to
>>         >>     "kill" the job. However, because the job was never
>>         started, there is a
>>         >>     NotAliveException which in print to stdout.
>>         >>
>>         >>     -Matthias
>>         >>
>>         >>
>>         >>
>>         >>     On 09/01/2015 10:26 PM, Jerry Peng wrote:
>>         >>     > When I run WordCount-StormTopology I get the following
>>         exception:
>>         >>     >
>>         >>     > ~/flink/bin/flink run WordCount-StormTopology.jar
>>         >>     > hdfs:///home/jerrypeng/hadoop/hadoop_dir/data/data.txt
>>         >>     > hdfs:///home/jerrypeng/hadoop/hadoop_dir/data/results.txt
>>         >>     >
>>         >>     >
>>         org.apache.flink.client.program.ProgramInvocationException: The main
>>         >>     > method caused an error.
>>         >>     >
>>         >>     > at
>>         >>     >
>>         >>   
>>          org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:452)
>>         >>     >
>>         >>     > at
>>         >>     >
>>         >>   
>>          org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:353)
>>         >>     >
>>         >>     > at
>>         org.apache.flink.client.program.Client.run(Client.java:278)
>>         >>     >
>>         >>     > at
>>         >>   
>>          org.apache.flink.client.CliFrontend.executeProgram(CliFrontend.java:631)
>>         >>     >
>>         >>     > at
>>         org.apache.flink.client.CliFrontend.run(CliFrontend.java:319)
>>         >>     >
>>         >>     > at
>>         >>   
>>          org.apache.flink.client.CliFrontend.parseParameters(CliFrontend.java:954)
>>         >>     >
>>         >>     > at
>>         org.apache.flink.client.CliFrontend.main(CliFrontend.java:1004)
>>         >>     >
>>         >>     > Caused by: NotAliveException(msg:null)
>>         >>     >
>>         >>     > at
>>         >>     >
>>         >>   
>>          org.apache.flink.stormcompatibility.api.FlinkClient.killTopologyWithOpts(FlinkClient.java:209)
>>         >>     >
>>         >>     > at
>>         >>     >
>>         >>   
>>          org.apache.flink.stormcompatibility.api.FlinkClient.killTopology(FlinkClient.java:203)
>>         >>     >
>>         >>     > at
>>         >>     >
>>         >>   
>>          org.apache.flink.stormcompatibility.wordcount.StormWordCountRemoteBySubmitter.main(StormWordCountRemoteBySubmitter.java:80)
>>         >>     >
>>         >>     > 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:483)
>>         >>     >
>>         >>     > at
>>         >>     >
>>         >>   
>>          org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:437)
>>         >>     >
>>         >>     > ... 6 more
>>         >>     >
>>         >>     >
>>         >>     > The exception above occurred while trying to run your
>>         command.
>>         >>     >
>>         >>     >
>>         >>     > Any idea how to fix this?
>>         >>     >
>>         >>     > On Tue, Sep 1, 2015 at 3:10 PM, Matthias J. Sax
>>         >>     > <mjsax@informatik.hu-berlin.de
>>         <mailto:mjsax@informatik.hu-berlin.de>
>>         >>     <mailto:mjsax@informatik.hu-berlin.de
>>         <mailto:mjsax@informatik.hu-berlin.de>>
>>         >>     <mailto:mjsax@informatik.hu-berlin.de
>>         <mailto:mjsax@informatik.hu-berlin.de>
>>         >>     <mailto:mjsax@informatik.hu-berlin.de
>>         <mailto:mjsax@informatik.hu-berlin.de>>>>
>>         >>     > wrote:
>>         >>     >
>>         >>     >     Hi Jerry,
>>         >>     >
>>         >>     >     WordCount-StormTopology uses a hard coded dop of 4.
>>         If you
>>         >>     start up
>>         >>     >     Flink in local mode (bin/start-local-streaming.sh),
>>         you need
>>         >>     to increase
>>         >>     >     the number of task slots to at least 4 in
>>         conf/flink-conf.yaml
>>         >>     before
>>         >>     >     starting Flink -> taskmanager.numberOfTaskSlots
>>         >>     >
>>         >>     >     You should actually see the following exception in
>>         >>     >     log/flink-...-jobmanager-...log
>>         >>     >
>>         >>     >     > NoResourceAvailableException: Not enough free
>>         slots available to
>>         >>     >     run the job. You can decrease the operator
>>         parallelism or increase
>>         >>     >     the number of slots per TaskManager in the
>>         configuration.
>>         >>     >
>>         >>     >     WordCount-StormTopology does use
>>         StormWordCountRemoteBySubmitter
>>         >>     >     internally. So, you do use it already ;)
>>         >>     >
>>         >>     >     I am not sure what you mean by "get rid of
>>         KafkaSource"? It is
>>         >>     still in
>>         >>     >     the code base. Which version to you use? In
>>         >>     flink-0.10-SNAPSHOT it is
>>         >>     >     located in submodule "flink-connector-kafka" (which
is
>>         >>     submodule of
>>         >>     >     "flink-streaming-connector-parent" -- which is
>>         submodule of
>>         >>     >     "flink-streamping-parent").
>>         >>     >
>>         >>     >
>>         >>     >     -Matthias
>>         >>     >
>>         >>     >
>>         >>     >     On 09/01/2015 09:40 PM, Jerry Peng wrote:
>>         >>     >     > Hello,
>>         >>     >     >
>>         >>     >     > I have some questions regarding how to run one
of the
>>         >>     >     > flink-storm-examples, the WordCountTopology. 
How
>>         should I
>>         >>     run the
>>         >>     >     job?
>>         >>     >     > On github its says I should just execute
>>         >>     >     > bin/flink run example.jar but when I execute:
>>         >>     >     >
>>         >>     >     > bin/flink run WordCount-StormTopology.jar
>>         >>     >     >
>>         >>     >     > nothing happens.  What am I doing wrong? and How
>>         can I run the
>>         >>     >     > WordCounttopology via
>>         StormWordCountRemoteBySubmitter?
>>         >>     >     >
>>         >>     >     > Also why did you guys get rid of the KafkaSource
>>         class?  What is
>>         >>     >     the API
>>         >>     >     > now for subscribing to a kafka source?
>>         >>     >     >
>>         >>     >     > Best,
>>         >>     >     >
>>         >>     >     > Jerry
>>         >>     >
>>         >>     >
>>         >>
>>         >>
>>         >
>>
>>
> 


Mime
View raw message