Re: Handling large state (incremental snapshot?)

[Hironori Ogibayashi]

I tried RocksDB, but the result was almost the same.

I used the following code and put 2.6million distinct records into Kafka.
After processing all records, the state on the HDFS become about 250MB
and time needed for
the checkpoint was almost 5sec. Processing throughput was
FsStateBackend-> 8000msg/sec, RocksDBSsateBackend -> 9000msg/sec

---
env.setStateBackend(new
RocksDBStateBackend("hdfs://:8020/apps/flink/checkpoints"));

val stream = env
  .addSource(new FlinkKafkaConsumer09[String]("kafka.json2", new
SimpleStringSchema(), properties))
  .map(parseJson(_))
  .timeWindowAll(Time.of(10, TimeUnit.DAYS))
  .trigger(MyContinuousProcessingTimeTrigger.of(Time.seconds(5)))
  // count distinct values
  .fold(Set[String]()){(r,i) => { r + i}}
  .map{x => (System.currentTimeMillis(), x.size)}
  .addSink(new ElasticsearchSink(config, transports, new
IndexRequestBuilder[Tuple2[Long, Int]]  {
override def createIndexRequest(element: Tuple2[Long, Int],
ctx: RuntimeContext): IndexRequest = {
  val json = new HashMap[String, AnyRef]
  json.put("@timestamp", new Timestamp(element._1))
  json.put("count", element._2: java.lang.Integer)
  Requests.indexRequest.index("dummy3").`type`("my-type").source(json)
}
  }))
---

I guess this is because I used non-keyed stream, so I had one state
record with a big value (all distinct value).
I think copying all 250MB(or more) file to HDFS in every checkpoint
will be heavy, so I will try storing the distinct values
in the external datastore (e.g. redis).
Also, when incremental snapshot get implemented, I want to try.

Regards,
Hironori

2016-04-05 21:40 GMT+09:00 Hironori Ogibayashi :
> Aljoscha,
>
> Thank you for your quick response.
> Yes, I am using FsStateBackend, so I will try RocksDB backend.
>
> Regards,
> Hironori
>
> 2016-04-05 21:23 GMT+09:00 Aljoscha Krettek :
>> Hi,
>> I guess you are using the FsStateBackend, is that correct? You could try
>> using the RocksDB state backend:
>> https://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/state_backends.html#the-rocksdbstatebackend
>>
>> With this, throughput will be lower but the overhead per checkpoint could be
>> lower. Also, with this most of the file copying necessary for the checkpoint
>> will be done while data processing keeps running (asynchronous snapshot).
>>
>> As to incremental snapshots. I'm afraid this feature is not yet implemented
>> but we're working on it.
>>
>> Cheers,
>> Aljoscha
>>
>> On Tue, 5 Apr 2016 at 14:06 Hironori Ogibayashi 
>> wrote:
>>>
>>> Hello,
>>>
>>> I am trying to implement windowed distinct count on a stream. In this
>>> case, the state
>>> have to hold all distinct value in the window, so can be large.
>>>
>>> In my test, if the state size become about 400MB, checkpointing takes
>>> 40sec and spends most of Taskmanager's CPU.
>>> Are there any good way to handle this situation?
>>>
>>> Flink document mentions about incremental snapshot, and I am interested in
>>> it,
>>> but could not find how to enable it. (not implemented yet?)
>>>
>>> https://ci.apache.org/projects/flink/flink-docs-release-1.0/internals/stream_checkpointing.html
>>>
>>> Regards,
>>> Hironori

Re: Possible use case: Simulating iterative batch processing by rewinding source

[Christophe Salperwyck]

Hi,

I am interested too. For my part, I was thinking to use HBase as a backend
so that my data are stored sorted. Nice to have to generate timeseries in
the good order.

Cheers,
Christophe

2016-04-06 21:22 GMT+02:00 Raul Kripalani :

> Hello,
>
> I'm getting started with Flink for a use case that could leverage the
> window processing abilities of Flink that Spark does not offer.
>
> Basically I have dumps of timeseries data (10y in ticks) which I need to
> calculate many metrics in an exploratory manner based on event time. NOTE:
> I don't have the metrics beforehand, it's gonna be an exploratory and
> iterative data analytics effort.
>
> Flink doesn't seem to support windows on batch processing, so I'm thinking
> about emulating batch by using the Kafka stream connector and rewinding the
> data stream for every new metric that I calculate, to process the full
> timeseries series in a batch.
>
> Each metric I calculate should in turn be sent to another Kafka topic so I
> can use it in a subsequent processing batch, e.g.
>
> Iteration 1)   raw timeseries data ---> metric1
> Iteration 2)   raw timeseries data + metric1 (composite) ---> metric2
> Iteration 3)   metric1 + metric2 ---> metric3
> Iteration 4)   raw timeseries data + metric3 ---> metric4
> ...
>
> Does this sound like a usecase for Flink? Could you guide me a little bit
> on whether this is feasible currently?
>
> Cheers,
>
> *Raúl Kripalani*
> PMC & Committer @ Apache Ignite, Apache Camel | Integration, Big Data and
> Messaging Engineer
> http://about.me/raulkripalani | http://www.linkedin.com/in/raulkripalani
> Blog: raul.io
>  |
> twitter: @raulvk 
>

Re: Back Pressure details

[Zach Cox]

Yeah I don't think that's the case for my setup either :)  I wrote a simple
Flink job that just consumes from Kafka and sinks events/sec rate to
Graphite. That consumes from Kafka several orders of magnitude higher than
the job that also sinks to Elasticsearch. As you said, the downstream back
pressure must be also slowing down consumption from Kafka, even though job
manager UI doesn't show HIGH back pressure on the Kafka source.

Thanks for the details!

-Zach


On Wed, Apr 6, 2016 at 2:37 PM Ufuk Celebi  wrote:

> Ah sorry, I forgot to mention that in the docs.
>
> The way that data is pulled from Kafka is bypassing Flink's task
> Thread. The topic is consumed in a separate Thread and the task Thread
> is just waiting. That's why you don't see any back pressure for Kafka
> sources. I would expect your Kafka source to be back pressured as well
> then.
>
> In theory it is possible that the speed at which data is consumed in
> the source "matches" the speed of the back pressured operator down
> stream. That would result in a non back pressured source with a down
> stream back pressured task. But I don't think that's the case for your
> setup. ;-)
>
> On Wed, Apr 6, 2016 at 9:27 PM, Zach Cox  wrote:
> > The new back pressure docs are great, thanks Ufuk! I'm sure those will
> help
> > others as well.
> >
> > In the Source => A => B => Sink example, if A and B show HIGH back
> pressure,
> > should Source also show HIGH? In my case it's a Kafka source and
> > Elasticsearch sink. I know currently our Kafka can provide data at a much
> > higher rate than our Elasticsearch can ingest (I'm working on scaling up
> > Elasticsearch), just curious why the Kafka source wouldn't also show HIGH
> > back pressure.
> >
> > Thanks,
> > Zach
> >
> >
> > On Wed, Apr 6, 2016 at 5:36 AM Ufuk Celebi  wrote:
> >>
> >> Hey Zach,
> >>
> >> just added some documentation, which will be available in ~ 30 mins
> >> here:
> >>
> https://ci.apache.org/projects/flink/flink-docs-release-1.0/internals/back_pressure_monitoring.html
> >>
> >> If you think that something is missing there, I would appreciate some
> >> feedback. :-)
> >>
> >> Back pressure is determined by repeatedly calling getStackTrace() on
> >> the task Threads executing the job. By default, 100 times with 50ms
> >> delay between calls. If the task thread is stuck in an internal method
> >> call requesting buffers from the network stack, this indicates back
> >> pressure.
> >>
> >> The ratio you see tells you how many of the stack traces were stuck in
> >> that method (e.g. 1 out of 100) and the status codes just group those
> >> in a (hopefully) reasonable way (<= 0.10 is OK, <= 0.5 is LOW, > 0.5
> >> is HIGH).
> >>
> >> If you have a task with back pressure this means that it is producing
> >> data faster than the network can consume, for example because the
> >> downstream operator is slow or the network can't handle it. Your
> >> Source => A => B => Sink example suggests that the sink is slowing
> >> down/back pressuring B, which is in turn slowing down/back pressuring
> >> A.
> >>
> >> Does this help?
> >>
> >> Keep in mind though that it is not a rock solid approach and there is
> >> a chance that we miss the back pressure indicators or always sample
> >> when we the task is requesting buffers (which is happening all the
> >> time). It often works better at the extremes, e.g. when there is no
> >> back pressure at all or very high back pressure.
> >>
> >> – Ufuk
> >>
> >>
> >> On Tue, Apr 5, 2016 at 10:47 PM, Zach Cox  wrote:
> >> > Hi - I'm trying to identify bottlenecks in my Flink streaming job, and
> >> > am
> >> > curious about the Back Pressure view in the job manager web UI. If
> there
> >> > are
> >> > already docs for Back Pressure please feel free to just point me to
> >> > those.
> >> > :)
> >> >
> >> > When "Sampling in progress..." is displayed, what exactly is
> happening?
> >> >
> >> > What do the values in the Ratio column for each Subtask mean exactly?
> >> >
> >> > What does Status such as OK, High, etc mean? Are these determined from
> >> > the
> >> > Ratio values?
> >> >
> >> > If my job graph looks like Source => A => B => Sink, with Back
> Pressure
> >> > OK
> >> > for Source and Sink, but High for A and B, what does that suggest?
> >> >
> >> > Thanks,
> >> > Zach
> >> >
>

Re: Back Pressure details

[Ufuk Celebi]

Ah sorry, I forgot to mention that in the docs.

The way that data is pulled from Kafka is bypassing Flink's task
Thread. The topic is consumed in a separate Thread and the task Thread
is just waiting. That's why you don't see any back pressure for Kafka
sources. I would expect your Kafka source to be back pressured as well
then.

In theory it is possible that the speed at which data is consumed in
the source "matches" the speed of the back pressured operator down
stream. That would result in a non back pressured source with a down
stream back pressured task. But I don't think that's the case for your
setup. ;-)

