Re: What is the best way to load/add patterns dynamically (at runtime) with Flink?

[kaelumania]

Hey,

the javascript solution seems very limited. Is there a solution with
compiling new patterns to native Flink CEP Patterns and add them to a stream
dynamically?

best Stephan



--
View this message in context: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/What-is-the-best-way-to-load-add-patterns-dynamically-at-runtime-with-Flink-tp9461p10325.html
Sent from the Apache Flink User Mailing List archive. mailing list archive at 
Nabble.com.

Re: Maintaining watermarks per key, instead of per operator instance

[kaelumania]

Sounds good to me. But I still need to have some kind of side output 
(cassandra) that stores the accumulating aggregates on each time scale (minute, 
hour). Thus I would need to have something like this

var hourly = stream.window(1.hour).apply(..)
//write to cassandra
hourly.trigger(accumulating).addSink(cassandra)
//forward to next acc step
var daily = hourly.trigger(discarding).window(1.day).apply(…)
//write to cassandra
daily.trigger(accumulating).addSink(cassandra)

Would this be possible?

best, Stephan
> On 23 Nov 2016, at 11:16, Aljoscha Krettek [via Apache Flink User Mailing 
> List archive.]  wrote:
> 
> You can implement discarding behaviour by writing a custom trigger (based on 
> EventTimeTrigger) that returns FIRE_AND_PURGE when firing. With this you 
> could maybe implement a cascade of windows where the first aggregates for the 
> smallest time interval and is discarding and where the other triggers take 
> these "pre-aggregated" values and accumulate.
> 
> On Tue, 22 Nov 2016 at 08:11 Stephan Epping <[hidden email] 
> > wrote:
> Hey Aljoscha,
> 
> the first solution did not work out as expected. As when late elements arrive 
> the first window is triggered again and would emit a new (accumulated) event, 
> that would be counted twice (in time accumulation and late accumulation) in 
> the second window.I could implement my own (discarding strategy) like in 
> Apache Beam, but the out stream should contain accumulated events that are 
> stored in cassandra. The second solution just gave an compiler error, thus I 
> think is not possible right now.
> 
> best Stephan
> 
> 
> 
>> On 21 Nov 2016, at 17:56, Aljoscha Krettek <[hidden email] 
>> > wrote:
>> 
>> Hi,
>> why did you settle for the last solution?
>> 
>> Cheers,
>> Aljoscha
>> 
>> On Thu, 17 Nov 2016 at 15:57 kaelumania <[hidden email] 
>> > wrote:
>> Hi Fabian,
>> 
>> your proposed solution for:
>>  
>> Multiple window aggregations
>> You can construct a data flow of cascading window operators and fork off (to 
>> emit or further processing) the result after each window.
>> 
>> Input -> window(15 secs) -> window(1 min) -> window(15 min) -> ...
>> \-> out_1\-> out_2 \-> out_3
>> does not work, am I missing something?
>> 
>> First I tried the following
>> DataStream values = input.assignTimestampsAndWatermarks(new 
>> StrictWatermarkAssigner()); // force lateness
>> 
>> DataStream aggregatesPerMinute = values
>> .keyBy("id")
>> .timeWindow(Time.minutes(1))
>> .allowedLateness(Time.minutes(2))
>> .apply(new ReadingAggregate(), new AggregateReadings(), new 
>> AggregateReadings());
>> 
>> DataStream aggregatesPerHour = aggregatesPerMinute
>> .keyBy("id")
>> .timeWindow(Time.hours(1))
>> .allowedLateness(Time.hours(2))
>> .apply(new AggregateReadingAggregates(), new 
>> AggregateReadingAggregates());
>> but due to late data the first fold function would emit 2 rolling aggregates 
>> (one with and one without the late element), which results in being counted 
>> twice within the second reducer. Therefore i tried
>> WindowedStream readingsPerMinute = input
>> .assignTimestampsAndWatermarks(new StrictWatermarkAssigner()) // 
>> force lateness
>> .keyBy("id")
>> .timeWindow(Time.minutes(1))
>> .allowedLateness(Time.hours(2));
>> 
>> WindowedStream readingsPerHours = 
>> readingsPerMinute
>> .timeWindow(Time.hours(1))
>> .allowedLateness(Time.hours(2));
>> 
>> DataStream aggregatesPerMinute = 
>> readingsPerMinute.apply(new ReadingAggregate(), new AggregateReadings(), new 
>> AggregateReadings());
>> DataStream aggregatesPerHour = readingsPerHours.apply(new 
>> ReadingAggregate(), new AggregateReadings(), new AggregateReadings());
>> which gives me a compiler error as WindowedStream does not provide a 
>> timeWindow method.
>> 
>> Finally I settled with this:
>> KeyedStream readings = input
>> .assignTimestampsAndWatermarks(new StrictWatermarkAssigner()) // 
>> force lateness
>> .keyBy("id");
>> 
>> DataStream aggregatesPerMinute = readings
>> .timeWindow(Time.minutes(1))
>> .allowedLateness(Time.hours(2))
>> .apply(new ReadingAggregate(), new AggregateReadings(), new 
>> AggregateReadings());
>> 
>> DataStream aggregatesPerHour = readings
>&

