dataartisans / flink-streaming-demo Goto Github PK

hi,thanks a lot for this repository.
will it support elastic 2.x? I'm asking because I know that elastic 2 'update' works differently

Lior

how can i set the map picture for "Shaded Geohash Grid"

Although this is a very nice example for connecting flink to elastic, I don't get the underlying assumption here: For what do I need Flink here? What's wrong about writing the actual rides into elastic and then use a geohashgrid to count by target-location?

Currently, the flink setup hardcodes the 100x100meters cell-size into the counts, simply pushing the rides into ES not only get's rid of a hefty piece infrastructure, but also allows more flexibility for querying, visualisations: You're basically using flink to destroy information that for other scenarios might still be useful.

I really do not want to be disrespectful of all the effort you put into this example, but I'm really trying to figure out, where flink fits into an architecture involving ES and where the boundaries between the two systems are...

I get a null pointer exception while executing the sample code from the examples folder. The error is as :

Exception in thread "main" java.lang.NullPointerException
    at org.apache.flink.runtime.minicluster.FlinkMiniCluster.startWebServer(FlinkMiniCluster.scala:295)
    at org.apache.flink.runtime.minicluster.FlinkMiniCluster.start(FlinkMiniCluster.scala:268)
    at org.apache.flink.runtime.minicluster.FlinkMiniCluster.start(FlinkMiniCluster.scala:226)
    at org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:101)
    at com.dataartisans.flink_demo.examples.TotalArrivalCount$.main(TotalArrivalCount.scala:96)
    at com.dataartisans.flink_demo.examples.TotalArrivalCount.main(TotalArrivalCount.scala)
    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 com.intellij.rt.execution.application.AppMain.main(AppMain.java:140)

Process finished with exit code 130

Am I missing something ??

The flink version is quite old <flink.version>0.10.0</flink.version> and there are other articles.

• https://github.com/ververica/flink-training-exercises which can work with flink 1.8.1.
• https://github.com/aws-samples/flink-stream-processing-refarch

Please advise if there is a plan to update this to work with Flink 1.8.x or later.

Hello,
Apologies if this is a very basic question, I am having an issue running your code EarlyArrivalCount; I followed all the steps but when I try running the code, I get the following error message (in Linux and in Windows). Please note that when I set
val writeToElasticsearch = false, it runs and echoes data in the console.
Looks like it's complaining about elasticsearch, I've confirmed that it is running, I also had to make the change recommended in one of the posting:
config.setInteger(ConfigConstants.JOB_MANAGER_WEB_PORT_KEY,9999)
Any help, advice will be greatly appreciated, Thanks in advance.
Frank