On Wed, Apr 6, 2016 at 9:27 PM, Zach Cox  wrote:
> The new back pressure docs are great, thanks Ufuk! I'm sure those will help
> others as well.
>
> In the Source => A => B => Sink example, if A and B show HIGH back pressure,
> should Source also show HIGH? In my case it's a Kafka source and
> Elasticsearch sink. I know currently our Kafka can provide data at a much
> higher rate than our Elasticsearch can ingest (I'm working on scaling up
> Elasticsearch), just curious why the Kafka source wouldn't also show HIGH
> back pressure.
>
> Thanks,
> Zach
>
>
> On Wed, Apr 6, 2016 at 5:36 AM Ufuk Celebi  wrote:
>>
>> Hey Zach,
>>
>> just added some documentation, which will be available in ~ 30 mins
>> here:
>> https://ci.apache.org/projects/flink/flink-docs-release-1.0/internals/back_pressure_monitoring.html
>>
>> If you think that something is missing there, I would appreciate some
>> feedback. :-)
>>
>> Back pressure is determined by repeatedly calling getStackTrace() on
>> the task Threads executing the job. By default, 100 times with 50ms
>> delay between calls. If the task thread is stuck in an internal method
>> call requesting buffers from the network stack, this indicates back
>> pressure.
>>
>> The ratio you see tells you how many of the stack traces were stuck in
>> that method (e.g. 1 out of 100) and the status codes just group those
>> in a (hopefully) reasonable way (<= 0.10 is OK, <= 0.5 is LOW, > 0.5
>> is HIGH).
>>
>> If you have a task with back pressure this means that it is producing
>> data faster than the network can consume, for example because the
>> downstream operator is slow or the network can't handle it. Your
>> Source => A => B => Sink example suggests that the sink is slowing
>> down/back pressuring B, which is in turn slowing down/back pressuring
>> A.
>>
>> Does this help?
>>
>> Keep in mind though that it is not a rock solid approach and there is
>> a chance that we miss the back pressure indicators or always sample
>> when we the task is requesting buffers (which is happening all the
>> time). It often works better at the extremes, e.g. when there is no
>> back pressure at all or very high back pressure.
>>
>> – Ufuk
>>
>>
>> On Tue, Apr 5, 2016 at 10:47 PM, Zach Cox  wrote:
>> > Hi - I'm trying to identify bottlenecks in my Flink streaming job, and
>> > am
>> > curious about the Back Pressure view in the job manager web UI. If there
>> > are
>> > already docs for Back Pressure please feel free to just point me to
>> > those.
>> > :)
>> >
>> > When "Sampling in progress..." is displayed, what exactly is happening?
>> >
>> > What do the values in the Ratio column for each Subtask mean exactly?
>> >
>> > What does Status such as OK, High, etc mean? Are these determined from
>> > the
>> > Ratio values?
>> >
>> > If my job graph looks like Source => A => B => Sink, with Back Pressure
>> > OK
>> > for Source and Sink, but High for A and B, what does that suggest?
>> >
>> > Thanks,
>> > Zach
>> >

Re: Checkpoint state stored in backend, and deleting old checkpoint state

[Zach Cox]

Hi Stephan - incremental checkpointing sounds really interesting and
useful, I look forward to trying it out.

Thanks,
Zach


On Wed, Apr 6, 2016 at 4:39 AM Stephan Ewen  wrote:

> Hi Zach!
>
> I am working on incremental checkpointing, hope to have it in the master
> in the next weeks.
>
> The current approach is a to have a full self-contained checkpoint every
> once in a while, and have incremental checkpoints most of the time. Having
> a full checkpoint every now and then spares you from re-applying an endless
> set of deltas on recovery.
>
> Related to that is also making the checkpointing asynchronous, so that
> normal operations do not see any disruption any more.
>
> Greetings,
> Stephan
>
> On Tue, Apr 5, 2016 at 10:25 PM, Zach Cox  wrote:
>
>> Thanks for the details Konstantin and Ufuk!
>>
>>
>> On Tue, Apr 5, 2016 at 2:39 PM Konstantin Knauf <
>> konstantin.kn...@tngtech.com> wrote:
>>
>>> Hi Ufuk,
>>>
>>> I thought so, but I am not sure when and where ;) I will let you know,
>>> if I come across it again.
>>>
>>> Cheers,
>>>
>>> Konstantin
>>>
>>> On 05.04.2016 21:10, Ufuk Celebi wrote:
>>> > Hey Zach and Konstantin,
>>> >
>>> > Great questions and answers. We can try to make this more explicit in
>>> the docs.
>>> >
>>> > On Tue, Apr 5, 2016 at 8:54 PM, Konstantin Knauf
>>> >  wrote:
>>> >> To my knowledge flink takes care of deleting old checkpoints (I think
>>> it
>>> >> says so in the documentation about savepoints.). In my experience
>>> >> though, if a job is cancelled or crashes, the checkpoint files are
>>> >> usually not cleaned up. So some housekeeping might be necessary.
>>> >
>>> > Regarding cleanup: currently only the latest successful checkpoint is
>>> retained.
>>> >
>>> > On graceful shutdown, all checkpoints should be cleaned up as far as I
>>> > know. Savepoints always have to be cleaned up manually.
>>> >
>>> > On crashes, the checkpoint state has to be cleaned up manually (if the
>>> > JVM shut down hooks did not run).
>>> >
>>> > @Konstantin: did you have lingering state without crashes?
>>> >
>>> > – Ufuk
>>> >
>>>
>>> --
>>> Konstantin Knauf * konstantin.kn...@tngtech.com * +49-174-3413182
>>> TNG Technology Consulting GmbH, Betastr. 13a, 85774 Unterföhring
>>> Geschäftsführer: Henrik Klagges, Christoph Stock, Dr. Robert Dahlke
>>> Sitz: Unterföhring * Amtsgericht München * HRB 135082
>>>
>>
>

Re: Back Pressure details

[Zach Cox]

The new back pressure docs are great, thanks Ufuk! I'm sure those will help
others as well.

In the Source => A => B => Sink example, if A and B show HIGH back
pressure, should Source also show HIGH? In my case it's a Kafka source and
Elasticsearch sink. I know currently our Kafka can provide data at a much
higher rate than our Elasticsearch can ingest (I'm working on scaling up
Elasticsearch), just curious why the Kafka source wouldn't also show HIGH
back pressure.

Thanks,
Zach


On Wed, Apr 6, 2016 at 5:36 AM Ufuk Celebi  wrote:

> Hey Zach,
>
> just added some documentation, which will be available in ~ 30 mins
> here:
> https://ci.apache.org/projects/flink/flink-docs-release-1.0/internals/back_pressure_monitoring.html
>
> If you think that something is missing there, I would appreciate some
> feedback. :-)
>
> Back pressure is determined by repeatedly calling getStackTrace() on
> the task Threads executing the job. By default, 100 times with 50ms
> delay between calls. If the task thread is stuck in an internal method
> call requesting buffers from the network stack, this indicates back
> pressure.
>
> The ratio you see tells you how many of the stack traces were stuck in
> that method (e.g. 1 out of 100) and the status codes just group those
> in a (hopefully) reasonable way (<= 0.10 is OK, <= 0.5 is LOW, > 0.5
> is HIGH).
>
> If you have a task with back pressure this means that it is producing
> data faster than the network can consume, for example because the
> downstream operator is slow or the network can't handle it. Your
> Source => A => B => Sink example suggests that the sink is slowing
> down/back pressuring B, which is in turn slowing down/back pressuring
> A.
>
> Does this help?
>
> Keep in mind though that it is not a rock solid approach and there is
> a chance that we miss the back pressure indicators or always sample
> when we the task is requesting buffers (which is happening all the
> time). It often works better at the extremes, e.g. when there is no
> back pressure at all or very high back pressure.
>
> – Ufuk
>
>
> On Tue, Apr 5, 2016 at 10:47 PM, Zach Cox  wrote:
> > Hi - I'm trying to identify bottlenecks in my Flink streaming job, and am
> > curious about the Back Pressure view in the job manager web UI. If there
> are
> > already docs for Back Pressure please feel free to just point me to
> those.
> > :)
> >
> > When "Sampling in progress..." is displayed, what exactly is happening?
> >
> > What do the values in the Ratio column for each Subtask mean exactly?
> >
> > What does Status such as OK, High, etc mean? Are these determined from
> the
> > Ratio values?
> >
> > If my job graph looks like Source => A => B => Sink, with Back Pressure
> OK
> > for Source and Sink, but High for A and B, what does that suggest?
> >
> > Thanks,
> > Zach
> >
>

Possible use case: Simulating iterative batch processing by rewinding source

[Raul Kripalani]

Hello,

I'm getting started with Flink for a use case that could leverage the
window processing abilities of Flink that Spark does not offer.

Basically I have dumps of timeseries data (10y in ticks) which I need to
calculate many metrics in an exploratory manner based on event time. NOTE:
I don't have the metrics beforehand, it's gonna be an exploratory and
iterative data analytics effort.

Flink doesn't seem to support windows on batch processing, so I'm thinking
about emulating batch by using the Kafka stream connector and rewinding the
data stream for every new metric that I calculate, to process the full
timeseries series in a batch.

Each metric I calculate should in turn be sent to another Kafka topic so I
can use it in a subsequent processing batch, e.g.

Iteration 1)   raw timeseries data ---> metric1
Iteration 2)   raw timeseries data + metric1 (composite) ---> metric2
Iteration 3)   metric1 + metric2 ---> metric3
Iteration 4)   raw timeseries data + metric3 ---> metric4
...

Does this sound like a usecase for Flink? Could you guide me a little bit
on whether this is feasible currently?

Cheers,

*Raúl Kripalani*
PMC & Committer @ Apache Ignite, Apache Camel | Integration, Big Data and
Messaging Engineer
http://about.me/raulkripalani | http://www.linkedin.com/in/raulkripalani
Blog: raul.io
 |
twitter: @raulvk 

Re: Running Flink jobs directly from Eclipse

[Christophe Salperwyck]

I exported it in an environment variable before starting Flink:
flink-0.10.1/bin/yarn-session.sh -n 64 -s 4 -jm 4096 -tm 4096

2016-04-06 15:36 GMT+02:00 Serhiy Boychenko :

