Thanks a lot Gordon!
On 29 June 2017 at 13:39, Tzu-Li (Gordon) Tai <tzuli...@apache.org> wrote:
> Sorry, one typo.
>
> public AverageAccumulator merge(WindowStats a, WindowStats b) {
>
> should be:
>
> public WindowStats merge(WindowStats a, WindowStats b) {
>
> On 29 June 2017 at 8:22:34 PM, Tzu-Li (Gordon) Tai (tzuli...@apache.org)
> wrote:
>
> I see. Then yes, a fold operation would be more efficient here.
>
>
> However, can you give an idea on how to use aggregateFunction in latest
> flink to replace the following fold function?
>
> Sure! The documentation for 1.3 is still a bit lagging behind for some of
> the new features, but the Javadoc for `AggregateFunction` should be rather
> self-explaining.
>
> As a quick sketch, here’s what you would do to achieve the same thing:
>
> public class WindowStatsAggregator implements AggregateFunction<IN,
> WindowStats, OUT> {
>
> public WindowStats createAccumulator() {
> return new WindowStats();
> }
>
> public AverageAccumulator merge(WindowStats a, WindowStats b) {
> // merge the two unique products map in your WindowStats
> }
>
> public void add(IN value, WindowStats acc) {
> // update your unique products map
> }
>
> public OUT getResult(WindowStats acc) {
> return acc.getMap();
> }
> }
>
>
> As you can see, the `AggregateFunction` is more generic, and should
> subsume whatever you were previously doing with fold.
> Your previous `WindowStats` class is basically the state accumulator, and
> you need to implement how to update it, merge two accumulators, and
> retrieve the final accumulated result.
>
> For more info, I would point to the class Javadocs of `AggregateFunction`.
>
> Best,
> Gordon
>
> On 29 June 2017 at 8:06:25 PM, Ahmad Hassan (ahmad.has...@gmail.com)
> wrote:
>
> Hi Gordon,
>
> Thanks for the details. I am using fold to process events and maintain
> statistics per each product ID within WindowStats instance. So fold is much
> efficient because events can be in millions but unique products will be
> less than 50k. However, if i use generic window function, It will be less
> efficient because window function will receive a collection of millions of
> events and they will be replicated for each sliding window as Flink
> replicate events for each sliding window.
>
> However, can you give an idea on how to use aggregateFunction in latest
> flink to replace the following fold function?
>
> final DataStream<WindowStats> eventStream = inputStream
> .keyBy(TENANT, CATEGORY)
> .window(SlidingProcessingTimeWindows.of(Time.hour(1,Time.minute(5)))
> *.fold(new WindowStats(),* newProductAggregationMapper(),
> newProductAggregationWindowFunction());
>
> Thanks!
>
> On 29 June 2017 at 12:57, Tzu-Li (Gordon) Tai <tzuli...@apache.org> wrote:
>
>> Hi Ahmad,
>>
>> Yes, that is correct. The aggregated fold value (i.e. your WindowStats
>> instance) will be checkpointed by Flink as managed state, and restored from
>> the last complete checkpoint in case of failures.
>> One comment on using the fold function: if what you’re essentially doing
>> in the fold is just collecting the elements of the windows per key and
>> performing the actual aggregation in the window function, then you don't
>> need the fold.
>> A generic window function should suit that case. See [1].
>>
>> Cheers,
>> Gordon
>>
>> [1] https://ci.apache.org/projects/flink/flink-docs-release-
>> 1.3/dev/windows.html#windowfunction---the-generic-case
>>
>>
>> On 29 June 2017 at 5:11:58 PM, Ahmad Hassan (ahmad.has...@gmail.com)
>> wrote:
>>
>> Any thoughts on this problem please?
>>
>>
>> Hi All,
>>
>> I am collecting millions of events per 24hour for 'N' number of products
>> where 'N' can be 50k. I use the following fold mechanism with sliding
>> window:
>>
>> final DataStream<WindowStats> eventStream = inputStream
>> .keyBy(TENANT, CATEGORY)
>> .window(SlidingProcessingTimeWindows.of(Time.hour(24,Time.minute(5)))
>> *.fold(new WindowStats(),* newProductAggregationMapper(),
>> newProductAggregationWindowFunction());
>>
>> In WindowStats class, I keep a map of HashMap<String productID,
>> ProductMetric ProductMetric>> which keeps products event count and other
>> various metrics. So for 50k products I will have 50k entries in the map
>> within WindowStats instance instead of millions of Events as fold
>> function will process them as the event arrives.
>>
>> My question is, if I set (env.enableCheckpointing(1000)), then the
>> WindowStats
>> instance for each existing window will automatically be checkpointed and
>> restored on recovery? If not then how can I better a implement above
>> usecase to store the state of WindowStats object within fold operation
>> please?
>>
>> Thanks for all the help.
>>
>> Best Regards,
>>
>>
>