Re: Maintaining watermarks per key, instead of per operator instance

[kaelumania]

Hi Fabian,

your proposed solution for:
 
Multiple window aggregations
You can construct a data flow of cascading window operators and fork off (to 
emit or further processing) the result after each window.

Input -> window(15 secs) -> window(1 min) -> window(15 min) -> ...
\-> out_1\-> out_2 \-> out_3
does not work, am I missing something?

First I tried the following
DataStream values = input.assignTimestampsAndWatermarks(new 
StrictWatermarkAssigner()); // force lateness

DataStream aggregatesPerMinute = values
.keyBy("id")
.timeWindow(Time.minutes(1))
.allowedLateness(Time.minutes(2))
.apply(new ReadingAggregate(), new AggregateReadings(), new 
AggregateReadings());

DataStream aggregatesPerHour = aggregatesPerMinute
.keyBy("id")
.timeWindow(Time.hours(1))
.allowedLateness(Time.hours(2))
.apply(new AggregateReadingAggregates(), new 
AggregateReadingAggregates());
but due to late data the first fold function would emit 2 rolling aggregates 
(one with and one without the late element), which results in being counted 
twice within the second reducer. Therefore i tried
WindowedStream readingsPerMinute = input
.assignTimestampsAndWatermarks(new StrictWatermarkAssigner()) // force 
lateness
.keyBy("id")
.timeWindow(Time.minutes(1))
.allowedLateness(Time.hours(2));

WindowedStream readingsPerHours = readingsPerMinute
.timeWindow(Time.hours(1))
.allowedLateness(Time.hours(2));

DataStream aggregatesPerMinute = readingsPerMinute.apply(new 
ReadingAggregate(), new AggregateReadings(), new AggregateReadings());
DataStream aggregatesPerHour = readingsPerHours.apply(new 
ReadingAggregate(), new AggregateReadings(), new AggregateReadings());
which gives me a compiler error as WindowedStream does not provide a timeWindow 
method.

Finally I settled with this:
KeyedStream readings = input
.assignTimestampsAndWatermarks(new StrictWatermarkAssigner()) // force 
lateness
.keyBy("id");

DataStream aggregatesPerMinute = readings
.timeWindow(Time.minutes(1))
.allowedLateness(Time.hours(2))
.apply(new ReadingAggregate(), new AggregateReadings(), new 
AggregateReadings());

DataStream aggregatesPerHour = readings
.timeWindow(Time.hours(1))
.allowedLateness(Time.hours(2))
.apply(new ReadingAggregate(), new AggregateReadings(), new 
AggregateReadings());


Feedback is very welcome.

best, Stephan



> On 11 Nov 2016, at 00:29, Fabian Hueske-2 [via Apache Flink User Mailing List 
> archive.]  wrote:
> 
> Hi Stephan,
> 
> I just wrote an answer to your SO question. 
> 
> Best, Fabian
> 
> 2016-11-10 11:01 GMT+01:00 Stephan Epping <[hidden email] 
> >:
> Hello,
> 
> I found this question in the Nabble archive 
> (http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Maintaining-watermarks-per-key-instead-of-per-operator-instance-tp7288.html
>  
> )
>  but was unable/dont know how to reply.
> 
> Here is my question regarding the mentioned thread:
> 
>> Hello, 
>> 
>> I have similar requirements (see StackOverflor 
>> http://stackoverflow.com/questions/40465335/apache-flink-multiple-window-aggregations-and-late-data
>>  
>> ).
>>  I am pretty new to flink, could you elaborate on a possible solution? We 
>> can guarantee good ordering by sensor_id, thus watermarking by key would be 
>> the only reasonable way for us 
>> (sensorData.keyBy('id').timeWindow(1.minute).sum('value')), could I do my 
>> own watermarking aftersensorData.keyBy('id').overwriteWatermarking()... per 
>> key? Or maybe using custom state plus a custom trigger? What happens if a 
>> sensor dies or is being removed completely, how can this be detected as 
>> watermarks would be ignored for window garbage collection. Or could we 
>> dynamically schedule a job of each sensor? Which would result in 1000 Jobs.
> 
> 
> Thanks,
> Stephan
> 
> 
> 
> 
> 
> If you reply to this email, your message will be added to the discussion 
> below:
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Maintaining-watermarks-per-key-instead-of-per-operator-instance-tp7288p10033.html
>  
> 
> To unsubscribe from Maintaining watermarks per key, instead of per operator 
> instance, click here 
> .
> NAML 
> 