> What about YARN(and HDFS) configuration? I put yarn-site.xml directly into
> classpath? Or I can set the variables in the execution environment? I will
> give it a try tomorrow morning, will report back and if successful blog
> about it ofc J
>
>
>
> *From:* Christophe Salperwyck [mailto:christophe.salperw...@gmail.com]
> *Sent:* 06 April 2016 13:41
> *To:* user@flink.apache.org
> *Subject:* Re: Running Flink jobs directly from Eclipse
>
>
>
> For me it was taking the local jar and uploading it into the cluster.
>
>
>
> 2016-04-06 13:16 GMT+02:00 Shannon Carey :
>
> Thanks for the info! It is a bit difficult to tell based on the
> documentation whether or not you need to put your jar onto the Flink master
> node and run the flink command from there in order to get a job running.
> The documentation on
> https://ci.apache.org/projects/flink/flink-docs-release-1.0/setup/yarn_setup.html
>  isn't
> very explicit about where you can run the flink command from, and doesn't
> mention that you can run the job programmatically instead of using the CLI.
>
>
>
> *From: *Christophe Salperwyck 
> *Date: *Wednesday, April 6, 2016 at 1:24 PM
> *To: *
> *Subject: *Re: Running Flink jobs directly from Eclipse
>
>
>
> From my side I was starting the YARN session from the cluster:
>
> flink-0.10.1/bin/yarn-session.sh -n 64 -s 4 -jm 4096 -tm 4096
>
>
>
> Then getting the IP/port from the WebUI and then from Eclipse:
>
> ExecutionEnvironment env =
> ExecutionEnvironment.createRemoteEnvironment("xx.xx.xx.xx", 40631,
> "target/FlinkTest-0.0.1-SNAPSHOT-jar-with-dependencies.jar");
>
>
>
> The JAR need to be compiled before.
>
>
>
> Hope it helps!
>
> Christophe
>
>
>
> 2016-04-06 9:25 GMT+02:00 Serhiy Boychenko :
>
> Cheerz,
>
>
>
> I have been working last few month on the comparison of different data
> processing engines and recently came across Apache Flink. After reading
> different academic papers on comparison of Flink with other data processing
> I would definitely give it a shot. The only issue I am currently having is
> that I am unable to submit Flink jobs directly from Eclipse (to YARN
> cluster). I am wondering if you got any guildelines how I could do the
> submission not from the client but from Eclipse directly? (I was unable to
> find anything related, with the exception of setting up Eclipse for working
> on Flink core)
>
>
>
> Best regards,
>
> Serhiy.
>
>
>
>
>
>
>

Re: Accessing RDF triples using Flink

[Ritesh Kumar Singh]

Hi Flavio,

   1. How do you access your rdf dataset via flink? Are you reading it as a
   normal input file and splitting the records or you have some wrappers in
   place to convert the rdf data into triples? Can you please share some code
   samples if possible?
   2. I am using Jena TDB command line utilities to make queries against
   the dataset in order to avoid java garbage collection issues. I am also
   using Jena java APIs as a dependency but command line utils are way faster
   (Though it comes with an extra requirement to have Jena command line utils
   installed in the system). Main reason for this approach being able to pass
   the string output from the command line to Flink as part of my pipeline.
   Can you tell me your approach to this?
   3. Should I dump my query output to a file and then consume it as a
   normal input source for Flink?


Basically, any help regarding this will be helpful.

Regards,
Ritesh



Ritesh Kumar Singh
[image: https://]about.me/riteshoneinamillion


On Wed, Apr 6, 2016 at 2:45 PM, Flavio Pompermaier 
wrote:

> Ho Ritesh,
> I have sone experience with Rdf and Flink. What do you mean for accessing
> a Jena model? How do you create it?
>
> From my experience reading triples from jena models is evil because it has
> some problems with garbage collection.
> On 6 Apr 2016 00:51, "Ritesh Kumar Singh" 
> wrote:
>
>> Hi,
>>
>> I need some suggestions regarding accessing RDF triples from flink. I'm
>> trying to integrate flink in a pipeline where the input for flink comes
>> from SPARQL query on a Jena model. And after modification of triples using
>> flink, I will be performing SPARQL update using Jena to save my changes.
>>
>>- Are there any recommended input format for loading the triples to
>>flink?
>>- Will this use case be classified as a flink streaming job or a
>>batch processing job?
>>- How will loading of the dataset vary with the input size?
>>- Are there any recommended packages/ projects for these type of
>>projects?
>>
>> Any suggestion will be of great help.
>>
>> Regards,
>> Ritesh
>> https://riteshtoday.wordpress.com/
>>
>

Re: RemoteTransportException when trying to redis in flink code

[Till Rohrmann]

Great to hear that you solved your problem :-)

On Wed, Apr 6, 2016 at 2:29 PM, Balaji Rajagopalan <
balaji.rajagopa...@olacabs.com> wrote:

> Till,
>   Found the issue, it was my bad assumption about GlobalConfiguration,
> what I thought was once the configuration is read from the client machine
> GlobalConfiguration params will passed on to the task manager nodes, as
> well, it was not and values from default was getting pickup, which was
> localhost 6379 and there was no redis running in localhost of task manager.
>
> balaji
>
> On Wed, Apr 6, 2016 at 3:29 PM, Till Rohrmann 
> wrote:
>
>> Hmm I'm not a Redis expert, but are you sure that you see a successful
>> ping reply in the logs of the TaskManagers and not only in the client logs?
>>
>> Another thing: Is the redisClient thread safe? Multiple map tasks might
>> be accessing the set and get methods concurrently.
>>
>> Another question: The code of DriverStreamHelper you've just sent is not
>> the code you've used when receiving the stack trace, right? Because in the
>> stack trace it's written that you access a RedisClientPool from the
>> DriverStreamHelper.set method.
>>
>> Cheers,
>> Till
>>
>>
>> On Wed, Apr 6, 2016 at 11:42 AM, Balaji Rajagopalan <
>> balaji.rajagopa...@olacabs.com> wrote:
>>
>>> Till,
>>>   I have checked from all the taskmanager nodes I am able to establish a
>>> connection by installing a redis-cli on those nodes. The thing is in the
>>> constructor I am able to set and get values, also I am getting PONG for the
>>> ping. But once object is initialized when I try to call 
>>> DriverStreamHelper.get
>>> and DriverStreamHelper.set from map/apply function I get the connection
>>> refused. This may not be related to flink but rather to some security
>>> setting with Amazon AWS EMR, this is assumption now. I have also tried with
>>> 3 different redis libraries to rule out any errors with libraries the same
>>> exception in all.
>>>
>>> object DriverStreamHelper {
>>>
>>>
>>>   implicit val akkaSystem = akka.actor.ActorSystem("flink-actorsystem")
>>>
>>>   val redisClient = RedisClient(host=redisHost, port=redisPort)
>>>
>>>   val p = redisClient.ping()
>>>   p.map{ res => LOG.info(s"Reply from Redis client : $res") }
>>>
>>>
>>>
>>>   val postFix = System.currentTimeMillis()
>>>   val key = "some-key" + postFix
>>>   val value = "some-value" + postFix
>>>   set(key, value, Some(1L))
>>>   LOG.info(s"Going to get the value from Redis ${get(key)}")
>>>
>>>   def set(k: String, v: String): Unit = {
>>> redisClient.set(k,v)
>>>   }
>>>
>>>   def set(k: String, v: String, exTime: Option[Long]): Unit = {
>>>   redisClient.set(k,v,exTime)
>>>   }
>>>
>>>
>>> def get(k: String): Option[String] = {
>>> import scala.concurrent.duration._
>>> val f = redisClient.get[String](k)
>>> Await.result(f, 1.seconds) //FIXME - really bad need to return future
>>> here.
>>> }
>>>
>>> }
>>>
>>>
>>> On Wed, Apr 6, 2016 at 2:42 PM, Till Rohrmann 
>>> wrote:
>>>
 Hi Balaji,

 from the stack trace it looks as if you cannot open a connection redis.
 Have you checked that you can access redis from all your TaskManager nodes?

 Cheers,
 Till

 On Wed, Apr 6, 2016 at 7:46 AM, Balaji Rajagopalan <
 balaji.rajagopa...@olacabs.com> wrote:

> I am trying to use AWS EMR yarn cluster where the flink code runs, in
> one of apply window function, I try to set some values in redis it fails. 
> I
> have tried to access the same redis with no flink code and get/set works,
> but from the flink I get  into this exception. Any inputs on what might be
> going wrong.
>
> org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException:
> Error at remote task manager 'some-ip'.
>
> at
> org.apache.flink.runtime.io.network.netty.PartitionRequestClientHandler.decodeMsg(PartitionRequestClientHandler.java:241)
>
> at
> org.apache.flink.runtime.io.network.netty.PartitionRequestClientHandler.channelRead(PartitionRequestClientHandler.java:164)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>
> at
> io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>
> at
> io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:242)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>
> at
> 

RE: Running Flink jobs directly from Eclipse

[Serhiy Boychenko]

What about YARN(and HDFS) configuration? I put yarn-site.xml directly into 
classpath? Or I can set the variables in the execution environment? I will give 
it a try tomorrow morning, will report back and if successful blog about it ofc 
☺

From: Christophe Salperwyck [mailto:christophe.salperw...@gmail.com]
Sent: 06 April 2016 13:41
To: user@flink.apache.org
Subject: Re: Running Flink jobs directly from Eclipse

For me it was taking the local jar and uploading it into the cluster.

2016-04-06 13:16 GMT+02:00 Shannon Carey 
>:
Thanks for the info! It is a bit difficult to tell based on the documentation 
whether or not you need to put your jar onto the Flink master node and run the 
flink command from there in order to get a job running. The documentation on 
https://ci.apache.org/projects/flink/flink-docs-release-1.0/setup/yarn_setup.html
 isn't very explicit about where you can run the flink command from, and 
doesn't mention that you can run the job programmatically instead of using the 
CLI.

From: Christophe Salperwyck 
>
Date: Wednesday, April 6, 2016 at 1:24 PM
To: >
Subject: Re: Running Flink jobs directly from Eclipse

From my side I was starting the YARN session from the cluster:
flink-0.10.1/bin/yarn-session.sh -n 64 -s 4 -jm 4096 -tm 4096

Then getting the IP/port from the WebUI and then from Eclipse:
ExecutionEnvironment env = 
ExecutionEnvironment.createRemoteEnvironment("xx.xx.xx.xx", 40631, 
"target/FlinkTest-0.0.1-SNAPSHOT-jar-with-dependencies.jar");

The JAR need to be compiled before.

Hope it helps!
Christophe

2016-04-06 9:25 GMT+02:00 Serhiy Boychenko 
>:
Cheerz,

