Need help on Streaming API | Flink | GlobalWindow and Customized Trigger

[Samim Ahmed]

[Aljoscha Krettek]

Hi,

If your could give us a look at your custom Trigger we might be able to figure out what’s going on.

Best,

Aljoscha

On 22. May 2017, at 09:06, Samim Ahmed <[hidden email]> wrote:

[Aljoscha Krettek]

Hi,

I think the problem is that the Trigger only uses processing time to determine when to trigger. If the job shuts down (which happens when the sources shut down and the whole pipeline is flushed) pending processing-time triggers are not fired.

You can use the fact that when sources shutdown they emit a Long.MAX_VALUE watermark to signal that there won’t be any data in the future. For this, you have to enable watermarks (you can do this via env.setStreamTimeCharacteristic(EventTime)) and then set an event-time timer in your Trigger for Long.MAX_VALUE. This will call your onEventTime() method and allow the Trigger to fire before shutting down.

Best,

Aljoscha

On 24. May 2017, at 07:08, Samim Ahmed <[hidden email]> wrote:

Hello Aljoscha,

Sorry !!! , I am asking for updates on my issue. 

Did you get a chance to have a look on the trigger files, I am blocked. Please have a look if you have time in your hand. 

Waiting for your answer and again sorry for this mail.

Regards,

Samim 

On Tue, May 23, 2017 at 1:00 AM, Samim Ahmed <[hidden email]> wrote:

Hello Aljoscha,

Thanks for have look on this issue.

I have coppied the mail code for the flow of execution and the trigger code are attached with this mail.

In main class:

AggregationProfile vtrAggProfile = new VtrAggregationProfile();

1. decode the input file

2. Filter Events

DataStream<EventBean> vtrFilteredStream = vtrDecodedEventsDs.filter(vtrAggProfile).setParallelism(vtrParserParallelism);

3. Correlate VTR Records 

DataStream<ISession> vtrSessionStream=vtrFilteredStream

.keyBy(vtrAggProfile)

.window(GlobalWindows.create())

.trigger(vtrAggProfile)  <=== this is the trigger creates problem for last few minutes data.

.apply(vtrAggProfile).setParallelism(maxParallelism);

Attached file names :

1. VerAggregationProfile.java

2. AggregationProfile.java

Please let me know if you need any other information. Thanks in advance .

//Regards,

Samim

On Mon, May 22, 2017 at 6:30 PM, Aljoscha Krettek <[hidden email]> wrote:

Hi,

If your could give us a look at your custom Trigger we might be able to figure out what’s going on.

Best,

Aljoscha

On 22. May 2017, at 09:06, Samim Ahmed <[hidden email]> wrote:

-- 

--