Re: Maintaining watermarks per key, instead of per operator instance

[kaelumania]

Hey Fabian,

thank you very much. 

- yes, I would window by event time and fire/purge by processing time
- Cheaper in the end meant, that having too much state in the flink cluster 
would be more expensive, as we store all data in cassandra too.I think the 
fault tolerance would be okay, as we would make a compare and set with 
cassandra. 

With the flatMap Operator wouldn’t it be like running my own windowing 
mechanism? I need to keep the aggregate window per sensor open (with 
checkpointing and state management) until I receive an element for a sensor 
that is later in time than the windows time and then purge the state and emit a 
new event (which is like having a watermark per sensor). Further, I need a 
timer that fires like after 24 hours, in case a sensor dies and doesn’t send 
more data which might is possible with window assigner/trigger, right? But not 
inside normal functions, e.g. flatMap? We can guarantee that all sensor data 
per sensor comes almost in order (might be out of order within a few seconds), 
but there might be gaps of several hours after network partitions.

There is now way to define/redefine the watermark per keyed stream? Or adjust 
the window assigner + trigger to achieve the desired behaviour? I am a bit 
reserved in implementing the whole state management. Do you plan to support 
such use cases on keyed streams? Maybe the WatermarkAssigner could also receive 
information about the key for wich the watermark should be calculated etc.

best, Stephan


> On 14 Nov 2016, at 15:17, Fabian Hueske-2 [via Apache Flink User Mailing List 
> archive.]  wrote:
> 
> Hi Stephan,
> 
> I'm skeptical about two things: 
> - using processing time will result in inaccurately bounded aggregates (or do 
> you want to group by event time in a processing time window?)
> - writing to and reading from Cassandra might be expensive (not sure what you 
> mean by cheaper in the end) and it is not integrated with Flink's 
> checkpointing mechanism for fault-tolerance.
> 
> To me, the stateful FlatMapOperator looks like the best approach. There is an 
> upcoming feature for registering timers in user-functions, i.e., a function 
> is called after the timer exceeds. This could be helpful to overcome the 
> problem of closing the window without new data.
> 
> Best, 
> Fabian
> 
> 2016-11-14 8:39 GMT+01:00 Stephan Epping <[hidden email] 
> >:
> Hello Fabian,
> 
> Thank you very much. What is your opinion on the following solution:
> 
> - Window data per time window, e.g. 15 minutes
> - using processing time as trigger, e.g. 15 minutes
> - which results in an aggregate over sensor values
> - then use cassandra to select the previous aggregate (as there can be 
> multiple for the time window due to processing time)
> - then update the aggregate and put it into a cassandra sink again
> 
> The cassandra select will be a bit slower than using an in memory/flink 
> state, but will be cheaper in the end. Further, what does this have for 
> consequences?
> For example, replaying events will be more difficult, right? Also, what about 
> Snapshots? Will they work with the mentioned design?
> 
> kind regards,
> Stephan
> 
> 
>> On 11 Nov 2016, at 00:39, Fabian Hueske <[hidden email] 
>> > wrote:
>> 
>> Hi Stephan,
>> 
>> I just wrote an answer to your SO question. 
>> 
>> Best, Fabian
>> 
>> 2016-11-10 11:01 GMT+01:00 Stephan Epping <[hidden email] 
>> >:
>> Hello,
>> 
>> I found this question in the Nabble archive 
>> (http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Maintaining-watermarks-per-key-instead-of-per-operator-instance-tp7288.html
>>  
>> )
>>  but was unable/dont know how to reply.
>> 
>> Here is my question regarding the mentioned thread:
>> 
>>> Hello, 
>>> 
>>> I have similar requirements (see StackOverflor 
>>> http://stackoverflow.com/questions/40465335/apache-flink-multiple-window-aggregations-and-late-data
>>>  
>>> ).
>>>  I am pretty new to flink, could you elaborate on a possible solution? We 
>>> can guarantee good ordering by sensor_id, thus watermarking by key would be 
>>> the only reasonable way for us 
>>> (sensorData.keyBy('id').timeWindow(1.minute).sum('value')), could I do my 
>>> own watermarking aftersensorData.keyBy('id').overwriteWatermarking()... per 
>>> key? Or maybe using custom state plus a custom trigger? What happens if a 
>>> sensor dies or is being removed completely, how can this be detected as 
>>> watermarks would be ignored for window garbage collection. Or could we 
>>> dynamically schedule a job of each sensor? Which would result in 1000 Jobs.
>> 
>> 
>> Thanks,
>> Stephan
>> 
>> 
>> 
> 
> 
> 
> 
> If you reply to this email, your message will be added to the discussion 
> below:
> http://apache-flink-user-