Error Message
log4j:WARN No appenders could be found for logger (org.apache.flink.api.java.typeutils.TypeExtractor).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
05/09/2016 10:37:23 Job execution switched to status RUNNING.
05/09/2016 10:37:23 Source: Custom Source(1/1) switched to SCHEDULED
05/09/2016 10:37:23 Source: Custom Source(1/1) switched to DEPLOYING
05/09/2016 10:37:23 Filter -> Filter -> Map(1/1) switched to SCHEDULED
05/09/2016 10:37:23 Filter -> Filter -> Map(1/1) switched to DEPLOYING
05/09/2016 10:37:23 TriggerWindow(SlidingTimeWindows(900000, 300000), com.dataartisans.flink_demo.examples.EarlyArrivalCount$EarlyCountTrigger@1f1c7bf6, WindowApply at apply(WindowedStream.scala:209)) -> Map -> (Sink: Unnamed, Sink: Unnamed)(1/1) switched to SCHEDULED
05/09/2016 10:37:23 TriggerWindow(SlidingTimeWindows(900000, 300000), com.dataartisans.flink_demo.examples.EarlyArrivalCount$EarlyCountTrigger@1f1c7bf6, WindowApply at apply(WindowedStream.scala:209)) -> Map -> (Sink: Unnamed, Sink: Unnamed)(1/1) switched to DEPLOYING
05/09/2016 10:37:23 Source: Custom Source(1/1) switched to RUNNING
05/09/2016 10:37:23 TriggerWindow(SlidingTimeWindows(900000, 300000), com.dataartisans.flink_demo.examples.EarlyArrivalCount$EarlyCountTrigger@1f1c7bf6, WindowApply at apply(WindowedStream.scala:209)) -> Map -> (Sink: Unnamed, Sink: Unnamed)(1/1) switched to RUNNING
05/09/2016 10:37:23 Filter -> Filter -> Map(1/1) switched to RUNNING
(49551,1356998700000,GeoPoint(-73.97789764404297,40.75187683105469),2)
05/09/2016 10:37:31 TriggerWindow(SlidingTimeWindows(900000, 300000), com.dataartisans.flink_demo.examples.EarlyArrivalCount$EarlyCountTrigger@1f1c7bf6, WindowApply at apply(WindowedStream.scala:209)) -> Map -> (Sink: Unnamed, Sink: Unnamed)(1/1) switched to FAILED
java.lang.RuntimeException: Could not forward element to next operator
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:319)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:300)
at org.apache.flink.streaming.runtime.io.CollectorWrapper.collect(CollectorWrapper.java:48)
at org.apache.flink.streaming.runtime.io.CollectorWrapper.collect(CollectorWrapper.java:29)
at org.apache.flink.streaming.api.operators.TimestampedCollector.collect(TimestampedCollector.java:51)
at com.dataartisans.flink_demo.examples.EarlyArrivalCount$$anonfun$5.apply(EarlyArrivalCount.scala:96)
at com.dataartisans.flink_demo.examples.EarlyArrivalCount$$anonfun$5.apply(EarlyArrivalCount.scala:95)
at org.apache.flink.streaming.api.scala.WindowedStream$$anon$3.apply(WindowedStream.scala:206)
at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.emitWindow(WindowOperator.java:302)
at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.processTriggerResult(WindowOperator.java:335)
at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.processWatermark(WindowOperator.java:362)
at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:156)
at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:63)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:218)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.RuntimeException: Could not forward element to next operator
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:319)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:300)
at org.apache.flink.streaming.runtime.io.CollectorWrapper.collect(CollectorWrapper.java:48)
at org.apache.flink.streaming.runtime.io.CollectorWrapper.collect(CollectorWrapper.java:29)
at org.apache.flink.streaming.api.operators.StreamMap.processElement(StreamMap.java:37)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:316)
... 15 more
Caused by: org.elasticsearch.client.transport.NoNodeAvailableException: None of the configured nodes are available: []
at org.elasticsearch.client.transport.TransportClientNodesService.ensureNodesAreAvailable(TransportClientNodesService.java:305)
at org.elasticsearch.client.transport.TransportClientNodesService.execute(TransportClientNodesService.java:200)
at org.elasticsearch.client.transport.support.InternalTransportClient.execute(InternalTransportClient.java:106)
at org.elasticsearch.client.transport.support.InternalTransportClient.execute(InternalTransportClient.java:97)
at org.elasticsearch.client.support.AbstractClient.update(AbstractClient.java:122)
at org.elasticsearch.client.transport.TransportClient.update(TransportClient.java:345)
at com.dataartisans.flink_demo.sinks.ElasticsearchUpsertSink.invoke(ElasticsearchUpsertSink.scala:80)
at org.apache.flink.streaming.api.operators.StreamSink.processElement(StreamSink.java:37)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:316)
... 20 more

05/09/2016 10:37:31 Job execution switched to status FAILING.
05/09/2016 10:37:31 Source: Custom Source(1/1) switched to CANCELING
05/09/2016 10:37:31 Filter -> Filter -> Map(1/1) switched to CANCELING
05/09/2016 10:37:31 Filter -> Filter -> Map(1/1) switched to CANCELED
05/09/2016 10:37:31 Source: Custom Source(1/1) switched to CANCELED
05/09/2016 10:37:31 Job execution switched to status FAILED.

Origin words:
[Execute the main() method of one of the demo applications, for example com.dataartisans.flink_demo.examples.TotalArrivalCount.scala. Running an application will start a local Flink instance in the JVM process of your IDE.]

I've follow steps / READ-ME.txt , but I still do not understand what's the words meaning. There is no able to execution of main() method in the project. ( I even use the IntelliJ-IDEA/Netbeans/Eclipse , 3 IDES to check up the codes.) I also installed scala for just testing a single file. I'm confused up. Could you please be kind of making the intros more clearly? Thanks!

Hi,

I clone the repo https://github.com/dataArtisans/flink-streaming-demo.git.
After that I imported the project to an IDE. However i am getting compilation error, see screenshot below :

Also as suggested i run the project with the following configuration (screenshot below), however i end up getting an error "Error: Could not find or load main class com.dataartisans.flink_demo.examples.TotalArrivalCount.scala"