I have been working last few month on the comparison of different data 
processing engines and recently came across Apache Flink. After reading 
different academic papers on comparison of Flink with other data processing I 
would definitely give it a shot. The only issue I am currently having is that I 
am unable to submit Flink jobs directly from Eclipse (to YARN cluster). I am 
wondering if you got any guildelines how I could do the submission not from the 
client but from Eclipse directly? (I was unable to find anything related, with 
the exception of setting up Eclipse for working on Flink core)

Best regards,
Serhiy.

Re: CEP blog post

[Ufuk Celebi]

On Wed, Apr 6, 2016 at 2:18 PM, Matthias J. Sax  wrote:
> "Getting Started" in main page shows "Download 1.0" instead of 1.0.1

We always had it like that, but I agree that it can be confusing. 1.0
indicates the "series" and the download page shows the exact version.
We can certainly change it.

– Ufuk

Re: CEP blog post

[Till Rohrmann]

That is a good point Ufuk. Will add the note.

On Wed, Apr 6, 2016 at 2:03 PM, Ufuk Celebi  wrote:

> The website has been updated for 1.0.1. :-)
>
> @Till: If you don't mention it in the post, it makes sense to have a
> note at the end of the post saying that the code examples only work
> with 1.0.1.
>
> On Mon, Apr 4, 2016 at 3:35 PM, Till Rohrmann 
> wrote:
> > Thanks a lot to all for the valuable feedback. I've incorporated your
> > suggestions and will publish the article, once Flink 1.0.1 has been
> released
> > (we need 1.0.1 to run the example code).
> >
> > Cheers,
> > Till
> >
> > On Mon, Apr 4, 2016 at 10:29 AM, gen tang  wrote:
> >>
> >> It is really a good article. Please put it on Flink Blog
> >>
> >> Cheers
> >> Gen
> >>
> >>
> >> On Fri, Apr 1, 2016 at 9:56 PM, Till Rohrmann 
> >> wrote:
> >>>
> >>> Hi Flink community,
> >>>
> >>> I've written a short blog [1] post about Flink's new CEP library which
> >>> basically showcases its functionality using a monitoring example. I
> would
> >>> like to publish the post on the flink.apache.org blog next week, if
> nobody
> >>> objects. Feedback is highly appreciated :-)
> >>>
> >>> [1]
> >>>
> https://docs.google.com/document/d/1rF2zVjitdTcooIwzJKNCIvAOi85j-wDXf1goXWXHHbk/edit?usp=sharing
> >>>
> >>> Cheers,
> >>> Till
> >>
> >>
> >
>

Re: Accessing RDF triples using Flink

[Flavio Pompermaier]

Ho Ritesh,
I have sone experience with Rdf and Flink. What do you mean for accessing a
Jena model? How do you create it?

>From my experience reading triples from jena models is evil because it has
some problems with garbage collection.
On 6 Apr 2016 00:51, "Ritesh Kumar Singh" 
wrote:

> Hi,
>
> I need some suggestions regarding accessing RDF triples from flink. I'm
> trying to integrate flink in a pipeline where the input for flink comes
> from SPARQL query on a Jena model. And after modification of triples using
> flink, I will be performing SPARQL update using Jena to save my changes.
>
>- Are there any recommended input format for loading the triples to
>flink?
>- Will this use case be classified as a flink streaming job or a batch
>processing job?
>- How will loading of the dataset vary with the input size?
>- Are there any recommended packages/ projects for these type of
>projects?
>
> Any suggestion will be of great help.
>
> Regards,
> Ritesh
> https://riteshtoday.wordpress.com/
>

Re: RemoteTransportException when trying to redis in flink code

[Balaji Rajagopalan]

Till,
  Found the issue, it was my bad assumption about GlobalConfiguration, what
I thought was once the configuration is read from the client machine
GlobalConfiguration params will passed on to the task manager nodes, as
well, it was not and values from default was getting pickup, which was
localhost 6379 and there was no redis running in localhost of task manager.

balaji

On Wed, Apr 6, 2016 at 3:29 PM, Till Rohrmann  wrote:

> Hmm I'm not a Redis expert, but are you sure that you see a successful
> ping reply in the logs of the TaskManagers and not only in the client logs?
>
> Another thing: Is the redisClient thread safe? Multiple map tasks might be
> accessing the set and get methods concurrently.
>
> Another question: The code of DriverStreamHelper you've just sent is not
> the code you've used when receiving the stack trace, right? Because in the
> stack trace it's written that you access a RedisClientPool from the
> DriverStreamHelper.set method.
>
> Cheers,
> Till
>
>
> On Wed, Apr 6, 2016 at 11:42 AM, Balaji Rajagopalan <
> balaji.rajagopa...@olacabs.com> wrote:
>
>> Till,
>>   I have checked from all the taskmanager nodes I am able to establish a
>> connection by installing a redis-cli on those nodes. The thing is in the
>> constructor I am able to set and get values, also I am getting PONG for the
>> ping. But once object is initialized when I try to call 
>> DriverStreamHelper.get
>> and DriverStreamHelper.set from map/apply function I get the connection
>> refused. This may not be related to flink but rather to some security
>> setting with Amazon AWS EMR, this is assumption now. I have also tried with
>> 3 different redis libraries to rule out any errors with libraries the same
>> exception in all.
>>
>> object DriverStreamHelper {
>>
>>
>>   implicit val akkaSystem = akka.actor.ActorSystem("flink-actorsystem")
>>
>>   val redisClient = RedisClient(host=redisHost, port=redisPort)
>>
>>   val p = redisClient.ping()
>>   p.map{ res => LOG.info(s"Reply from Redis client : $res") }
>>
>>
>>
>>   val postFix = System.currentTimeMillis()
>>   val key = "some-key" + postFix
>>   val value = "some-value" + postFix
>>   set(key, value, Some(1L))
>>   LOG.info(s"Going to get the value from Redis ${get(key)}")
>>
>>   def set(k: String, v: String): Unit = {
>> redisClient.set(k,v)
>>   }
>>
>>   def set(k: String, v: String, exTime: Option[Long]): Unit = {
>>   redisClient.set(k,v,exTime)
>>   }
>>
>>
>> def get(k: String): Option[String] = {
>> import scala.concurrent.duration._
>> val f = redisClient.get[String](k)
>> Await.result(f, 1.seconds) //FIXME - really bad need to return future
>> here.
>> }
>>
>> }
>>
>>
>> On Wed, Apr 6, 2016 at 2:42 PM, Till Rohrmann 
>> wrote:
>>
>>> Hi Balaji,
>>>
>>> from the stack trace it looks as if you cannot open a connection redis.
>>> Have you checked that you can access redis from all your TaskManager nodes?
>>>
>>> Cheers,
>>> Till
>>>
>>> On Wed, Apr 6, 2016 at 7:46 AM, Balaji Rajagopalan <
>>> balaji.rajagopa...@olacabs.com> wrote:
>>>
 I am trying to use AWS EMR yarn cluster where the flink code runs, in
 one of apply window function, I try to set some values in redis it fails. I
 have tried to access the same redis with no flink code and get/set works,
 but from the flink I get  into this exception. Any inputs on what might be
 going wrong.

 org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException:
 Error at remote task manager 'some-ip'.

 at
 org.apache.flink.runtime.io.network.netty.PartitionRequestClientHandler.decodeMsg(PartitionRequestClientHandler.java:241)

 at
 org.apache.flink.runtime.io.network.netty.PartitionRequestClientHandler.channelRead(PartitionRequestClientHandler.java:164)

 at
 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)

 at
 io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)

 at
 io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)

 at
 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)

 at
 io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)

 at
 io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:242)

 at
 io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)

 at
 io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)

 at
 io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:847)

 at
 

Re: CEP blog post

[Matthias J. Sax]

"Getting Started" in main page shows "Download 1.0" instead of 1.0.1

-Matthias

On 04/06/2016 02:03 PM, Ufuk Celebi wrote:
> The website has been updated for 1.0.1. :-)
> 
> @Till: If you don't mention it in the post, it makes sense to have a
> note at the end of the post saying that the code examples only work
> with 1.0.1.
> 
> On Mon, Apr 4, 2016 at 3:35 PM, Till Rohrmann  wrote:
>> Thanks a lot to all for the valuable feedback. I've incorporated your
>> suggestions and will publish the article, once Flink 1.0.1 has been released
>> (we need 1.0.1 to run the example code).
>>
>> Cheers,
>> Till
>>
>> On Mon, Apr 4, 2016 at 10:29 AM, gen tang  wrote:
>>>
>>> It is really a good article. Please put it on Flink Blog
>>>
>>> Cheers
>>> Gen
>>>
>>>
>>> On Fri, Apr 1, 2016 at 9:56 PM, Till Rohrmann 
>>> wrote:

 Hi Flink community,

 I've written a short blog [1] post about Flink's new CEP library which
 basically showcases its functionality using a monitoring example. I would
 like to publish the post on the flink.apache.org blog next week, if nobody
 objects. Feedback is highly appreciated :-)

 [1]
 https://docs.google.com/document/d/1rF2zVjitdTcooIwzJKNCIvAOi85j-wDXf1goXWXHHbk/edit?usp=sharing

 Cheers,
 Till
>>>
>>>
>>



signature.asc
Description: OpenPGP digital signature

Re: CEP blog post

[Ufuk Celebi]

The website has been updated for 1.0.1. :-)

@Till: If you don't mention it in the post, it makes sense to have a
note at the end of the post saying that the code examples only work
with 1.0.1.

On Mon, Apr 4, 2016 at 3:35 PM, Till Rohrmann  wrote:
> Thanks a lot to all for the valuable feedback. I've incorporated your
> suggestions and will publish the article, once Flink 1.0.1 has been released
> (we need 1.0.1 to run the example code).
>
> Cheers,
> Till
>
> On Mon, Apr 4, 2016 at 10:29 AM, gen tang  wrote:
>>
>> It is really a good article. Please put it on Flink Blog
>>
>> Cheers
>> Gen
>>
>>
>> On Fri, Apr 1, 2016 at 9:56 PM, Till Rohrmann 
>> wrote:
>>>
>>> Hi Flink community,
>>>
>>> I've written a short blog [1] post about Flink's new CEP library which
>>> basically showcases its functionality using a monitoring example. I would
>>> like to publish the post on the flink.apache.org blog next week, if nobody
>>> objects. Feedback is highly appreciated :-)
>>>
>>> [1]
>>> https://docs.google.com/document/d/1rF2zVjitdTcooIwzJKNCIvAOi85j-wDXf1goXWXHHbk/edit?usp=sharing
>>>
>>> Cheers,
>>> Till
>>
>>
>

