https://flink.apache.org/ logo
Join Slack
Powered by
Hi, I have a problem understanding how `CURRENT_WA...
# troubleshooting
p
Hi, I have a problem understanding how 
CURRENT_WATERMARK()
function works in Flink SQL. Let’s say I have a source table:
Copy code
create table source (
   symbol STRING,
   ts TIMESTAMP,
   WATERMARK FOR ts AS ts - INTERVAL '1' MINUTE
)
WITH (
….kafka connection properties….
);
Now I have some windows aggregation:
Copy code
CREATE VIEW windows_1m AS (
	select symbol, count(*), window_start, window_end
	from TABLE(TUMBLE(TABLE source, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
	GROUP by symbol, window_start, window_end
)
But I would like also to identify late events, so I take:
Copy code
CREATE VIEW events_with_current_watermark AS (
	select *, CURRENT_WATERMARK(ts)
	from source
);
Now I insert both my 
windows_1m
and 
events_with_current_watermark
into jdbc DB. Let’s say my DB tables look like: windows_1m:
Copy code
symbol|count |window_start       |window_end         |
------+--------------------------+-------------------+
s1    |1     |2023-07-20 07:04:00|2023-07-20 07:05:00|
events_with_current_watermark:
Copy code
symbol|ts                     |current_watermark      |
------+-----------------------+-----------------------+
s1    |2023-07-20 07:04:01.429|2023-07-20 07:04:16.612|
so in my DB in 
events_with_current_watermark
table the event seems to be late (
ts < current_watermark
) however 1 minute window was created and contains this particular event (so while processing it wasn’t late?) I’m quite confused, could it be that saving 
events_with_current_watermark
to DB was after processing windows (
windows_1m
) and 
CURRENT_WATERMARK(ts)
returned different values corresponding to different points in time?
m
This depends on the data that you have in your source
Especially since you're defining bounded out of orderness
p
if I understand correctly allowed lateness only shifts watermark by given value (1 min in this example) and it is already reflected in values returned by 
CURRENT_WATERMARK()
function. shouldn’t 
CURRENT_WATERMARK()
return the same, deterministic value at every stage of processing? in the above example we see event that was processed within allowed out of orderness (and put into window) but at the same time source data with 
current_watermark
value attached suggests it was a late event and should be ignored. isn’t it contradictive?
m
Watermark progression depends in the next input records (or a defined idleness)
p
ok, that’s understood 🙂 so in my case my 1min-windows operator has a different current_watermark and my sink (
events_with_current_watermark
) uses different watermark for 
current_watermark
column? so in other words 
current_watermark()
value doesn’t correspond to given event but to the current 
ts
in source?
I imagine my pipeline looks like this:
so even both red and blue rectangles are using current_watermark calling 
current_watermark()
function in the same time doesn’t guarantee the returned timestamp will be equal? that would be even fine since the watermark changes dynamically depending on the values in source but with the setup above the difference between blue and red are sometimes > 1 min
or maybe I should describe what I want to achieve I would like to know which events were late and ignored because of the lateness when creating 1min windows
And send such events to DB so I can use them for other processing
should I first create views with events and late events before I start processing them?
Copy code
create table source (
   symbol STRING,
   ts TIMESTAMP,
   WATERMARK FOR ts AS ts - INTERVAL '1' MINUTE
)
WITH (
….kafka connection properties….
);
Copy code
create view events AS (
   SELECT
   symbol,
   ts
   FROM source where ts > CURRENT_WATERMARK(ts)
);

create view late_events AS (
   SELECT
   symbol,
   ts
   FROM source where ts <= CURRENT_WATERMARK(ts)
);
and then create windows based only on 
events
table:
Copy code
CREATE VIEW windows_1m AS (
	select symbol, count(*), window_start, window_end
	from TABLE(TUMBLE(TABLE events, DESCRIPTOR(ts), INTERVAL '1' MINUTE))
	GROUP by symbol, window_start, window_end
)
is this aproach correct? 🔼
Open in Slack
PreviousNext
Powered by