Question: How to avoid local execution being terminated before session window closes

[Klemens Muthmann]

Hi,

I have written an Apache Flink Pipeline containing the following piece
of code (Java):

stream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(50))).aggregate(new
CustomAggregator()).print();

If I run the pipeline using local execution I see the following
behavior. The "CustomAggregator" calls the `createAccumulator` and `add`
methods correctly with the correct data. However it never calls
`getResult` and my pipeline simply finishes.

So I did a little research and found out that it works if I change the
code to:

stream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(1))).aggregate(new
CustomAggregator()).print();

Notice the reduced gap time for the processing time session window. So
it seems that execution only continues if the window has been closed and
if that takes too long, the execution simply aborts. I guess another
factor playing a part in the problem is, that my initial data is read in
much faster than 50 seconds. This results in the pipeline being in a
state where it only waits for the window to be closed and having nothing
else to do it decides that there is no work left and simply shuts down.

My question now is if it is possible to tell the local execution
environment to wait for that window to be closed, instead of just
shutting down.

Thanks and Regards

     Klemens Muthmann

[Timo Walther]

Hi Klemens,

what you are observing are reasons why event-time should be preferred
over processing-time. Event-time uses the timestamp of your data while
processing-time is to basic for many use cases. Esp. when you want to
reprocess historic data, you want to do that at full speed instead of
waiting 1 hour for 1-hour-windows.

If you want to use processing-time nevertheless, you need to use a
source that produced unbounded streams instead of bounded streams such
that the pipeline execution theoretically is infinite. Some
documentation can be found here [1] where you need to use the
`FileProcessingMode.PROCESS_CONTINUOUSLY`. But what kind of connector
are you currently using?

Regards,
Timo

[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/datastream_api.html#data-sources

On 24.11.20 09:59, Klemens Muthmann wrote:

> Hi,
>
> I have written an Apache Flink Pipeline containing the following piece
> of code (Java):
>
> stream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(50))).aggregate(new
> CustomAggregator()).print();
>
> If I run the pipeline using local execution I see the following
> behavior. The "CustomAggregator" calls the `createAccumulator` and `add`
> methods correctly with the correct data. However it never calls
> `getResult` and my pipeline simply finishes.
>
> So I did a little research and found out that it works if I change the
> code to:
>
> stream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(1))).aggregate(new
> CustomAggregator()).print();
>
> Notice the reduced gap time for the processing time session window. So
> it seems that execution only continues if the window has been closed and
> if that takes too long, the execution simply aborts. I guess another
> factor playing a part in the problem is, that my initial data is read in
> much faster than 50 seconds. This results in the pipeline being in a
> state where it only waits for the window to be closed and having nothing
> else to do it decides that there is no work left and simply shuts down.
>
> My question now is if it is possible to tell the local execution
> environment to wait for that window to be closed, instead of just
> shutting down.
>
> Thanks and Regards
>
>      Klemens Muthmann
>

[Klemens Muthmann]

Hi,

Thanks for your reply. I am using processing time instead of event time,
since we do get the events in batches and some might arrive days later.

But for my current dev setup I just use a CSV dump of finite size as
input. I will hand over the pipeline to some other guys, who will need
to integrate it with an Apache Kafka Service. Output is written to a
Postgres-Database-System.

I'll have a look at your proposal and let you know if it worked, after
having finished a few prerequisite parts.

Regards

     Klemens

Am 24.11.20 um 12:59 schrieb Timo Walther:

> Hi Klemens,
>
> what you are observing are reasons why event-time should be preferred
> over processing-time. Event-time uses the timestamp of your data while
> processing-time is to basic for many use cases. Esp. when you want to
> reprocess historic data, you want to do that at full speed instead of
> waiting 1 hour for 1-hour-windows.
>
> If you want to use processing-time nevertheless, you need to use a
> source that produced unbounded streams instead of bounded streams such
> that the pipeline execution theoretically is infinite. Some
> documentation can be found here [1] where you need to use the
> `FileProcessingMode.PROCESS_CONTINUOUSLY`. But what kind of connector
> are you currently using?
>
> Regards,
> Timo
>
> [1]
> https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/datastream_api.html#data-sources
>
> On 24.11.20 09:59, Klemens Muthmann wrote:
>> Hi,
>>
>> I have written an Apache Flink Pipeline containing the following
>> piece of code (Java):
>>
>> stream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(50))).aggregate(new
>> CustomAggregator()).print();
>>
>> If I run the pipeline using local execution I see the following
>> behavior. The "CustomAggregator" calls the `createAccumulator` and
>> `add` methods correctly with the correct data. However it never calls
>> `getResult` and my pipeline simply finishes.
>>
>> So I did a little research and found out that it works if I change
>> the code to:
>>
>> stream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(1))).aggregate(new
>> CustomAggregator()).print();
>>
>> Notice the reduced gap time for the processing time session window.
>> So it seems that execution only continues if the window has been
>> closed and if that takes too long, the execution simply aborts. I
>> guess another factor playing a part in the problem is, that my
>> initial data is read in much faster than 50 seconds. This results in
>> the pipeline being in a state where it only waits for the window to
>> be closed and having nothing else to do it decides that there is no
>> work left and simply shuts down.
>>
>> My question now is if it is possible to tell the local execution
>> environment to wait for that window to be closed, instead of just
>> shutting down.
>>
>> Thanks and Regards
>>
>>      Klemens Muthmann
>>
>

--
Mit freundlichen Grüßen
           Dr.-Ing. Klemens Muthmann

-----------------------------------
Cyface GmbH
Hertha-Lindner-Straße 10
01067 Dresden
web: www.cyface.de
email: [hidden email]

[Timo Walther]

For debugging you can also implement a simple non-parallel source using
`org.apache.flink.streaming.api.functions.source.SourceFunction`. You
would need to implement the run() method with an endless loop after
emitting all your records.

Regards,
Timo

On 24.11.20 16:07, Klemens Muthmann wrote:

> Hi,
>
> Thanks for your reply. I am using processing time instead of event time,
> since we do get the events in batches and some might arrive days later.
>
> But for my current dev setup I just use a CSV dump of finite size as
> input. I will hand over the pipeline to some other guys, who will need
> to integrate it with an Apache Kafka Service. Output is written to a
> Postgres-Database-System.
>
> I'll have a look at your proposal and let you know if it worked, after
> having finished a few prerequisite parts.
>
> Regards
>
>      Klemens
>
> Am 24.11.20 um 12:59 schrieb Timo Walther:
>> Hi Klemens,
>>
>> what you are observing are reasons why event-time should be preferred
>> over processing-time. Event-time uses the timestamp of your data while
>> processing-time is to basic for many use cases. Esp. when you want to
>> reprocess historic data, you want to do that at full speed instead of
>> waiting 1 hour for 1-hour-windows.
>>
>> If you want to use processing-time nevertheless, you need to use a
>> source that produced unbounded streams instead of bounded streams such
>> that the pipeline execution theoretically is infinite. Some
>> documentation can be found here [1] where you need to use the
>> `FileProcessingMode.PROCESS_CONTINUOUSLY`. But what kind of connector
>> are you currently using?
>>
>> Regards,
>> Timo
>>
>> [1]
>> https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/datastream_api.html#data-sources 
>>
>>
>> On 24.11.20 09:59, Klemens Muthmann wrote:
>>> Hi,
>>>
>>> I have written an Apache Flink Pipeline containing the following
>>> piece of code (Java):
>>>
>>> stream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(50))).aggregate(new
>>> CustomAggregator()).print();
>>>
>>> If I run the pipeline using local execution I see the following
>>> behavior. The "CustomAggregator" calls the `createAccumulator` and
>>> `add` methods correctly with the correct data. However it never calls
>>> `getResult` and my pipeline simply finishes.
>>>
>>> So I did a little research and found out that it works if I change
>>> the code to:
>>>
>>> stream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(1))).aggregate(new
>>> CustomAggregator()).print();
>>>
>>> Notice the reduced gap time for the processing time session window.
>>> So it seems that execution only continues if the window has been
>>> closed and if that takes too long, the execution simply aborts. I
>>> guess another factor playing a part in the problem is, that my
>>> initial data is read in much faster than 50 seconds. This results in
>>> the pipeline being in a state where it only waits for the window to
>>> be closed and having nothing else to do it decides that there is no
>>> work left and simply shuts down.
>>>
>>> My question now is if it is possible to tell the local execution
>>> environment to wait for that window to be closed, instead of just
>>> shutting down.
>>>
>>> Thanks and Regards
>>>
>>>      Klemens Muthmann
>>>
>>