[ANNOUNCE] Flink 1.0.1 Released

[Ufuk Celebi]

The Flink PMC is pleased to announce the availability of Flink 1.0.1.

The official release announcement:
http://flink.apache.org/news/2016/04/06/release-1.0.1.html

Release binaries:
http://apache.openmirror.de/flink/flink-1.0.1/

Please update your Maven dependencies to the new 1.0.1 version and
update your binaries.

On behalf of the Flink PMC, I would like to thank everybody who
contributed to the release.

Re: Running Flink jobs directly from Eclipse

[Christophe Salperwyck]

For me it was taking the local jar and uploading it into the cluster.

2016-04-06 13:16 GMT+02:00 Shannon Carey :

> Thanks for the info! It is a bit difficult to tell based on the
> documentation whether or not you need to put your jar onto the Flink master
> node and run the flink command from there in order to get a job running.
> The documentation on
> https://ci.apache.org/projects/flink/flink-docs-release-1.0/setup/yarn_setup.html
>  isn't
> very explicit about where you can run the flink command from, and doesn't
> mention that you can run the job programmatically instead of using the CLI.
>
> From: Christophe Salperwyck 
> Date: Wednesday, April 6, 2016 at 1:24 PM
> To: 
> Subject: Re: Running Flink jobs directly from Eclipse
>
> From my side I was starting the YARN session from the cluster:
> flink-0.10.1/bin/yarn-session.sh -n 64 -s 4 -jm 4096 -tm 4096
>
> Then getting the IP/port from the WebUI and then from Eclipse:
> ExecutionEnvironment env =
> ExecutionEnvironment.createRemoteEnvironment("xx.xx.xx.xx", 40631,
> "target/FlinkTest-0.0.1-SNAPSHOT-jar-with-dependencies.jar");
>
> The JAR need to be compiled before.
>
> Hope it helps!
> Christophe
>
> 2016-04-06 9:25 GMT+02:00 Serhiy Boychenko :
>
>> Cheerz,
>>
>>
>>
>> I have been working last few month on the comparison of different data
>> processing engines and recently came across Apache Flink. After reading
>> different academic papers on comparison of Flink with other data processing
>> I would definitely give it a shot. The only issue I am currently having is
>> that I am unable to submit Flink jobs directly from Eclipse (to YARN
>> cluster). I am wondering if you got any guildelines how I could do the
>> submission not from the client but from Eclipse directly? (I was unable to
>> find anything related, with the exception of setting up Eclipse for working
>> on Flink core)
>>
>>
>>
>> Best regards,
>>
>> Serhiy.
>>
>>
>>
>
>

Re: Running Flink jobs directly from Eclipse

[Shannon Carey]

Thanks for the info! It is a bit difficult to tell based on the documentation 
whether or not you need to put your jar onto the Flink master node and run the 
flink command from there in order to get a job running. The documentation on 
https://ci.apache.org/projects/flink/flink-docs-release-1.0/setup/yarn_setup.html
 isn't very explicit about where you can run the flink command from, and 
doesn't mention that you can run the job programmatically instead of using the 
CLI.

From: Christophe Salperwyck 
>
Date: Wednesday, April 6, 2016 at 1:24 PM
To: >
Subject: Re: Running Flink jobs directly from Eclipse

From my side I was starting the YARN session from the cluster:
flink-0.10.1/bin/yarn-session.sh -n 64 -s 4 -jm 4096 -tm 4096

Then getting the IP/port from the WebUI and then from Eclipse:
ExecutionEnvironment env = 
ExecutionEnvironment.createRemoteEnvironment("xx.xx.xx.xx", 40631, 
"target/FlinkTest-0.0.1-SNAPSHOT-jar-with-dependencies.jar");

The JAR need to be compiled before.

Hope it helps!
Christophe

2016-04-06 9:25 GMT+02:00 Serhiy Boychenko 
>:
Cheerz,

I have been working last few month on the comparison of different data 
processing engines and recently came across Apache Flink. After reading 
different academic papers on comparison of Flink with other data processing I 
would definitely give it a shot. The only issue I am currently having is that I 
am unable to submit Flink jobs directly from Eclipse (to YARN cluster). I am 
wondering if you got any guildelines how I could do the submission not from the 
client but from Eclipse directly? (I was unable to find anything related, with 
the exception of setting up Eclipse for working on Flink core)

Best regards,
Serhiy.

Re: Back Pressure details

[Ufuk Celebi]

Hey Zach,

just added some documentation, which will be available in ~ 30 mins
here: 
https://ci.apache.org/projects/flink/flink-docs-release-1.0/internals/back_pressure_monitoring.html

If you think that something is missing there, I would appreciate some
feedback. :-)

Back pressure is determined by repeatedly calling getStackTrace() on
the task Threads executing the job. By default, 100 times with 50ms
delay between calls. If the task thread is stuck in an internal method
call requesting buffers from the network stack, this indicates back
pressure.

The ratio you see tells you how many of the stack traces were stuck in
that method (e.g. 1 out of 100) and the status codes just group those
in a (hopefully) reasonable way (<= 0.10 is OK, <= 0.5 is LOW, > 0.5
is HIGH).

If you have a task with back pressure this means that it is producing
data faster than the network can consume, for example because the
downstream operator is slow or the network can't handle it. Your
Source => A => B => Sink example suggests that the sink is slowing
down/back pressuring B, which is in turn slowing down/back pressuring
A.

Does this help?

Keep in mind though that it is not a rock solid approach and there is
a chance that we miss the back pressure indicators or always sample
when we the task is requesting buffers (which is happening all the
time). It often works better at the extremes, e.g. when there is no
back pressure at all or very high back pressure.

– Ufuk


On Tue, Apr 5, 2016 at 10:47 PM, Zach Cox  wrote:
> Hi - I'm trying to identify bottlenecks in my Flink streaming job, and am
> curious about the Back Pressure view in the job manager web UI. If there are
> already docs for Back Pressure please feel free to just point me to those.
> :)
>
> When "Sampling in progress..." is displayed, what exactly is happening?
>
> What do the values in the Ratio column for each Subtask mean exactly?
>
> What does Status such as OK, High, etc mean? Are these determined from the
> Ratio values?
>
> If my job graph looks like Source => A => B => Sink, with Back Pressure OK
> for Source and Sink, but High for A and B, what does that suggest?
>
> Thanks,
> Zach
>

Re: RemoteTransportException when trying to redis in flink code

[Till Rohrmann]

Hmm I'm not a Redis expert, but are you sure that you see a successful ping
reply in the logs of the TaskManagers and not only in the client logs?

Another thing: Is the redisClient thread safe? Multiple map tasks might be
accessing the set and get methods concurrently.

Another question: The code of DriverStreamHelper you've just sent is not
the code you've used when receiving the stack trace, right? Because in the
stack trace it's written that you access a RedisClientPool from the
DriverStreamHelper.set method.

Cheers,
Till


On Wed, Apr 6, 2016 at 11:42 AM, Balaji Rajagopalan <
balaji.rajagopa...@olacabs.com> wrote:

> Till,
>   I have checked from all the taskmanager nodes I am able to establish a
> connection by installing a redis-cli on those nodes. The thing is in the
> constructor I am able to set and get values, also I am getting PONG for the
> ping. But once object is initialized when I try to call DriverStreamHelper.get
> and DriverStreamHelper.set from map/apply function I get the connection
> refused. This may not be related to flink but rather to some security
> setting with Amazon AWS EMR, this is assumption now. I have also tried with
> 3 different redis libraries to rule out any errors with libraries the same
> exception in all.
>
> object DriverStreamHelper {
>
>
>   implicit val akkaSystem = akka.actor.ActorSystem("flink-actorsystem")
>
>   val redisClient = RedisClient(host=redisHost, port=redisPort)
>
>   val p = redisClient.ping()
>   p.map{ res => LOG.info(s"Reply from Redis client : $res") }
>
>
>
>   val postFix = System.currentTimeMillis()
>   val key = "some-key" + postFix
>   val value = "some-value" + postFix
>   set(key, value, Some(1L))
>   LOG.info(s"Going to get the value from Redis ${get(key)}")
>
>   def set(k: String, v: String): Unit = {
> redisClient.set(k,v)
>   }
>
>   def set(k: String, v: String, exTime: Option[Long]): Unit = {
>   redisClient.set(k,v,exTime)
>   }
>
>
> def get(k: String): Option[String] = {
> import scala.concurrent.duration._
> val f = redisClient.get[String](k)
> Await.result(f, 1.seconds) //FIXME - really bad need to return future
> here.
> }
>
> }
>
>
> On Wed, Apr 6, 2016 at 2:42 PM, Till Rohrmann 
> wrote:
>
>> Hi Balaji,
>>
>> from the stack trace it looks as if you cannot open a connection redis.
>> Have you checked that you can access redis from all your TaskManager nodes?
>>
>> Cheers,
>> Till
>>
>> On Wed, Apr 6, 2016 at 7:46 AM, Balaji Rajagopalan <
>> balaji.rajagopa...@olacabs.com> wrote:
>>
>>> I am trying to use AWS EMR yarn cluster where the flink code runs, in
>>> one of apply window function, I try to set some values in redis it fails. I
>>> have tried to access the same redis with no flink code and get/set works,
>>> but from the flink I get  into this exception. Any inputs on what might be
>>> going wrong.
>>>
>>> org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException:
>>> Error at remote task manager 'some-ip'.
>>>
>>> at
>>> org.apache.flink.runtime.io.network.netty.PartitionRequestClientHandler.decodeMsg(PartitionRequestClientHandler.java:241)
>>>
>>> at
>>> org.apache.flink.runtime.io.network.netty.PartitionRequestClientHandler.channelRead(PartitionRequestClientHandler.java:164)
>>>
>>> at
>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>>
>>> at
>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>>
>>> at
>>> io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
>>>
>>> at
>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>>
>>> at
>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>>
>>> at
>>> io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:242)
>>>
>>> at
>>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>>
>>> at
>>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>>
>>> at
>>> io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:847)
>>>
>>> at
>>> io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
>>>
>>> at
>>> io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
>>>
>>> at
>>> io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
>>>
>>> at
>>> io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
>>>
>>> at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
>>>
>>> at
>>> io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
>>>
>>> at java.lang.Thread.run(Thread.java:745)
>>>
>>> Caused by:
>>> 

