Scratch that - your WatermarkStrategy DOES work (when I implement
it correctly!).
Well, almost: As you can see below (code pushed to repo), the
Timer events are still appearing somewhat late in the stream - 4
events late in this case. It may be just good-enough for my
purposes, though it will make building test cases painful, so if
you have any ideas how I could fix that, would be much appreciated.
{"ts":0,"id":"0","value":0.01,"is_online":true,"log":"new state. "}
{"ts":0,"id":"1","value":0.02,"is_online":true,"log":"new state. "}
{"ts":0,"id":"2","value":0.03,"is_online":true,"log":"new state. "}
{"ts":1000,"id":"0","value":0.04,"is_online":true,"log":"prevMsg.ts
0 msg_in.ts 1000 Cancelled previous timer. "}
{"ts":1000,"id":"1","value":0.05,"is_online":true,"log":"prevMsg.ts
0 msg_in.ts 1000 Cancelled previous timer. "}
{"ts":2000,"id":"0","value":0.06,"is_online":true,"log":"prevMsg.ts
1000 msg_in.ts 2000 Cancelled previous timer. "}
{"ts":2000,"id":"1","value":0.07,"is_online":true,"log":"prevMsg.ts
1000 msg_in.ts 2000 Cancelled previous timer. "}
{"ts":3000,"id":"0","value":0.08,"is_online":true,"log":"prevMsg.ts
2000 msg_in.ts 3000 Cancelled previous timer. "}
{"ts":1000,"id":"2","is_online":false,"log":"timestamp is 1000"}
{"ts":3000,"id":"1","value":0.09,"is_online":true,"log":"prevMsg.ts
2000 msg_in.ts 3000 Cancelled previous timer. "}
{"ts":4000,"id":"0","value":0.1,"is_online":true,"log":"prevMsg.ts
3000 msg_in.ts 4000 Cancelled previous timer. "}
{"ts":4000,"id":"1","value":0.11,"is_online":true,"log":"prevMsg.ts
3000 msg_in.ts 4000 Cancelled previous timer. "}
{"ts":5000,"id":"0","value":0.12,"is_online":true,"log":"prevMsg.ts
4000 msg_in.ts 5000 Cancelled previous timer. "}
{"ts":5000,"id":"2","value":0.13,"is_online":true,"log":"prevMsg.ts
0 msg_in.ts 5000 Cancelled previous timer. "}
{"ts":6000,"id":"0","value":0.14,"is_online":true,"log":"prevMsg.ts
5000 msg_in.ts 6000 Cancelled previous timer. "}
{"ts":6000,"id":"2","value":0.15,"is_online":true,"log":"prevMsg.ts
5000 msg_in.ts 6000 Cancelled previous timer. "}
{"ts":7000,"id":"0","value":0.16,"is_online":true,"log":"prevMsg.ts
6000 msg_in.ts 7000 Cancelled previous timer. "}
{"ts":5000,"id":"1","is_online":false,"log":"timestamp is 5000"}
{"ts":7000,"id":"1","value":0.17,"is_online":true,"log":"prevMsg.ts
4000 msg_in.ts 7000 Cancelled previous timer. "}
{"ts":7000,"id":"2","value":0.18,"is_online":true,"log":"prevMsg.ts
6000 msg_in.ts 7000 Cancelled previous timer. "}
{"ts":8000,"id":"0","value":0.19,"is_online":true,"log":"prevMsg.ts
7000 msg_in.ts 8000 Cancelled previous timer. "}
{"ts":8000,"id":"1","value":0.2,"is_online":true,"log":"prevMsg.ts
7000 msg_in.ts 8000 Cancelled previous timer. "}
{"ts":8000,"id":"2","value":0.21,"is_online":true,"log":"prevMsg.ts
7000 msg_in.ts 8000 Cancelled previous timer. "}
{"ts":9000,"id":"0","value":0.22,"is_online":true,"log":"prevMsg.ts
8000 msg_in.ts 9000 Cancelled previous timer. "}
{"ts":9000,"id":"1","value":0.23,"is_online":true,"log":"prevMsg.ts
8000 msg_in.ts 9000 Cancelled previous timer. "}
{"ts":9000,"id":"2","value":0.24,"is_online":true,"log":"prevMsg.ts
8000 msg_in.ts 9000 Cancelled previous timer. "}
{"ts":10000,"id":"0","value":0.25,"is_online":true,"log":"prevMsg.ts
9000 msg_in.ts 10000 Cancelled previous timer. "}
{"ts":10000,"id":"1","value":0.26,"is_online":true,"log":"prevMsg.ts
9000 msg_in.ts 10000 Cancelled previous timer. "}
{"ts":10000,"id":"2","value":0.27,"is_online":true,"log":"prevMsg.ts
9000 msg_in.ts 10000 Cancelled previous timer. "}
{"ts":11000,"id":"1","is_online":false,"log":"timestamp is 11000"}
{"ts":11000,"id":"2","is_online":false,"log":"timestamp is 11000"}
{"ts":11000,"id":"0","is_online":false,"log":"timestamp is 11000"}
-Pilgrim
--
Learn more at https://devicepilot.com <https://devicepilot.com>
@devicepilot
<https://t.sidekickopen70.com/s2t/c/5/f18dQhb0S7kv8cpgQZVc6VPt59hl3kW7_k2842PjkFxW2R1KhZ7v4vclW2Rxbb82bzNKzf7GYHvr01?te=W3R5hFj4cm2zwW4fQ47l4fGCmnW3Fbt5S3H4THtF3F6jFSWsSg1&si=5987503666495488&pi=adc3545f-9610-4164-fa4a-2bddbd615e33>
+44 7961 125282
See our latest features
<https://t.sidekickopen70.com/s2t/c/5/f18dQhb0S7kv8cpgQZVc6VPt59hl3kW7_k2842PjkFxW2R1KhZ7v4vclW2Rxbb82bzNKzf7GYHvr01?te=W3R5hFj4cm2zwW3Q_1QY1JxXxvW3SYLMH3T0vWRW1JxwY51LDhHjW3K1M0S1GFyF-3b732&si=5987503666495488&pi=adc3545f-9610-4164-fa4a-2bddbd615e33>
and book me
<https://t.sidekickopen70.com/s2t/c/5/f18dQhb0S7kv8cpgQZVc6VPt59hl3kW7_k2842PjkFxW2R1KhZ7v4vclW2Rxbb82bzNKzf7GYHvr01?te=W3R5hFj4cm2zwW3T1k3k1JxXxvW3SYLMH3T0vWRW1JxwY51LBcj1W4fJfX_4cgB3QW3ZW3jf3_qrT_4Wt5h1&si=5987503666495488&pi=adc3545f-9610-4164-fa4a-2bddbd615e33>
for
a video call.
On Thu, 28 Jan 2021 at 08:37, Pilgrim Beart
<pilgrim.be...@devicepilot.com
<mailto:pilgrim.be...@devicepilot.com>> wrote:
Chesnay,
I cannot reproduce this - I've tried the approaches you
suggest, but nothing I've done makes the timers fire at the
correct time in the stream - they only fire when the stream
has ended. If you have an EventTime example where they fire
at the right time in the stream, I'd love to see it. Or any
ideas for other things to try? Could it perhaps be related to
using a file source?
Thanks,
-Pilgrim
--
Learn more at https://devicepilot.com
<https://devicepilot.com> @devicepilot
<https://t.sidekickopen70.com/s2t/c/5/f18dQhb0S7kv8cpgQZVc6VPt59hl3kW7_k2842PjkFxW2R1KhZ7v4vclW2Rxbb82bzNKzf7GYHvr01?te=W3R5hFj4cm2zwW4fQ47l4fGCmnW3Fbt5S3H4THtF3F6jFSWsSg1&si=5987503666495488&pi=a563531c-b630-4c52-feeb-e32ba4302f87>
+44 7961 125282
See our latest features
<https://t.sidekickopen70.com/s2t/c/5/f18dQhb0S7kv8cpgQZVc6VPt59hl3kW7_k2842PjkFxW2R1KhZ7v4vclW2Rxbb82bzNKzf7GYHvr01?te=W3R5hFj4cm2zwW3Q_1QY1JxXxvW3SYLMH3T0vWRW1JxwY51LDhHjW3K1M0S1GFyF-3b732&si=5987503666495488&pi=a563531c-b630-4c52-feeb-e32ba4302f87>
and book me
<https://t.sidekickopen70.com/s2t/c/5/f18dQhb0S7kv8cpgQZVc6VPt59hl3kW7_k2842PjkFxW2R1KhZ7v4vclW2Rxbb82bzNKzf7GYHvr01?te=W3R5hFj4cm2zwW3T1k3k1JxXxvW3SYLMH3T0vWRW1JxwY51LBcj1W4fJfX_4cgB3QW3ZW3jf3_qrT_4Wt5h1&si=5987503666495488&pi=a563531c-b630-4c52-feeb-e32ba4302f87>
for
a video call.
On Wed, 27 Jan 2021 at 17:55, Chesnay Schepler
<ches...@apache.org <mailto:ches...@apache.org>> wrote:
Actually, if the parallelism is 1 then it works as it
should. sigh....
On 1/27/2021 6:52 PM, Chesnay Schepler wrote:
Note that while this does fix the issue of timers not
firing while the job is running, it seems to be firing
too many timers...
On 1/27/2021 6:49 PM, Chesnay Schepler wrote:
My bad, I was still using the custom WatermarkStrategy
that emits a watermark for each event.
.assignTimestampsAndWatermarks( new
WatermarkStrategy<T>() { @Override public
WatermarkGenerator<T>
createWatermarkGenerator(WatermarkGeneratorSupplier.Context
context) { return new
BoundedOutOfOrdernessWatermarks<T>(Duration.ofSeconds(1))
{ @Override public void onEvent(Tevent, long
eventTimestamp, WatermarkOutput output) {
super.onEvent(event, eventTimestamp, output);
super.onPeriodicEmit(output); } }; } } .withTimestampAssigner(...)
@Aljoscha This is about Flink 1.11. Since the periodic
watermarks are dependent on processing time, am I
assuming correctly if the job finishes quickly that
watermarks may never be emitted (except for those at
the job)? Is there any way to emit periodic watermarks
based on event time?
Is there any way to enable punctuated watermarks for
the existing watermark strategies without having to
implement a custom one?
On 1/27/2021 5:57 PM, Pilgrim Beart wrote:
Chesnay,
Thanks for this - I've made the change you suggested
(setAutoWatermarkInterval) but it hasn't changed the
behaviour - timers still get processed only on stream end.
I have pushed a new version, with this change, and
also emitting some information in a .log field.
If you search for "!!!" in Ingest.java
and DPTimeoutFunction.java you'll see the relevant
changes.
In DPTimeoutFunction you'll see that if I add code to
say "cancel the timer only if it wouldn't have gone
off" then the output is now correct - individual
devices do timeout. However, this output only appears
at the end of the stream (i.e. time jumps backwards as
all the timers are processed) so I still appear not to
be seeing timer processing at the correct event time.
If there was no end of stream, I would never get any
timeouts.
Below is the output I get when I run. This output is
correct but:
a) only because I am manually cancelling timers in
DPTimeoutFunction (search for "!!!")
b) the timer events are timestamped correctly, but are
not emitted into the stream at the right time - and if
the stream didn't end then no timeouts would ever
occur (which in particular means that devices that
never come back online will never get marked as offline).
Perhaps I do need to implement an onPeriodicEmit
function? Does that require a customer watermark
strategy? I can see how to define a custom watermark
at link below, but unclear how to install that?
https://stackoverflow.com/questions/64369613/how-to-add-a-custom-watermarkgenerator-to-a-watermarkstrategy
<https://stackoverflow.com/questions/64369613/how-to-add-a-custom-watermarkgenerator-to-a-watermarkstrategy>
{"ts":0,"id":"0","value":0.01,"is_online":true,"log":"new
state. "}
{"ts":0,"id":"1","value":0.02,"is_online":true,"log":"new
state. "}
{"ts":0,"id":"2","value":0.03,"is_online":true,"log":"new
state. "}
{"ts":1000,"id":"0","value":0.04,"is_online":true,"log":"prevMsg.ts
0 msg_in.ts 1000 Cancelling previous timer. "}
{"ts":1000,"id":"1","value":0.05,"is_online":true,"log":"prevMsg.ts
0 msg_in.ts 1000 Cancelling previous timer. "}
{"ts":2000,"id":"0","value":0.06,"is_online":true,"log":"prevMsg.ts
1000 msg_in.ts 2000 Cancelling previous timer. "}
{"ts":2000,"id":"1","value":0.07,"is_online":true,"log":"prevMsg.ts
1000 msg_in.ts 2000 Cancelling previous timer. "}
{"ts":3000,"id":"0","value":0.08,"is_online":true,"log":"prevMsg.ts
2000 msg_in.ts 3000 Cancelling previous timer. "}
{"ts":3000,"id":"1","value":0.09,"is_online":true,"log":"prevMsg.ts
2000 msg_in.ts 3000 Cancelling previous timer. "}
{"ts":4000,"id":"0","value":0.1,"is_online":true,"log":"prevMsg.ts
3000 msg_in.ts 4000 Cancelling previous timer. "}
{"ts":4000,"id":"1","value":0.11,"is_online":true,"log":"prevMsg.ts
3000 msg_in.ts 4000 Cancelling previous timer. "}
{"ts":5000,"id":"0","value":0.12,"is_online":true,"log":"prevMsg.ts
4000 msg_in.ts 5000 Cancelling previous timer. "}
{"ts":5000,"id":"2","value":0.13,"is_online":true,"log":"prevMsg.ts
0 msg_in.ts 5000 "}
{"ts":6000,"id":"0","value":0.14,"is_online":true,"log":"prevMsg.ts
5000 msg_in.ts 6000 Cancelling previous timer. "}
{"ts":6000,"id":"2","value":0.15,"is_online":true,"log":"prevMsg.ts
5000 msg_in.ts 6000 Cancelling previous timer. "}
{"ts":7000,"id":"0","value":0.16,"is_online":true,"log":"prevMsg.ts
6000 msg_in.ts 7000 Cancelling previous timer. "}
{"ts":7000,"id":"1","value":0.17,"is_online":true,"log":"prevMsg.ts
4000 msg_in.ts 7000 "}
{"ts":7000,"id":"2","value":0.18,"is_online":true,"log":"prevMsg.ts
6000 msg_in.ts 7000 Cancelling previous timer. "}
{"ts":8000,"id":"0","value":0.19,"is_online":true,"log":"prevMsg.ts
7000 msg_in.ts 8000 Cancelling previous timer. "}
{"ts":8000,"id":"1","value":0.2,"is_online":true,"log":"prevMsg.ts
7000 msg_in.ts 8000 Cancelling previous timer. "}
{"ts":8000,"id":"2","value":0.21,"is_online":true,"log":"prevMsg.ts
7000 msg_in.ts 8000 Cancelling previous timer. "}
{"ts":9000,"id":"0","value":0.22,"is_online":true,"log":"prevMsg.ts
8000 msg_in.ts 9000 Cancelling previous timer. "}
{"ts":9000,"id":"1","value":0.23,"is_online":true,"log":"prevMsg.ts
8000 msg_in.ts 9000 Cancelling previous timer. "}
{"ts":9000,"id":"2","value":0.24,"is_online":true,"log":"prevMsg.ts
8000 msg_in.ts 9000 Cancelling previous timer. "}
{"ts":10000,"id":"0","value":0.25,"is_online":true,"log":"prevMsg.ts
9000 msg_in.ts 10000 Cancelling previous timer. "}
{"ts":10000,"id":"1","value":0.26,"is_online":true,"log":"prevMsg.ts
9000 msg_in.ts 10000 Cancelling previous timer. "}
{"ts":10000,"id":"2","value":0.27,"is_online":true,"log":"prevMsg.ts
9000 msg_in.ts 10000 Cancelling previous timer. "}
{"ts":1001,"id":"2","is_online":false} // These are
the "going offline" events that we want to see. But
they are emitted only once the stream has ended.
{"ts":5001,"id":"1","is_online":false}
{"ts":11001,"id":"1","is_online":false}
{"ts":11001,"id":"0","is_online":false}
{"ts":11001,"id":"2","is_online":false}
Thanks,
-Pilgrim
--
Learn more at https://devicepilot.com
<https://devicepilot.com> @devicepilot
<https://t.sidekickopen70.com/s2t/c/5/f18dQhb0S7kv8cpgQZVc6VPt59hl3kW7_k2842PjkFxW2R1KhZ7v4vclW2Rxbb82bzNKzf7GYHvr01?te=W3R5hFj4cm2zwW4fQ47l4fGCmnW3Fbt5S3H4THtF3F6jFSWsSg1&si=5987503666495488&pi=5558f660-ceb0-461c-af58-a0d24fbcef34>
+44 7961 125282
See our latest features
<https://t.sidekickopen70.com/s2t/c/5/f18dQhb0S7kv8cpgQZVc6VPt59hl3kW7_k2842PjkFxW2R1KhZ7v4vclW2Rxbb82bzNKzf7GYHvr01?te=W3R5hFj4cm2zwW3Q_1QY1JxXxvW3SYLMH3T0vWRW1JxwY51LDhHjW3K1M0S1GFyF-3b732&si=5987503666495488&pi=5558f660-ceb0-461c-af58-a0d24fbcef34>
and book me
<https://t.sidekickopen70.com/s2t/c/5/f18dQhb0S7kv8cpgQZVc6VPt59hl3kW7_k2842PjkFxW2R1KhZ7v4vclW2Rxbb82bzNKzf7GYHvr01?te=W3R5hFj4cm2zwW3T1k3k1JxXxvW3SYLMH3T0vWRW1JxwY51LBcj1W4fJfX_4cgB3QW3ZW3jf3_qrT_4Wt5h1&si=5987503666495488&pi=5558f660-ceb0-461c-af58-a0d24fbcef34>
for
a video call.
On Wed, 27 Jan 2021 at 14:09, Chesnay Schepler
<ches...@apache.org <mailto:ches...@apache.org>> wrote:
||
You were right that it is an issue with the
watermarks; outside of the when the job was
stopped they were never emitted downstream, so no
timer was ever triggered.
It appears that you need to set the
setAutoWatermarkInterval in the ExecutionConfig via
env.getConfig().setAutoWatermarkInterval(Duration.ofMillis(500).toMillis());
to have them periodically emitted. Alternatively
you could override
BoundedOutOfOrdernessWatermarks#onEvent to also
emit a watermark for event (for example, by
calling #onPeriodicEmit).
Put another way, if you use any of the built-in
WatermarkGenerators and use event-time, then it
appears that you *must* set this interval.
This behavior is...less than ideal I must admit,
and it does not appear to be properly documented.
On 1/27/2021 1:56 PM, Chesnay Schepler wrote:
Based on your description you aren't doing
anything obviously wrong.
Would it be possible for you to share the code
with us?
On 1/27/2021 1:02 PM, Pilgrim Beart wrote:
A newbie question:
I've created a basic Flink DataStream job for an
IoT use-case, with file source and sink for testing.
I key by device ID, then in a ProcessFunction
set an EventTime Timer to fire if a device falls
silent, i.e. a timeout, which I cancel if
another message arrives from that device within
the timeout.
My test source generates 3 devices, one of which
falls silent for more than the timeout period
during the stream, then resumes again. So I
expect the Timer to fire for that device during
the stream, and then for all the Timers to fire
after the end of the stream.
The timers do indeed fire at the end of the
stream (e.g. with a timeout of 1000, the timers
all fire 1000 after the end of the stream, which
is correct). But no timer fires for the device
which falls silent during the stream (even
though other devices are still talking,
advancing event time). I've verified that I am
keying correctly by ID.
I suspect this is something to do with
Watermarks. I'm using forBoundedOutOfOrderness
watermarking with a duration of 0.
All suggestions welcome, thanks.
-Pilgrim
--
Learn more at https://devicepilot.com
<https://devicepilot.com> @devicepilot
<https://t.sidekickopen70.com/s2t/c/5/f18dQhb0S7kv8cpgQZVc6VPt59hl3kW7_k2842PjkFxW2R1KhZ7v4vclW2Rxbb82bzNKzf7GYHvr01?te=W3R5hFj4cm2zwW4fQ47l4fGCmnW3Fbt5S3H4THtF3F6jFSWsSg1&si=5987503666495488&pi=6c5d2342-2e03-4926-a6bd-49f564169759>
+44 7961 125282
See our latest features
<https://t.sidekickopen70.com/s2t/c/5/f18dQhb0S7kv8cpgQZVc6VPt59hl3kW7_k2842PjkFxW2R1KhZ7v4vclW2Rxbb82bzNKzf7GYHvr01?te=W3R5hFj4cm2zwW3Q_1QY1JxXxvW3SYLMH3T0vWRW1JxwY51LDhHjW3K1M0S1GFyF-3b732&si=5987503666495488&pi=6c5d2342-2e03-4926-a6bd-49f564169759>
and book me
<https://t.sidekickopen70.com/s2t/c/5/f18dQhb0S7kv8cpgQZVc6VPt59hl3kW7_k2842PjkFxW2R1KhZ7v4vclW2Rxbb82bzNKzf7GYHvr01?te=W3R5hFj4cm2zwW3T1k3k1JxXxvW3SYLMH3T0vWRW1JxwY51LBcj1W4fJfX_4cgB3QW3ZW3jf3_qrT_4Wt5h1&si=5987503666495488&pi=6c5d2342-2e03-4926-a6bd-49f564169759>
for
a video call.
