You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Shankar Mane <ss...@gmail.com> on 2022/05/16 16:40:22 UTC
Help - for single input, getting multiple same output over each slide's
Hi,
Even when a user doesn't perform actions in subsequent windows, the same
output is produced for subsequent sliding intervals. How do I limit it to
not produce it multiple times?
*Code Snippet :*
DataStream<Tuple2<MyAverage,Double>> addcashCountStream = processedStream
> .filter(Objects::*nonNull*)
>
> .map(*new* MyMapper())
>
> .keyBy( x -> x.userid)
>
> .window(SlidingProcessingTimeWindows.*of*(Time.*seconds*(120), Time.
> *seconds*(10)))
>
> .aggregate(*new* AverageAggregatorFunc())
>
Please find the attached image as an example.
[image: flink-single-input-multiple-same-output.png]
Regards,
Shankar
Re:Re: Re: Help - for single input, getting multiple same output over each slide's
Posted by Xuyang <xy...@163.com>.
Hi, what about using UDAF[1] to define the logic what your want?
[1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/functions/udfs/#aggregate-functions
--
Best！
Xuyang
在 2022-05-18 16:39:25，"Shankar Mane" <ss...@gmail.com> 写道：
Can someone answer my questions in trail mail ?
On Tue, May 17, 2022 at 1:02 PM Shankar Mane <ss...@gmail.com> wrote:
Hi, please find my comments below :
1. Change the sliding window to other types of windows.
Our use case is to use sliding window. (for e.g. at any point, whenever user performs any actions at time [ t1 ], we would like to see his activity in [ t1 - last 24 hours]. Using this, to show users some recommendations.)
2. Deduplicate the deplicated rows after this operator using UDF and so on
Have some questions. please check below points..
one row data may exist in multiple sliding windows
Don't we think that we are doing the same deduplication (using UDF) again and again until the record is out of the sliding window ?. Why I am saying is that we would have millions of users in 24 hrs of window, as window slides we are doing this redundant work.
On Tue, May 17, 2022 at 12:12 PM Xuyang <xy...@163.com> wrote:
This is because one row data may exist in multiple sliding windows, and that is the feature of Sliding Window [1]. I think the following two ways may meet your requirement that you expect.
1. Change the sliding window to other types of windows.
2. Deduplicate the deplicated rows after this operator using UDF and so on.
[1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/operators/windows/#sliding-windows
--
Best！
Xuyang
在 2022-05-17 13:10:14，"Shankar Mane" <ss...@gmail.com> 写道：
Please help me to understand this..
On Mon, 16 May 2022, 22:10 Shankar Mane, <ss...@gmail.com> wrote:
Hi,
Even when a user doesn't perform actions in subsequent windows, the same output is produced for subsequent sliding intervals. How do I limit it to not produce it multiple times?
Code Snippet :
DataStream<Tuple2<MyAverage,Double>> addcashCountStream = processedStream.filter(Objects::nonNull)
.map(new MyMapper())
.keyBy( x -> x.userid)
.window(SlidingProcessingTimeWindows.of(Time.seconds(120), Time.seconds(10)))
.aggregate(new AverageAggregatorFunc())
Please find the attached image as an example.
Regards,
Shankar
Re: Re: Help - for single input, getting multiple same output over each slide's
Posted by Shankar Mane <ss...@gmail.com>.
Can someone answer my questions in trail mail ?
On Tue, May 17, 2022 at 1:02 PM Shankar Mane <ss...@gmail.com> wrote:
> Hi, please find my comments below :
>
> 1. Change the sliding window to other types of windows.
>
>> Our use case is to use sliding window. (for e.g. at any point, whenever
>> user performs any actions at time [ t1 ], we would like to see his activity
>> in [ t1 - last 24 hours]. Using this, to show users some recommendations.)
>
>
>
> 2. Deduplicate the deplicated rows after this operator using UDF and so on
>
>> Have some questions. please check below points..
>
>
> one row data may exist in multiple sliding windows
>
>> Don't we think that we are doing the same deduplication (using UDF) again
>> and again until the record is out of the sliding window ?.
>
> Why I am saying is that we would have millions of users in 24 hrs of
>> window, as window slides we are doing this redundant work.
>>
>
> On Tue, May 17, 2022 at 12:12 PM Xuyang <xy...@163.com> wrote:
>
>> This is because one row data may exist in multiple sliding windows, and
>> that is the feature of Sliding Window [1]. I think the following two ways
>> may meet your requirement that you expect.
>> 1. Change the sliding window to other types of windows.
>> 2. Deduplicate the deplicated rows after this operator using UDF and so
>> on.
>>
>>
>> [1]
>> https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/operators/windows/#sliding-windows
>>
>>
>> --
>> Best！
>> Xuyang
>>
>>
>> 在 2022-05-17 13:10:14，"Shankar Mane" <ss...@gmail.com> 写道：
>>
>> Please help me to understand this..
>>
>> On Mon, 16 May 2022, 22:10 Shankar Mane, <ss...@gmail.com> wrote:
>>
>>> Hi,
>>>
>>> Even when a user doesn't perform actions in subsequent windows, the same
>>> output is produced for subsequent sliding intervals. How do I limit it to
>>> not produce it multiple times?
>>>
>>> *Code Snippet :*
>>>
>>> DataStream<Tuple2<MyAverage,Double>> addcashCountStream =
>>>> processedStream.filter(Objects::*nonNull*)
>>>>
>>>> .map(*new* MyMapper())
>>>>
>>>> .keyBy( x -> x.userid)
>>>>
>>>> .window(SlidingProcessingTimeWindows.*of*(Time.*seconds*(120), Time.
>>>> *seconds*(10)))
>>>>
>>>> .aggregate(*new* AverageAggregatorFunc())
>>>>
>>>
>>> Please find the attached image as an example.
>>> [image: flink-single-input-multiple-same-output.png]
>>>
>>> Regards,
>>> Shankar
>>>
>>
Re: Re: Help - for single input, getting multiple same output over each slide's
Posted by Shankar Mane <ss...@gmail.com>.
Hi, please find my comments below :
1. Change the sliding window to other types of windows.
> Our use case is to use sliding window. (for e.g. at any point, whenever
> user performs any actions at time [ t1 ], we would like to see his activity
> in [ t1 - last 24 hours]. Using this, to show users some recommendations.)
2. Deduplicate the deplicated rows after this operator using UDF and so on
> Have some questions. please check below points..
one row data may exist in multiple sliding windows
> Don't we think that we are doing the same deduplication (using UDF) again
> and again until the record is out of the sliding window ?.
Why I am saying is that we would have millions of users in 24 hrs of
> window, as window slides we are doing this redundant work.
>
On Tue, May 17, 2022 at 12:12 PM Xuyang <xy...@163.com> wrote:
> This is because one row data may exist in multiple sliding windows, and
> that is the feature of Sliding Window [1]. I think the following two ways
> may meet your requirement that you expect.
> 1. Change the sliding window to other types of windows.
> 2. Deduplicate the deplicated rows after this operator using UDF and so on.
>
>
> [1]
> https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/operators/windows/#sliding-windows
>
>
> --
> Best！
> Xuyang
>
>
> 在 2022-05-17 13:10:14，"Shankar Mane" <ss...@gmail.com> 写道：
>
> Please help me to understand this..
>
> On Mon, 16 May 2022, 22:10 Shankar Mane, <ss...@gmail.com> wrote:
>
>> Hi,
>>
>> Even when a user doesn't perform actions in subsequent windows, the same
>> output is produced for subsequent sliding intervals. How do I limit it to
>> not produce it multiple times?
>>
>> *Code Snippet :*
>>
>> DataStream<Tuple2<MyAverage,Double>> addcashCountStream = processedStream
>>> .filter(Objects::*nonNull*)
>>>
>>> .map(*new* MyMapper())
>>>
>>> .keyBy( x -> x.userid)
>>>
>>> .window(SlidingProcessingTimeWindows.*of*(Time.*seconds*(120), Time.
>>> *seconds*(10)))
>>>
>>> .aggregate(*new* AverageAggregatorFunc())
>>>
>>
>> Please find the attached image as an example.
>> [image: flink-single-input-multiple-same-output.png]
>>
>> Regards,
>> Shankar
>>
>
Re:Re: Help - for single input, getting multiple same output over each slide's
Posted by Xuyang <xy...@163.com>.
This is because one row data may exist in multiple sliding windows, and that is the feature of Sliding Window [1]. I think the following two ways may meet your requirement that you expect.
1. Change the sliding window to other types of windows.
2. Deduplicate the deplicated rows after this operator using UDF and so on.
[1] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/operators/windows/#sliding-windows
--
Best！
Xuyang
在 2022-05-17 13:10:14，"Shankar Mane" <ss...@gmail.com> 写道：
Please help me to understand this..
On Mon, 16 May 2022, 22:10 Shankar Mane, <ss...@gmail.com> wrote:
Hi,
Even when a user doesn't perform actions in subsequent windows, the same output is produced for subsequent sliding intervals. How do I limit it to not produce it multiple times?
Code Snippet :
DataStream<Tuple2<MyAverage,Double>> addcashCountStream = processedStream.filter(Objects::nonNull)
.map(new MyMapper())
.keyBy( x -> x.userid)
.window(SlidingProcessingTimeWindows.of(Time.seconds(120), Time.seconds(10)))
.aggregate(new AverageAggregatorFunc())
Please find the attached image as an example.
Regards,
Shankar
Re: Help - for single input, getting multiple same output over each slide's
Posted by Shankar Mane <ss...@gmail.com>.
Please help me to understand this..
On Mon, 16 May 2022, 22:10 Shankar Mane, <ss...@gmail.com> wrote:
> Hi,
>
> Even when a user doesn't perform actions in subsequent windows, the same
> output is produced for subsequent sliding intervals. How do I limit it to
> not produce it multiple times?
>
> *Code Snippet :*
>
> DataStream<Tuple2<MyAverage,Double>> addcashCountStream = processedStream
>> .filter(Objects::*nonNull*)
>>
>> .map(*new* MyMapper())
>>
>> .keyBy( x -> x.userid)
>>
>> .window(SlidingProcessingTimeWindows.*of*(Time.*seconds*(120), Time.
>> *seconds*(10)))
>>
>> .aggregate(*new* AverageAggregatorFunc())
>>
>
> Please find the attached image as an example.
> [image: flink-single-input-multiple-same-output.png]
>
> Regards,
> Shankar
>