Re: Flink CEP AbstractCEPPatternOperator fail after event detection

[Till Rohrmann]

Hi Norman,

which version of Flink are you using? We recently fixed some issues with
the CEP library which looked similar to your error message. The problem
occurred when using the CEP library with processing time. Switching to
event or ingestion time, solve the problem.

The fixes to make it also work with processing time are included in the
latest snapshot version 1.1-SNAPSHOT and will be part of the upcoming 1.0.1
bugfix release. The bugfix release will actually be released today.

If the problem should still remain with the latest version, it would be
good to see your complete Flink program.

Cheers,
Till

On Wed, Apr 6, 2016 at 11:04 AM, norman sp  wrote:

> Hi,
> I'm trying out the new CEP library but have some problems with event
> detection.
> In my case Flink detects the event pattern: A followed by B within 10
> seconds.
> But short time after event detection when the event pattern isn't matched
> anymore, the program crashes with the error message:
>
> 04/06/2016 11:04:47 Job execution switched to status FAILING.
> java.lang.NullPointerException
> at
>
> org.apache.flink.cep.nfa.SharedBuffer.extractPatterns(SharedBuffer.java:205)
> at org.apache.flink.cep.nfa.NFA.extractPatternMatches(NFA.java:305)
> at org.apache.flink.cep.nfa.NFA.process(NFA.java:142)
> at
>
> org.apache.flink.cep.operator.AbstractCEPPatternOperator.processEvent(AbstractCEPPatternOperator.java:93)
> at
>
> org.apache.flink.cep.operator.CEPPatternOperator.processWatermark(CEPPatternOperator.java:88)
> at
>
> org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:158)
> at
>
> org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:65)
> at
>
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:224)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)
> at java.lang.Thread.run(Thread.java:745)
>
>
> After that, the job execution is restarted and proceeds well until the next
> AbstractCEPPatternOperator failes.
>
> That's my code:
> Pattern, ?> FlowPattern =
> Pattern.>begin("start")
> .followedBy("FlowOver10")
> .where(new FilterFunction Double>>()
> {//some Filter}})
> .followedBy("PressureOver10")
> .where(new FilterFunction Double>>()
> {//some Filter}})
> .within(Time.seconds(10));
>
> PatternStream>
> FlowFirstPatternStream = CEP.pattern(windowedData, FlowFirstPattern);
> DataStream warning = FlowFirstPatternStream.select(new
> FlowPatternWarning());
> warning.print();
>
> private static class FlowPatternWarning implements
> PatternSelectFunction,
> String> {
> @Override
> public String select(Map Double, Double,
> Double>> pat) throws Exception {
>   Tuple5
> pressure =
> pat.get("PressureOver10");
>   Tuple5 flow =
> pat.get("FlowOver10");
>
> return "  ###   Warning! FlowPattern   ###
> " +
> pressure.toString() + " - " + flow.toString();
> }
> }
>
>
> How can I solve that?
> Hope somebody could help me.
>
> greetz Norman
>
>
>
>
> --
> View this message in context:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-CEP-AbstractCEPPatternOperator-fail-after-event-detection-tp5948.html
> Sent from the Apache Flink User Mailing List archive. mailing list archive
> at Nabble.com.
>

Re: RemoteTransportException when trying to redis in flink code

[Balaji Rajagopalan]

Till,
  I have checked from all the taskmanager nodes I am able to establish a
connection by installing a redis-cli on those nodes. The thing is in the
constructor I am able to set and get values, also I am getting PONG for the
ping. But once object is initialized when I try to call DriverStreamHelper.get
and DriverStreamHelper.set from map/apply function I get the connection
refused. This may not be related to flink but rather to some security
setting with Amazon AWS EMR, this is assumption now. I have also tried with
3 different redis libraries to rule out any errors with libraries the same
exception in all.

object DriverStreamHelper {


  implicit val akkaSystem = akka.actor.ActorSystem("flink-actorsystem")

  val redisClient = RedisClient(host=redisHost, port=redisPort)

  val p = redisClient.ping()
  p.map{ res => LOG.info(s"Reply from Redis client : $res") }



  val postFix = System.currentTimeMillis()
  val key = "some-key" + postFix
  val value = "some-value" + postFix
  set(key, value, Some(1L))
  LOG.info(s"Going to get the value from Redis ${get(key)}")

  def set(k: String, v: String): Unit = {
redisClient.set(k,v)
  }

  def set(k: String, v: String, exTime: Option[Long]): Unit = {
  redisClient.set(k,v,exTime)
  }


def get(k: String): Option[String] = {
import scala.concurrent.duration._
val f = redisClient.get[String](k)
Await.result(f, 1.seconds) //FIXME - really bad need to return future here.
}

}


On Wed, Apr 6, 2016 at 2:42 PM, Till Rohrmann  wrote:

> Hi Balaji,
>
> from the stack trace it looks as if you cannot open a connection redis.
> Have you checked that you can access redis from all your TaskManager nodes?
>
> Cheers,
> Till
>
> On Wed, Apr 6, 2016 at 7:46 AM, Balaji Rajagopalan <
> balaji.rajagopa...@olacabs.com> wrote:
>
>> I am trying to use AWS EMR yarn cluster where the flink code runs, in one
>> of apply window function, I try to set some values in redis it fails. I
>> have tried to access the same redis with no flink code and get/set works,
>> but from the flink I get  into this exception. Any inputs on what might be
>> going wrong.
>>
>> org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException:
>> Error at remote task manager 'some-ip'.
>>
>> at
>> org.apache.flink.runtime.io.network.netty.PartitionRequestClientHandler.decodeMsg(PartitionRequestClientHandler.java:241)
>>
>> at
>> org.apache.flink.runtime.io.network.netty.PartitionRequestClientHandler.channelRead(PartitionRequestClientHandler.java:164)
>>
>> at
>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>
>> at
>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>
>> at
>> io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
>>
>> at
>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>
>> at
>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>
>> at
>> io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:242)
>>
>> at
>> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>>
>> at
>> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>>
>> at
>> io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:847)
>>
>> at
>> io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
>>
>> at
>> io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
>>
>> at
>> io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
>>
>> at
>> io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
>>
>> at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
>>
>> at
>> io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
>>
>> at java.lang.Thread.run(Thread.java:745)
>>
>> Caused by:
>> org.apache.flink.runtime.io.network.partition.ProducerFailedException
>>
>> at
>> org.apache.flink.runtime.io.network.netty.PartitionRequestQueue.writeAndFlushNextMessageIfPossible(PartitionRequestQueue.java:164)
>>
>> at
>> org.apache.flink.runtime.io.network.netty.PartitionRequestQueue.userEventTriggered(PartitionRequestQueue.java:96)
>>
>> at
>> io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:308)
>>
>> at
>> io.netty.channel.AbstractChannelHandlerContext.fireUserEventTriggered(AbstractChannelHandlerContext.java:294)
>>
>> at
>> io.netty.channel.ChannelInboundHandlerAdapter.userEventTriggered(ChannelInboundHandlerAdapter.java:108)
>>
>> at
>> io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:308)
>>
>> at

Re: Checkpoint state stored in backend, and deleting old checkpoint state

[Stephan Ewen]

Hi Zach!

I am working on incremental checkpointing, hope to have it in the master in
the next weeks.

The current approach is a to have a full self-contained checkpoint every
once in a while, and have incremental checkpoints most of the time. Having
a full checkpoint every now and then spares you from re-applying an endless
set of deltas on recovery.

Related to that is also making the checkpointing asynchronous, so that
normal operations do not see any disruption any more.

Greetings,
Stephan

On Tue, Apr 5, 2016 at 10:25 PM, Zach Cox  wrote:

> Thanks for the details Konstantin and Ufuk!
>
>
> On Tue, Apr 5, 2016 at 2:39 PM Konstantin Knauf <
> konstantin.kn...@tngtech.com> wrote:
>
>> Hi Ufuk,
>>
>> I thought so, but I am not sure when and where ;) I will let you know,
>> if I come across it again.
>>
>> Cheers,
>>
>> Konstantin
>>
>> On 05.04.2016 21:10, Ufuk Celebi wrote:
>> > Hey Zach and Konstantin,
>> >
>> > Great questions and answers. We can try to make this more explicit in
>> the docs.
>> >
>> > On Tue, Apr 5, 2016 at 8:54 PM, Konstantin Knauf
>> >  wrote:
>> >> To my knowledge flink takes care of deleting old checkpoints (I think
>> it
>> >> says so in the documentation about savepoints.). In my experience
>> >> though, if a job is cancelled or crashes, the checkpoint files are
>> >> usually not cleaned up. So some housekeeping might be necessary.
>> >
>> > Regarding cleanup: currently only the latest successful checkpoint is
>> retained.
>> >
>> > On graceful shutdown, all checkpoints should be cleaned up as far as I
>> > know. Savepoints always have to be cleaned up manually.
>> >
>> > On crashes, the checkpoint state has to be cleaned up manually (if the
>> > JVM shut down hooks did not run).
>> >
>> > @Konstantin: did you have lingering state without crashes?
>> >
>> > – Ufuk
>> >
>>
>> --
>> Konstantin Knauf * konstantin.kn...@tngtech.com * +49-174-3413182
>> TNG Technology Consulting GmbH, Betastr. 13a, 85774 Unterföhring
>> Geschäftsführer: Henrik Klagges, Christoph Stock, Dr. Robert Dahlke
>> Sitz: Unterföhring * Amtsgericht München * HRB 135082
>>
>

Flink CEP AbstractCEPPatternOperator fail after event detection

[norman sp]

Hi,
I'm trying out the new CEP library but have some problems with event
detection.
In my case Flink detects the event pattern: A followed by B within 10
seconds.
But short time after event detection when the event pattern isn't matched
anymore, the program crashes with the error message:

04/06/2016 11:04:47 Job execution switched to status FAILING.
java.lang.NullPointerException
at
org.apache.flink.cep.nfa.SharedBuffer.extractPatterns(SharedBuffer.java:205)
at org.apache.flink.cep.nfa.NFA.extractPatternMatches(NFA.java:305)
at org.apache.flink.cep.nfa.NFA.process(NFA.java:142)
at
org.apache.flink.cep.operator.AbstractCEPPatternOperator.processEvent(AbstractCEPPatternOperator.java:93)
at
org.apache.flink.cep.operator.CEPPatternOperator.processWatermark(CEPPatternOperator.java:88)
at
org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:158)
at
org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:65)
at
org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:224)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)
at java.lang.Thread.run(Thread.java:745)