[Klemens Muthmann]

Hi,

Thanks for the hint. The infinite loop was the solution and my pipeline
works now.

Regards

     Klemens

Am 24.11.20 um 16:59 schrieb Timo Walther:

> For debugging you can also implement a simple non-parallel source
> using
> `org.apache.flink.streaming.api.functions.source.SourceFunction`. You
> would need to implement the run() method with an endless loop after
> emitting all your records.
>
> Regards,
> Timo
>
> On 24.11.20 16:07, Klemens Muthmann wrote:
>> Hi,
>>
>> Thanks for your reply. I am using processing time instead of event
>> time, since we do get the events in batches and some might arrive
>> days later.
>>
>> But for my current dev setup I just use a CSV dump of finite size as
>> input. I will hand over the pipeline to some other guys, who will
>> need to integrate it with an Apache Kafka Service. Output is written
>> to a Postgres-Database-System.
>>
>> I'll have a look at your proposal and let you know if it worked,
>> after having finished a few prerequisite parts.
>>
>> Regards
>>
>>      Klemens
>>
>> Am 24.11.20 um 12:59 schrieb Timo Walther:
>>> Hi Klemens,
>>>
>>> what you are observing are reasons why event-time should be
>>> preferred over processing-time. Event-time uses the timestamp of
>>> your data while processing-time is to basic for many use cases. Esp.
>>> when you want to reprocess historic data, you want to do that at
>>> full speed instead of waiting 1 hour for 1-hour-windows.
>>>
>>> If you want to use processing-time nevertheless, you need to use a
>>> source that produced unbounded streams instead of bounded streams
>>> such that the pipeline execution theoretically is infinite. Some
>>> documentation can be found here [1] where you need to use the
>>> `FileProcessingMode.PROCESS_CONTINUOUSLY`. But what kind of
>>> connector are you currently using?
>>>
>>> Regards,
>>> Timo
>>>
>>> [1]
>>> https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/datastream_api.html#data-sources 
>>>
>>>
>>> On 24.11.20 09:59, Klemens Muthmann wrote:
>>>> Hi,
>>>>
>>>> I have written an Apache Flink Pipeline containing the following
>>>> piece of code (Java):
>>>>
>>>> stream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(50))).aggregate(new
>>>> CustomAggregator()).print();
>>>>
>>>> If I run the pipeline using local execution I see the following
>>>> behavior. The "CustomAggregator" calls the `createAccumulator` and
>>>> `add` methods correctly with the correct data. However it never
>>>> calls `getResult` and my pipeline simply finishes.
>>>>
>>>> So I did a little research and found out that it works if I change
>>>> the code to:
>>>>
>>>> stream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(1))).aggregate(new
>>>> CustomAggregator()).print();
>>>>
>>>> Notice the reduced gap time for the processing time session window.
>>>> So it seems that execution only continues if the window has been
>>>> closed and if that takes too long, the execution simply aborts. I
>>>> guess another factor playing a part in the problem is, that my
>>>> initial data is read in much faster than 50 seconds. This results
>>>> in the pipeline being in a state where it only waits for the window
>>>> to be closed and having nothing else to do it decides that there is
>>>> no work left and simply shuts down.
>>>>
>>>> My question now is if it is possible to tell the local execution
>>>> environment to wait for that window to be closed, instead of just
>>>> shutting down.
>>>>
>>>> Thanks and Regards
>>>>
>>>>      Klemens Muthmann
>>>>
>>>
>

--
Mit freundlichen Grüßen
           Dr.-Ing. Klemens Muthmann

-----------------------------------
Cyface GmbH
Hertha-Lindner-Straße 10
01067 Dresden
web: www.cyface.de
email: [hidden email]