Hi all,
I am having trouble understanding why addi...
# troubleshootingn
Neha
07/31/2023, 5:13 AMHi all,
I am having trouble understanding why adding a 'UNION ALL' with a 'GROUP BY' condition in Flink is causing the checkpoint to become unbounded. I've attached two Flink SQL queries where the only difference is the addition of one more 'UNION ALL'. Surprisingly, as soon as we add the 'UNION ALL' part to the query, the checkpoint size keeps increasing, which is not expected. We anticipated it would follow the same pattern of increasing and decreasing based on the throughput, as it does without the 'UNION ALL'.
When I analyze the breakdown of the checkpoint size, I find that the 'Interval Join' operator has the largest size. I suspect there might be a bug causing this difference in checkpoint behavior due to the addition of the 'GROUP BY' in the query.
We are using Flink version 1.16.1 with RocksDB as the backend, and incremental checkpointing is enabled.
PS: Interestingly, when I take periodic savepoints for both jobs, their patterns are similar, and the savepoint size is smaller than the checkpoint in the job with the 'UNION ALL'. Attaching the queries in the comment below.
Please let me know if I should file a bug?
Neha
07/31/2023, 5:15 AMWith Union ALL:
select
*
from
(
with p2d as (
select
delivered.job_id,
delivered.store_id,
substring(
GetGeoHash(
CAST(delivered.metadata_json.address.lat as double),
CAST(delivered.metadata_json.address.lng as double)
)
from
1 for 6
) AS cx_gh6,
(
(
cast(
delivered.metadata_json.lastMileDistance as double
) / cast(1000 as double)
) /(
cast(
delivered.updated_at - pickedup.updated_at as double
) / cast(60 * 60 * 1000 as double)
)
) as lm_speed_kmph,
delivered.proctime as proctime
from
awz_s3_OrderLogsEvent pickedup
inner join awz_s3_OrderLogsEvent delivered on delivered.job_id = pickedup.job_id
and delivered.status = 'DELIVERY_DELIVERED'
and delivered.type = 'INSTA'
and pickedup.proctime between delivered.proctime - interval '95' minute
and delivered.proctime + interval '5' minute
where
pickedup.status = 'DELIVERY_PICKEDUP'
and pickedup.type = 'INSTA'
)
select
'lmSpeedKmph_avg_storeId_30m#' || cast(store_id as varchar) as key,
round(Avg(lm_speed_kmph), 4) AS
valuevalueNeha
07/31/2023, 5:15 AMWithout Union all which is showing expected behaviour:
select
*
from
(
with p2d as (
select
delivered.job_id,
delivered.store_id,
substring(
GetGeoHash(
CAST(delivered.metadata_json.address.lat as double),
CAST(delivered.metadata_json.address.lng as double)
)
from
1 for 6
) AS cx_gh6,
(
(
cast(
delivered.metadata_json.lastMileDistance as double
) / cast(1000 as double)
) /(
cast(
delivered.updated_at - pickedup.updated_at as double
) / cast(60 * 60 * 1000 as double)
)
) as lm_speed_kmph,
delivered.proctime as proctime
from
awz_s3_OrderLogsEvent pickedup
inner join awz_s3_OrderLogsEvent delivered on delivered.job_id = pickedup.job_id
and delivered.status = 'DELIVERY_DELIVERED'
and delivered.type = 'INSTA'
and pickedup.proctime between delivered.proctime - interval '95' minute
and delivered.proctime + interval '5' minute
where
pickedup.status = 'DELIVERY_PICKEDUP'
and pickedup.type = 'INSTA'
)
select
'lmSpeedKmph_avg_cxGh6_30m#' || cast(cx_gh6 as varchar) as key,
round(Avg(lm_speed_kmph), 4) AS
valuem
Martijn Visser
07/31/2023, 8:03 AMIs this a batch or a streaming job?
n
Neha
07/31/2023, 8:04 AMstreaming job
m
Martijn Visser
07/31/2023, 8:05 AMMy thought would be that the result of the UNION ALL requires the sources to be kept in state, because you can have the situation where a key doesn't appear anymore in both sources and therefore would need to be removed from the output stream
Martijn Visser
07/31/2023, 8:06 AMBut checkpoint size != your state. It also depends on things like when RocksDB compaction kicks in etc
n
Neha
07/31/2023, 8:14 AMHi @Martijn Visser, I dint really get your first explanation. Union all is just creating parallel set of operators in the Flink UI. Can you please elaborate it more?
Yes i understand the checkpoint size will depend upon compaction but the behaviour is changing when i add the union all. So it is not really related to RocksDB.
Should i create a Bug ticket to understand this in detail?
m
Martijn Visser
07/31/2023, 8:16 AMUnion all is just creating parallel set of operators in the Flink UI. Can you please elaborate it more?My hypothesis would be is that if you apply a UNION ALL, the output of that operation is a changelog stream, not an append-only stream.
n
Neha
07/31/2023, 8:16 AMGot it
Neha
07/31/2023, 8:17 AMIs there any way to verify this hypothesis?
Neha
07/31/2023, 9:16 AMhttps://issues.apache.org/jira/browse/FLINK-32718 Raised a bug.