After that, the job execution is restarted and proceeds well until the next
AbstractCEPPatternOperator failes.

That's my code:
Pattern, ?> FlowPattern =
Pattern.>begin("start")
.followedBy("FlowOver10")
.where(new FilterFunction>()
{//some Filter}})
.followedBy("PressureOver10")
.where(new FilterFunction>()
{//some Filter}})
.within(Time.seconds(10));

PatternStream>
FlowFirstPatternStream = CEP.pattern(windowedData, FlowFirstPattern);
DataStream warning = FlowFirstPatternStream.select(new
FlowPatternWarning());
warning.print();

private static class FlowPatternWarning implements
PatternSelectFunction,
String> {
@Override
public String select(Map> pat) throws Exception {
  Tuple5 pressure =
pat.get("PressureOver10");
  Tuple5 flow =
pat.get("FlowOver10");

return "  ###   Warning! FlowPattern   ### " +
pressure.toString() + " - " + flow.toString();  
}
}


How can I solve that?
Hope somebody could help me.

greetz Norman




--
View this message in context: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-CEP-AbstractCEPPatternOperator-fail-after-event-detection-tp5948.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at 
Nabble.com.

Re: RemoteTransportException when trying to redis in flink code

[Till Rohrmann]

Hi Balaji,

from the stack trace it looks as if you cannot open a connection redis.
Have you checked that you can access redis from all your TaskManager nodes?

Cheers,
Till

On Wed, Apr 6, 2016 at 7:46 AM, Balaji Rajagopalan <
balaji.rajagopa...@olacabs.com> wrote:

> I am trying to use AWS EMR yarn cluster where the flink code runs, in one
> of apply window function, I try to set some values in redis it fails. I
> have tried to access the same redis with no flink code and get/set works,
> but from the flink I get  into this exception. Any inputs on what might be
> going wrong.
>
> org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException:
> Error at remote task manager 'some-ip'.
>
> at
> org.apache.flink.runtime.io.network.netty.PartitionRequestClientHandler.decodeMsg(PartitionRequestClientHandler.java:241)
>
> at
> org.apache.flink.runtime.io.network.netty.PartitionRequestClientHandler.channelRead(PartitionRequestClientHandler.java:164)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>
> at
> io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>
> at
> io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:242)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:339)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:324)
>
> at
> io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:847)
>
> at
> io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
>
> at
> io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
>
> at
> io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
>
> at
> io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
>
> at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
>
> at
> io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
>
> at java.lang.Thread.run(Thread.java:745)
>
> Caused by:
> org.apache.flink.runtime.io.network.partition.ProducerFailedException
>
> at
> org.apache.flink.runtime.io.network.netty.PartitionRequestQueue.writeAndFlushNextMessageIfPossible(PartitionRequestQueue.java:164)
>
> at
> org.apache.flink.runtime.io.network.netty.PartitionRequestQueue.userEventTriggered(PartitionRequestQueue.java:96)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:308)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.fireUserEventTriggered(AbstractChannelHandlerContext.java:294)
>
> at
> io.netty.channel.ChannelInboundHandlerAdapter.userEventTriggered(ChannelInboundHandlerAdapter.java:108)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:308)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.fireUserEventTriggered(AbstractChannelHandlerContext.java:294)
>
> at
> io.netty.channel.ChannelInboundHandlerAdapter.userEventTriggered(ChannelInboundHandlerAdapter.java:108)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:308)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.fireUserEventTriggered(AbstractChannelHandlerContext.java:294)
>
> at
> io.netty.channel.ChannelInboundHandlerAdapter.userEventTriggered(ChannelInboundHandlerAdapter.java:108)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.invokeUserEventTriggered(AbstractChannelHandlerContext.java:308)
>
> at
> io.netty.channel.AbstractChannelHandlerContext.access$500(AbstractChannelHandlerContext.java:32)
>
> at
> io.netty.channel.AbstractChannelHandlerContext$6.run(AbstractChannelHandlerContext.java:299)
>
> at
> io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:357)
>
> at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:357)
>
> ... 2 more
>
>
> Caused by: java.lang.RuntimeException: java.net.ConnectException:
> Connection refused
>
> at com.redis.IO $class.connect(IO.scala:37)
>
> at com.redis.RedisClient.connect(RedisClient.scala:94)
>
> at com.redis.RedisCommand$class.initialize(RedisClient.scala:71)
>
> at com.redis.RedisClient.initialize(RedisClient.scala:94)
>
> at com.redis.RedisClient.(RedisClient.scala:98)
>
> at com.redis.RedisClientFactory.makeObject(Pool.scala:12)
>
> at 

Re: Integrate Flink with S3 on EMR cluster

[Ufuk Celebi]

Yes, for sure.

I added some documentation for AWS here:
https://ci.apache.org/projects/flink/flink-docs-release-1.0/setup/aws.html

Would be nice to update that page with your pull request. :-)

– Ufuk


On Wed, Apr 6, 2016 at 4:58 AM, Chiwan Park  wrote:
> Hi Timur,
>
> Great! Bootstrap action for Flink is good for AWS users. I think the 
> bootstrap action scripts would be placed in `flink-contrib` directory.
>
> If you want, one of people in PMC of Flink will be assign FLINK-1337 to you.
>
> Regards,
> Chiwan Park
>
>> On Apr 6, 2016, at 3:36 AM, Timur Fayruzov  wrote:
>>
>> I had a guide like that.
>>
>

Re: Running Flink jobs directly from Eclipse

[Christophe Salperwyck]

>From my side I was starting the YARN session from the cluster:
flink-0.10.1/bin/yarn-session.sh -n 64 -s 4 -jm 4096 -tm 4096

Then getting the IP/port from the WebUI and then from Eclipse:
ExecutionEnvironment env =
ExecutionEnvironment.createRemoteEnvironment("xx.xx.xx.xx", 40631,
"target/FlinkTest-0.0.1-SNAPSHOT-jar-with-dependencies.jar");

The JAR need to be compiled before.

Hope it helps!
Christophe

2016-04-06 9:25 GMT+02:00 Serhiy Boychenko :

> Cheerz,
>
>
>
> I have been working last few month on the comparison of different data
> processing engines and recently came across Apache Flink. After reading
> different academic papers on comparison of Flink with other data processing
> I would definitely give it a shot. The only issue I am currently having is
> that I am unable to submit Flink jobs directly from Eclipse (to YARN
> cluster). I am wondering if you got any guildelines how I could do the
> submission not from the client but from Eclipse directly? (I was unable to
> find anything related, with the exception of setting up Eclipse for working
> on Flink core)
>
>
>
> Best regards,
>
> Serhiy.
>
>
>

Running Flink jobs directly from Eclipse

[Serhiy Boychenko]

Cheerz,

I have been working last few month on the comparison of different data 
processing engines and recently came across Apache Flink. After reading 
different academic papers on comparison of Flink with other data processing I 
would definitely give it a shot. The only issue I am currently having is that I 
am unable to submit Flink jobs directly from Eclipse (to YARN cluster). I am 
wondering if you got any guildelines how I could do the submission not from the 
client but from Eclipse directly? (I was unable to find anything related, with 
the exception of setting up Eclipse for working on Flink core)

Best regards,
Serhiy.

Re: Powered by Flink

[Suneel Marthi]

I was gonna hold off on that until we get Mahout 0.12.0 out of the door
(targeted for this weekend).

I would add Apache NiFi to the list.

Future :

Apache Mahout
Apache BigTop

Openstack and Kubernetes (skunkworks)


On Wed, Apr 6, 2016 at 3:03 AM, Sebastian  wrote:

> You should also add Apache Mahout, whose new Samsara DSL also runs on
> Flink.
>
> -s
>
> On 06.04.2016 08:50, Henry Saputra wrote:
>
>> Thanks, Slim. I have just updated the wiki page with this entries.
>>
>> On Tue, Apr 5, 2016 at 10:20 AM, Slim Baltagi > > wrote:
>>
>> Hi
>>
>> The following are missing in the ‘Powered by Flink’ list:
>>
>>   * *king.com 
>> *
>> https://blogs.apache.org/foundation/entry/the_apache_software_foundation_announces88
>>   * *Otto Group
>> *
>> http://data-artisans.com/how-we-selected-apache-flink-at-otto-group/
>>   * *Eura Nova *https://research.euranova.eu/flink-forward-2015-talk/
>>   * *Big Data Europe *http://www.big-data-europe.eu
>>
>> Thanks
>>
>> Slim Baltagi
>>
>>
>> On Apr 5, 2016, at 10:08 AM, Robert Metzger >> > wrote:
>>>
>>> Hi everyone,
>>>
>>> I would like to bring the "Powered by Flink" wiki page [1] to the
>>> attention of Flink user's who recently joined the Flink community.
>>> The list tracks which organizations are using Flink.
>>> If your company / university / research institute / ... is using
>>> Flink but the name is not yet listed there, let me know and I'll
>>> add the name.
>>>
>>> Regards,
>>> Robert
>>>
>>> [1]
>>> https://cwiki.apache.org/confluence/display/FLINK/Powered+by+Flink
>>>
>>>
>>> On Mon, Oct 19, 2015 at 4:10 PM, Matthias J. Sax >> > wrote:
>>>
>>> +1
>>>
>>> On 10/19/2015 04:05 PM, Maximilian Michels wrote:
>>> > +1 Let's collect in the Wiki for now. At some point in time,
>>> we might
>>> > want to have a dedicated page on the Flink homepage.
>>> >
>>> > On Mon, Oct 19, 2015 at 3:31 PM, Timo Walther
>>> > wrote:
>>> >> Ah ok, sorry. I think linking to the wiki is also ok.
>>> >>
>>> >>
>>> >> On 19.10.2015 15:18, Fabian Hueske wrote:
>>> >>>
>>> >>> @Timo: The proposal was to keep the list in the wiki (can
>>> be easily
>>> >>> extended) but link from the main website to the wiki page.
>>> >>>
>>> >>> 2015-10-19 15:16 GMT+02:00 Timo Walther
>>> >:
>>> >>>
>>>  +1 for adding it to the website instead of wiki.
>>>  "Who is using Flink?" is always a question difficult to
>>> answer to
>>>  interested users.
>>> 
>>> 
>>>  On 19.10.2015 15:08, Suneel Marthi wrote:
>>> 
>>>  +1 to this.
>>> 
>>>  On Mon, Oct 19, 2015 at 3:00 PM, Fabian Hueske
>>> > wrote:
>>> 
>>> > Sounds good +1
>>> >
>>> > 2015-10-19 14:57 GMT+02:00 Márton Balassi <
>>> >
>>> > balassi.mar...@gmail.com >> >>:
>>> >
>>> >> Thanks for starting and big +1 for making it more
>>> prominent.
>>> >>
>>> >> On Mon, Oct 19, 2015 at 2:53 PM, Fabian Hueske <
>>> >
>>> >
>>> > fhue...@gmail.com > wrote:
>>> >>>
>>> >>> Thanks for starting this Kostas.
>>> >>>
>>> >>> I think the list is quite hidden in the wiki. Should
>>> we link from
>>> >>> flink.apache.org  to that
>>> page?
>>> >>>
>>> >>> Cheers, Fabian
>>> >>>
>>> >>> 2015-10-19 14:50 GMT+02:00 Kostas Tzoumas <
>>> >
>>> >
>>> > ktzou...@apache.org >:
>>> 
>>>  Hi everyone,
>>> 
>>>  I started a "Powered by Flink" wiki page, listing
>>> some of the
>>>  organizations that are using Flink:
>>> 
>>> 
>>>
>>> https://cwiki.apache.org/confluence/display/FLINK/Powered+by+Flink
>>> 
>>>  If you would like to be added to the list, just send
>>> me a short email
>>>  with your organization's name and a description and I
>>>   

Re: Powered by Flink

[Sebastian]

You should also add Apache Mahout, whose new Samsara DSL also runs on Flink.

-s

On 06.04.2016 08:50, Henry Saputra wrote:

Thanks, Slim. I have just updated the wiki page with this entries.

On Tue, Apr 5, 2016 at 10:20 AM, Slim Baltagi > wrote:

Hi

The following are missing in the ‘Powered by Flink’ list:

  * *king.com 

*https://blogs.apache.org/foundation/entry/the_apache_software_foundation_announces88
  * *Otto Group
*http://data-artisans.com/how-we-selected-apache-flink-at-otto-group/
  * *Eura Nova *https://research.euranova.eu/flink-forward-2015-talk/
  * *Big Data Europe *http://www.big-data-europe.eu

Thanks

Slim Baltagi



On Apr 5, 2016, at 10:08 AM, Robert Metzger > wrote:

Hi everyone,

I would like to bring the "Powered by Flink" wiki page [1] to the
attention of Flink user's who recently joined the Flink community.
The list tracks which organizations are using Flink.
If your company / university / research institute / ... is using
Flink but the name is not yet listed there, let me know and I'll
add the name.

Regards,
Robert

[1] https://cwiki.apache.org/confluence/display/FLINK/Powered+by+Flink


On Mon, Oct 19, 2015 at 4:10 PM, Matthias J. Sax > wrote:

+1

On 10/19/2015 04:05 PM, Maximilian Michels wrote:
> +1 Let's collect in the Wiki for now. At some point in time,
we might
> want to have a dedicated page on the Flink homepage.
>
> On Mon, Oct 19, 2015 at 3:31 PM, Timo Walther
> wrote:
>> Ah ok, sorry. I think linking to the wiki is also ok.
>>
>>
>> On 19.10.2015 15:18, Fabian Hueske wrote:
>>>
>>> @Timo: The proposal was to keep the list in the wiki (can
be easily
>>> extended) but link from the main website to the wiki page.
>>>
>>> 2015-10-19 15:16 GMT+02:00 Timo Walther
>:
>>>
 +1 for adding it to the website instead of wiki.
 "Who is using Flink?" is always a question difficult to
answer to
 interested users.


 On 19.10.2015 15:08, Suneel Marthi wrote:

 +1 to this.

 On Mon, Oct 19, 2015 at 3:00 PM, Fabian Hueske
> wrote:

> Sounds good +1
>
> 2015-10-19 14:57 GMT+02:00 Márton Balassi <
>
> balassi.mar...@gmail.com >:
>
>> Thanks for starting and big +1 for making it more
prominent.
>>
>> On Mon, Oct 19, 2015 at 2:53 PM, Fabian Hueske <
>
>
> fhue...@gmail.com > wrote:
>>>
>>> Thanks for starting this Kostas.
>>>
>>> I think the list is quite hidden in the wiki. Should
we link from
>>> flink.apache.org  to that page?
>>>
>>> Cheers, Fabian
>>>
>>> 2015-10-19 14:50 GMT+02:00 Kostas Tzoumas <
>
>
> ktzou...@apache.org >:

 Hi everyone,

 I started a "Powered by Flink" wiki page, listing
some of the
 organizations that are using Flink:


https://cwiki.apache.org/confluence/display/FLINK/Powered+by+Flink

 If you would like to be added to the list, just send
me a short email
 with your organization's name and a description and I
will add you to
>
> the

 wiki page.

 Best,
 Kostas

>>>


>>

Re: Powered by Flink

[Henry Saputra]

Thanks, Slim. I have just updated the wiki page with this entries.

On Tue, Apr 5, 2016 at 10:20 AM, Slim Baltagi  wrote:

> Hi
>
> The following are missing in the ‘Powered by Flink’ list:
>
>- *king.com  *
>
> https://blogs.apache.org/foundation/entry/the_apache_software_foundation_announces88
>- *Otto Group  *
>http://data-artisans.com/how-we-selected-apache-flink-at-otto-group/
>- *Eura Nova *https://research.euranova.eu/flink-forward-2015-talk/
>- *Big Data Europe *http://www.big-data-europe.eu
>
> Thanks
>
> Slim Baltagi
>
>
> On Apr 5, 2016, at 10:08 AM, Robert Metzger  wrote:
>
> Hi everyone,
>
> I would like to bring the "Powered by Flink" wiki page [1] to the
> attention of Flink user's who recently joined the Flink community. The list
> tracks which organizations are using Flink.
> If your company / university / research institute / ... is using Flink but
> the name is not yet listed there, let me know and I'll add the name.
>
> Regards,
> Robert
>
> [1] https://cwiki.apache.org/confluence/display/FLINK/Powered+by+Flink
>
>
> On Mon, Oct 19, 2015 at 4:10 PM, Matthias J. Sax  wrote:
>
>> +1
>>
>> On 10/19/2015 04:05 PM, Maximilian Michels wrote:
>> > +1 Let's collect in the Wiki for now. At some point in time, we might
>> > want to have a dedicated page on the Flink homepage.
>> >
>> > On Mon, Oct 19, 2015 at 3:31 PM, Timo Walther 
>> wrote:
>> >> Ah ok, sorry. I think linking to the wiki is also ok.
>> >>
>> >>
>> >> On 19.10.2015 15:18, Fabian Hueske wrote:
>> >>>
>> >>> @Timo: The proposal was to keep the list in the wiki (can be easily
>> >>> extended) but link from the main website to the wiki page.
>> >>>
>> >>> 2015-10-19 15:16 GMT+02:00 Timo Walther :
>> >>>
>>  +1 for adding it to the website instead of wiki.
>>  "Who is using Flink?" is always a question difficult to answer to
>>  interested users.
>> 
>> 
>>  On 19.10.2015 15:08, Suneel Marthi wrote:
>> 
>>  +1 to this.
>> 
>>  On Mon, Oct 19, 2015 at 3:00 PM, Fabian Hueske 
>> wrote:
>> 
>> > Sounds good +1
>> >
>> > 2015-10-19 14:57 GMT+02:00 Márton Balassi < <
>> balassi.mar...@gmail.com>
>> > balassi.mar...@gmail.com>:
>> >
>> >> Thanks for starting and big +1 for making it more prominent.
>> >>
>> >> On Mon, Oct 19, 2015 at 2:53 PM, Fabian Hueske < <
>> fhue...@gmail.com>
>> >
>> > fhue...@gmail.com> wrote:
>> >>>
>> >>> Thanks for starting this Kostas.
>> >>>
>> >>> I think the list is quite hidden in the wiki. Should we link from
>> >>> flink.apache.org to that page?
>> >>>
>> >>> Cheers, Fabian
>> >>>
>> >>> 2015-10-19 14:50 GMT+02:00 Kostas Tzoumas < 
>> >
>> > ktzou...@apache.org>:
>> 
>>  Hi everyone,
>> 
>>  I started a "Powered by Flink" wiki page, listing some of the
>>  organizations that are using Flink:
>> 
>> 
>> https://cwiki.apache.org/confluence/display/FLINK/Powered+by+Flink
>> 
>>  If you would like to be added to the list, just send me a short
>> email
>>  with your organization's name and a description and I will add
>> you to
>> >
>> > the
>> 
>>  wiki page.
>> 
>>  Best,
>>  Kostas
>> 
>> >>>
>> 
>> 
>> >>
>>
>>
>
>

Re: State in external db (dynamodb)

[Sanne de Roever]

FYI Cassandra has a TTL on data:
https://docs.datastax.com/en/cql/3.1/cql/cql_using/use_expire_t.html

On Wed, Apr 6, 2016 at 7:55 AM, Shannon Carey  wrote:

> Hi, new Flink user here!
>
> I found a discussion on user@flink.apache.org about using DynamoDB as a
> sink. However, as noted, sinks have an at-least-once guarantee so your
> operations must idempotent.
>
> However, another way to go about this (and correct me if I'm wrong) is to
> write the state to the external store via a custom State Backend. Since the
> state participates in checkpointing, you don't have to worry about
> idempotency: every time state is checkpointed, overwrite the value of that
> key.
>
> We are starting a project with Flink, and we are interested in evicting
> the state from memory once a TTL is reached during which no events have
> come in for that state. Subsequently, when an event is processed, we must
> be able to quickly load up any evicted state. Does this sound reasonable?
> We are considering using DynamoDB for our state backend because it seems
> like all we will need is a key-value store. The only weakness of this is
> that if state gets older than, say, 2 years we would like to get rid of it
> which might not be easy in DynamoDB. I don't suppose Flink has any
> behind-the-scenes features that deal with getting rid of old state (either
> evicting from memory or TTL/aging out entirely)?
>
> Thanks for your time!
> Shannon Carey
>