You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "garcia (Jira)" <ji...@apache.org> on 2024/04/23 13:40:00 UTC
[jira] [Updated] (FLINK-35218) Duplicated values caused by expired state TTL
[ https://issues.apache.org/jira/browse/FLINK-35218?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
garcia updated FLINK-35218:
---------------------------
Description:
Hi,
We utilize the state TTL to clean our Flink input tables through the `table.exec.state.ttl` configuration.
However, we encountered an issue when the TTL expires, as illustrated in our scenario:
Given this input_table
{code:java}
{
"$schema": "http://json-schema.org/draft-07/schema",
"$id": "http://example.com/example.json",
"type": "object",
"title": "bet placed schema",
"required": [
"placement_date"
],
"properties": {
"bet_id": {
"$id": "#/properties/bet_id",
"type": "string"
},
"regulator": {
"$id": "#/properties/regulator",
"type": "string"
},
"match_id": {
"$id": "#/properties/match_id",
"type": "integer"
},
"combo_id": {
"$id": "#/properties/combo_id",
"type": "integer"
},
"is_live": {
"$id": "#/properties/is_live",
"type": "boolean"
},
"offer_catalog": {
"$id": "#/properties/offer_catalog",
"type": "string"
},
"combo_selection_nbr": {
"$id": "#/properties/combo_selection_nbr",
"type": "integer"
}
},
"additionalProperties": true
} {code}
This configuration:
{code:java}
import org.apache.flink.streaming.api.environment.{StreamExecutionEnvironment => JavaStreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}val streamEnv = new StreamExecutionEnvironment(JavaStreamExecutionEnvironment.getExecutionEnvironment(conf))val tableEnv = StreamTableEnvironment.create(env)
tableEnv.getConfig.getConfiguration
.setString("table.local-time-zone", "UTC")
tableEnv.getConfig.getConfiguration
.setString("table.exec.mini-batch.enabled", "true")
tableEnv.getConfig.getConfiguration
.setString("table.exec.mini-batch.allow-latency", "5 s")
tableEnv.getConfig.getConfiguration
.setString("table.exec.mini-batch.size", "5000")
tableEnv.getConfig.getConfiguration
.setString("table.exec.state.ttl", TimeUnit.MILLISECONDS.convert(1, TimeUnit.MINUTES).toString) {code}
And this query (simplified query):
{code:java}
WITH exploded_combos AS (
select
event_timestamp,
CAST(JSON_VALUE(combos.combo, '$.match_id') AS BIGINT) as match_id,
CAST(
JSON_VALUE(combos.combo, '$.combo_selection_id') AS BIGINT
) as combo_id,
CAST(JSON_VALUE(combos.combo, '$.is_live') AS BOOLEAN) as is_live,
CAST(RegulatorToCatalog(regulator) AS VARCHAR) as offer_catalog,
CARDINALITY(
JsonToArray(JSON_QUERY(combos.combo, '$.bet_selections'))
) as combo_selections_nbr,
combo_bet_selections
from
bet_placed_view
CROSS JOIN UNNEST(JsonToArray(combo_bet_selections)) AS combos(combo)
),
agg_match AS (
SELECT
match_id,
LOWER(offer_catalog) as offer_catalog,
MAX(event_timestamp) AS last_event_time_utc,
COUNT(*) AS bet_count
FROM
exploded_combos
WHERE
match_id IS NOT NULL
AND combo_id IS NOT NULL
AND offer_catalog IS NOT NULL
AND combo_bet_selections IS NOT NULL
GROUP BY
match_id,
LOWER(offer_catalog)
),
agg_combo AS (
SELECT
match_id,
combo_id,
combo_selections_nbr,
is_live,
LOWER(offer_catalog) AS offer_catalog,
MAX(event_timestamp) AS last_event_time_utc,
COUNT(*) as bet_count
FROM
exploded_combos
WHERE
match_id IS NOT NULL
AND combo_id IS NOT NULL
AND (
combo_selections_nbr = 3
OR combo_selections_nbr = 2
)
AND offer_catalog IS NOT NULL
GROUP BY
match_id,
combo_id,
combo_selections_nbr,
is_live,
LOWER(offer_catalog)
),
ranked_filtered_agg_combo_main_page AS (
SELECT
match_id,
combo_id,
offer_catalog,
bet_count,
ROW_NUMBER() OVER (
PARTITION BY match_id,
offer_catalog
ORDER BY
bet_count DESC,
combo_id DESC
) AS rank_combo
FROM
agg_combo
WHERE
combo_selections_nbr = 3
),
joined_filtered_agg_match_main_page AS (
SELECT
ranked_filtered_agg_combo_main_page.match_id,
ranked_filtered_agg_combo_main_page.offer_catalog,
ranked_filtered_agg_combo_main_page.bet_count,
ranked_filtered_agg_combo_main_page.combo_id,
ROW_NUMBER() OVER (
PARTITION BY agg_match.offer_catalog
ORDER BY
agg_match.bet_count DESC,
agg_match.match_id DESC
) AS rank_match
FROM
agg_match
INNER JOIN ranked_filtered_agg_combo_main_page ON ranked_filtered_agg_combo_main_page.match_id = agg_match.match_id
AND ranked_filtered_agg_combo_main_page.offer_catalog = agg_match.offer_catalog
WHERE
ranked_filtered_agg_combo_main_page.rank_combo = 1
)
SELECT
partition_key,
match_id,
offer_catalog,
false as live,
LAST_VALUE(last_event_utc) AS last_event_utc,
LAST_VALUE(last_event_utc) AS max_last_event_utc,
LAST_VALUE(top) AS topFROM (
SELECT
'<all>' as match_id,
offer_catalog,
'<all>' || '#openmatch#' || offer_catalog as partition_key,
CAST('MAX(very_last_event_time_utc)' AS VARCHAR) AS last_event_utc,
'[' || LISTAGG(
'{"match_id":' || CAST(match_id AS VARCHAR) || ',"rank":' || CAST(rank_match AS VARCHAR) || ',"count":' || CAST(bet_count AS VARCHAR) || ',"combo_id":"' || CAST(combo_id AS VARCHAR) || '","offer_catalog":"' || CAST(offer_catalog AS VARCHAR) || '","live": ' || '}'
) || ']' AS top
FROM
joined_filtered_agg_match_main_page
WHERE
rank_match <= 5
GROUP BY
offer_catalog
)
GROUP BY
partition_key,
match_id,
offer_catalog{code}
As you can see in the result below, when the TTL is reached, we have duplicate values in our output.
{code:java}
+I[<all>#openmatch#fr, <all>, fr,
false, MAX(very_last_event_time_utc), MAX(very_last_event_time_utc),
[
{
"match_id": 2,
"rank": 1,
"count": 2,
"combo_id": "123",
"offer_catalog": "fr",
"live":
},
{
"match_id": 5,
"rank": 2,
"count": 1,
"combo_id": "123",
"offer_catalog": "fr",
"live":
}
]
]
+U[<all>#openmatch#fr, <all>, fr,
false, MAX(very_last_event_time_utc), MAX(very_last_event_time_utc),
[
{
"match_id": 5,
"rank": 2,
"count": 1,
"combo_id": "123",
"offer_catalog": "fr",
"live":
},
{
"match_id": 2,
"rank": 1,
"count": 3,
"combo_id": "123",
"offer_catalog": "fr",
"live":
},
{
"match_id": 3,
"rank": 3,
"count": 2,
"combo_id": "123",
"offer_catalog": "fr",
"live":
},
{
"match_id": 1,
"rank": 4,
"count": 3,
"combo_id": "123",
"offer_catalog": "fr",
"live":
},
{
"match_id": 4,
"rank": 5,
"count": 1,
"combo_id": "123",
"offer_catalog": "fr",
"live":
}
]
]
...
// DUPLICATED VALUES (multiple rank) :
+U[<all>#openmatch#fr, <all>, fr,
false, MAX(very_last_event_time_utc), MAX(very_last_event_time_utc),
[
{
"match_id": 3,
"rank": 1,
"count": 16,
"combo_id": "123",
"offer_catalog": "fr",
"live":
},
{
"match_id": 5,
"rank": 3,
"count": 14,
"combo_id": "123",
"offer_catalog": "fr",
"live":
},
{
"match_id": 2,
"rank": 2,
"count": 15,
"combo_id": "123",
"offer_catalog": "fr",
"live":
},
{
"match_id": 4,
"rank": 4,
"count": 14,
"combo_id": "123",
"offer_catalog": "fr",
"live":
},
{
"match_id": 1,
"rank": 5,
"count": 12,
"combo_id": "123",
"offer_catalog": "fr",
"live":
},
{
"match_id": 2,
"rank": 2,
"count": 15,
"combo_id": "123",
"offer_catalog": "fr",
"live":
},
{
"match_id": 1,
"rank": 5,
"count": 12,
"combo_id": "123",
"offer_catalog": "fr",
"live":
},
{
"match_id": 5,
"rank": 3,
"count": 16,
"combo_id": "123",
"offer_catalog": "fr",
"live":
},
{
"match_id": 3,
"rank": 1,
"count": 18,
"combo_id": "123",
"offer_catalog": "fr",
"live":
},
{
"match_id": 4,
"rank": 4,
"count": 15,
"combo_id": "123",
"offer_catalog": "fr",
"live":
}
]
]
... {code}
*Is this a bug or did we misunderstand how to implement the ttl?*
Flink is also sending to this warning:
!image-2024-04-23-15-34-32-860.png!
We actually found 2 workaround:
* set a verry large ttl (1year) This will eliminate most of the problems thanks to our business logic
* Use sliding window (but if we do that we need window of 50days sliding each minute)
We're afraid that these two solutions will be very expensive
Can you tell us if this is a normal behavior or if we miss understand something about the TTL?
was:
Hi,
We utilize the state TTL to clean our Flink input tables through the `table.exec.state.ttl` configuration.
However, we encountered an issue when the TTL expires, as illustrated in our scenario:
Given this input_table
{code:java}
{
"$schema": "http://json-schema.org/draft-07/schema",
"$id": "http://example.com/example.json",
"type": "object",
"title": "bet placed schema",
"required": [
"placement_date"
],
"properties": {
"bet_id": {
"$id": "#/properties/bet_id",
"type": "string"
},
"regulator": {
"$id": "#/properties/regulator",
"type": "string"
},
"match_id": {
"$id": "#/properties/match_id",
"type": "integer"
},
"combo_id": {
"$id": "#/properties/combo_id",
"type": "integer"
},
"is_live": {
"$id": "#/properties/is_live",
"type": "boolean"
},
"offer_catalog": {
"$id": "#/properties/offer_catalog",
"type": "string"
},
"combo_selection_nbr": {
"$id": "#/properties/combo_selection_nbr",
"type": "integer"
}
},
"additionalProperties": true
} {code}
This configuration:
{code:java}
import org.apache.flink.streaming.api.environment.{StreamExecutionEnvironment => JavaStreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}val streamEnv = new StreamExecutionEnvironment(JavaStreamExecutionEnvironment.getExecutionEnvironment(conf))val tableEnv = StreamTableEnvironment.create(env)
tableEnv.getConfig.getConfiguration
.setString("table.local-time-zone", "UTC")
tableEnv.getConfig.getConfiguration
.setString("table.exec.mini-batch.enabled", "true")
tableEnv.getConfig.getConfiguration
.setString("table.exec.mini-batch.allow-latency", "5 s")
tableEnv.getConfig.getConfiguration
.setString("table.exec.mini-batch.size", "5000")
tableEnv.getConfig.getConfiguration
.setString("table.exec.state.ttl", TimeUnit.MILLISECONDS.convert(1, TimeUnit.MINUTES).toString) {code}
And this query (simplified query):
{code:java}
WITH exploded_combos AS (
select
event_timestamp,
CAST(JSON_VALUE(combos.combo, '$.match_id') AS BIGINT) as match_id,
CAST(
JSON_VALUE(combos.combo, '$.combo_selection_id') AS BIGINT
) as combo_id,
CAST(JSON_VALUE(combos.combo, '$.is_live') AS BOOLEAN) as is_live,
CAST(RegulatorToCatalog(regulator) AS VARCHAR) as offer_catalog,
CARDINALITY(
JsonToArray(JSON_QUERY(combos.combo, '$.bet_selections'))
) as combo_selections_nbr,
combo_bet_selections
from
bet_placed_view
CROSS JOIN UNNEST(JsonToArray(combo_bet_selections)) AS combos(combo)
),
agg_match AS (
SELECT
match_id,
LOWER(offer_catalog) as offer_catalog,
MAX(event_timestamp) AS last_event_time_utc,
COUNT(*) AS bet_count
FROM
exploded_combos
WHERE
match_id IS NOT NULL
AND combo_id IS NOT NULL
AND offer_catalog IS NOT NULL
AND combo_bet_selections IS NOT NULL
GROUP BY
match_id,
LOWER(offer_catalog)
),
agg_combo AS (
SELECT
match_id,
combo_id,
combo_selections_nbr,
is_live,
LOWER(offer_catalog) AS offer_catalog,
MAX(event_timestamp) AS last_event_time_utc,
COUNT(*) as bet_count
FROM
exploded_combos
WHERE
match_id IS NOT NULL
AND combo_id IS NOT NULL
AND (
combo_selections_nbr = 3
OR combo_selections_nbr = 2
)
AND offer_catalog IS NOT NULL
GROUP BY
match_id,
combo_id,
combo_selections_nbr,
is_live,
LOWER(offer_catalog)
),
ranked_filtered_agg_combo_main_page AS (
SELECT
match_id,
combo_id,
offer_catalog,
bet_count,
ROW_NUMBER() OVER (
PARTITION BY match_id,
offer_catalog
ORDER BY
bet_count DESC,
combo_id DESC
) AS rank_combo
FROM
agg_combo
WHERE
combo_selections_nbr = 3
),
joined_filtered_agg_match_main_page AS (
SELECT
ranked_filtered_agg_combo_main_page.match_id,
ranked_filtered_agg_combo_main_page.offer_catalog,
ranked_filtered_agg_combo_main_page.bet_count,
ranked_filtered_agg_combo_main_page.combo_id,
ROW_NUMBER() OVER (
PARTITION BY agg_match.offer_catalog
ORDER BY
agg_match.bet_count DESC,
agg_match.match_id DESC
) AS rank_match
FROM
agg_match
INNER JOIN ranked_filtered_agg_combo_main_page ON ranked_filtered_agg_combo_main_page.match_id = agg_match.match_id
AND ranked_filtered_agg_combo_main_page.offer_catalog = agg_match.offer_catalog
WHERE
ranked_filtered_agg_combo_main_page.rank_combo = 1
)
SELECT
partition_key,
match_id,
offer_catalog,
false as live,
LAST_VALUE(last_event_utc) AS last_event_utc,
LAST_VALUE(last_event_utc) AS max_last_event_utc,
LAST_VALUE(top) AS topFROM (
SELECT
'<all>' as match_id,
offer_catalog,
'<all>' || '#openmatch#' || offer_catalog as partition_key,
CAST('MAX(very_last_event_time_utc)' AS VARCHAR) AS last_event_utc,
'[' || LISTAGG(
'{"match_id":' || CAST(match_id AS VARCHAR) || ',"rank":' || CAST(rank_match AS VARCHAR) || ',"count":' || CAST(bet_count AS VARCHAR) || ',"combo_id":"' || CAST(combo_id AS VARCHAR) || '","offer_catalog":"' || CAST(offer_catalog AS VARCHAR) || '","live": ' || '}'
) || ']' AS top
FROM
joined_filtered_agg_match_main_page
WHERE
rank_match <= 5
GROUP BY
offer_catalog
)
GROUP BY
partition_key,
match_id,
offer_catalog{code}
As you can see in the result below, when the TTL is reached, we have duplicate values in our output.
{code:java}
+I[<all>#openmatch#betclic_fr, <all>, betclic_fr,
false, MAX(very_last_event_time_utc), MAX(very_last_event_time_utc),
[
{
"match_id": 2,
"rank": 1,
"count": 2,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
},
{
"match_id": 5,
"rank": 2,
"count": 1,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
}
]
]
+U[<all>#openmatch#betclic_fr, <all>, betclic_fr,
false, MAX(very_last_event_time_utc), MAX(very_last_event_time_utc),
[
{
"match_id": 5,
"rank": 2,
"count": 1,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
},
{
"match_id": 2,
"rank": 1,
"count": 3,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
},
{
"match_id": 3,
"rank": 3,
"count": 2,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
},
{
"match_id": 1,
"rank": 4,
"count": 3,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
},
{
"match_id": 4,
"rank": 5,
"count": 1,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
}
]
]
...
// DUPLICATED VALUES (multiple times the same rank):
+U[<all>#openmatch#betclic_fr, <all>, betclic_fr,
false, MAX(very_last_event_time_utc), MAX(very_last_event_time_utc),
[
{
"match_id": 3,
"rank": 1,
"count": 16,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
},
{
"match_id": 5,
"rank": 3,
"count": 14,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
},
{
"match_id": 2,
"rank": 2,
"count": 15,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
},
{
"match_id": 4,
"rank": 4,
"count": 14,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
},
{
"match_id": 1,
"rank": 5,
"count": 12,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
},
{
"match_id": 2,
"rank": 2,
"count": 15,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
},
{
"match_id": 1,
"rank": 5,
"count": 12,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
},
{
"match_id": 5,
"rank": 3,
"count": 16,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
},
{
"match_id": 3,
"rank": 1,
"count": 18,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
},
{
"match_id": 4,
"rank": 4,
"count": 15,
"combo_id": "123",
"offer_catalog": "betclic_fr",
"live":
}
]
]
... {code}
*Is this a bug or did we misunderstand how to implement the ttl?*
Flink is also sending to this warning:
!image-2024-04-23-15-34-32-860.png!
We actually found 2 workaround:
* set a verry large ttl (1year) This will eliminate most of the problems thanks to our business logic
* Use sliding window (but if we do that we need window of 50days sliding each minute)
We're afraid that these two solutions will be very expensive
Can you tell us if this is a normal behavior or if we miss understand something about the TTL?
> Duplicated values caused by expired state TTL
> ------------------------------------------------
>
> Key: FLINK-35218
> URL: https://issues.apache.org/jira/browse/FLINK-35218
> Project: Flink
> Issue Type: Bug
> Reporter: garcia
> Priority: Major
> Attachments: image-2024-04-23-15-34-32-860.png
>
>
> Hi,
> We utilize the state TTL to clean our Flink input tables through the `table.exec.state.ttl` configuration.
> However, we encountered an issue when the TTL expires, as illustrated in our scenario:
> Given this input_table
> {code:java}
> {
> "$schema": "http://json-schema.org/draft-07/schema",
> "$id": "http://example.com/example.json",
> "type": "object",
> "title": "bet placed schema",
> "required": [
> "placement_date"
> ],
> "properties": {
> "bet_id": {
> "$id": "#/properties/bet_id",
> "type": "string"
> },
> "regulator": {
> "$id": "#/properties/regulator",
> "type": "string"
> },
> "match_id": {
> "$id": "#/properties/match_id",
> "type": "integer"
> },
> "combo_id": {
> "$id": "#/properties/combo_id",
> "type": "integer"
> },
> "is_live": {
> "$id": "#/properties/is_live",
> "type": "boolean"
> },
> "offer_catalog": {
> "$id": "#/properties/offer_catalog",
> "type": "string"
> },
> "combo_selection_nbr": {
> "$id": "#/properties/combo_selection_nbr",
> "type": "integer"
> }
> },
> "additionalProperties": true
> } {code}
> This configuration:
> {code:java}
> import org.apache.flink.streaming.api.environment.{StreamExecutionEnvironment => JavaStreamExecutionEnvironment}
> import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}val streamEnv = new StreamExecutionEnvironment(JavaStreamExecutionEnvironment.getExecutionEnvironment(conf))val tableEnv = StreamTableEnvironment.create(env)
> tableEnv.getConfig.getConfiguration
> .setString("table.local-time-zone", "UTC")
> tableEnv.getConfig.getConfiguration
> .setString("table.exec.mini-batch.enabled", "true")
> tableEnv.getConfig.getConfiguration
> .setString("table.exec.mini-batch.allow-latency", "5 s")
> tableEnv.getConfig.getConfiguration
> .setString("table.exec.mini-batch.size", "5000")
> tableEnv.getConfig.getConfiguration
> .setString("table.exec.state.ttl", TimeUnit.MILLISECONDS.convert(1, TimeUnit.MINUTES).toString) {code}
> And this query (simplified query):
> {code:java}
> WITH exploded_combos AS (
> select
> event_timestamp,
> CAST(JSON_VALUE(combos.combo, '$.match_id') AS BIGINT) as match_id,
> CAST(
> JSON_VALUE(combos.combo, '$.combo_selection_id') AS BIGINT
> ) as combo_id,
> CAST(JSON_VALUE(combos.combo, '$.is_live') AS BOOLEAN) as is_live,
> CAST(RegulatorToCatalog(regulator) AS VARCHAR) as offer_catalog,
> CARDINALITY(
> JsonToArray(JSON_QUERY(combos.combo, '$.bet_selections'))
> ) as combo_selections_nbr,
> combo_bet_selections
> from
> bet_placed_view
> CROSS JOIN UNNEST(JsonToArray(combo_bet_selections)) AS combos(combo)
> ),
> agg_match AS (
> SELECT
> match_id,
> LOWER(offer_catalog) as offer_catalog,
> MAX(event_timestamp) AS last_event_time_utc,
> COUNT(*) AS bet_count
> FROM
> exploded_combos
> WHERE
> match_id IS NOT NULL
> AND combo_id IS NOT NULL
> AND offer_catalog IS NOT NULL
> AND combo_bet_selections IS NOT NULL
> GROUP BY
> match_id,
> LOWER(offer_catalog)
> ),
> agg_combo AS (
> SELECT
> match_id,
> combo_id,
> combo_selections_nbr,
> is_live,
> LOWER(offer_catalog) AS offer_catalog,
> MAX(event_timestamp) AS last_event_time_utc,
> COUNT(*) as bet_count
> FROM
> exploded_combos
> WHERE
> match_id IS NOT NULL
> AND combo_id IS NOT NULL
> AND (
> combo_selections_nbr = 3
> OR combo_selections_nbr = 2
> )
> AND offer_catalog IS NOT NULL
> GROUP BY
> match_id,
> combo_id,
> combo_selections_nbr,
> is_live,
> LOWER(offer_catalog)
> ),
> ranked_filtered_agg_combo_main_page AS (
> SELECT
> match_id,
> combo_id,
> offer_catalog,
> bet_count,
> ROW_NUMBER() OVER (
> PARTITION BY match_id,
> offer_catalog
> ORDER BY
> bet_count DESC,
> combo_id DESC
> ) AS rank_combo
> FROM
> agg_combo
> WHERE
> combo_selections_nbr = 3
> ),
> joined_filtered_agg_match_main_page AS (
> SELECT
> ranked_filtered_agg_combo_main_page.match_id,
> ranked_filtered_agg_combo_main_page.offer_catalog,
> ranked_filtered_agg_combo_main_page.bet_count,
> ranked_filtered_agg_combo_main_page.combo_id,
> ROW_NUMBER() OVER (
> PARTITION BY agg_match.offer_catalog
> ORDER BY
> agg_match.bet_count DESC,
> agg_match.match_id DESC
> ) AS rank_match
> FROM
> agg_match
> INNER JOIN ranked_filtered_agg_combo_main_page ON ranked_filtered_agg_combo_main_page.match_id = agg_match.match_id
> AND ranked_filtered_agg_combo_main_page.offer_catalog = agg_match.offer_catalog
> WHERE
> ranked_filtered_agg_combo_main_page.rank_combo = 1
> )
> SELECT
> partition_key,
> match_id,
> offer_catalog,
> false as live,
> LAST_VALUE(last_event_utc) AS last_event_utc,
> LAST_VALUE(last_event_utc) AS max_last_event_utc,
> LAST_VALUE(top) AS topFROM (
> SELECT
> '<all>' as match_id,
> offer_catalog,
> '<all>' || '#openmatch#' || offer_catalog as partition_key,
> CAST('MAX(very_last_event_time_utc)' AS VARCHAR) AS last_event_utc,
> '[' || LISTAGG(
> '{"match_id":' || CAST(match_id AS VARCHAR) || ',"rank":' || CAST(rank_match AS VARCHAR) || ',"count":' || CAST(bet_count AS VARCHAR) || ',"combo_id":"' || CAST(combo_id AS VARCHAR) || '","offer_catalog":"' || CAST(offer_catalog AS VARCHAR) || '","live": ' || '}'
> ) || ']' AS top
> FROM
> joined_filtered_agg_match_main_page
> WHERE
> rank_match <= 5
> GROUP BY
> offer_catalog
> )
> GROUP BY
> partition_key,
> match_id,
> offer_catalog{code}
> As you can see in the result below, when the TTL is reached, we have duplicate values in our output.
>
>
> {code:java}
> +I[<all>#openmatch#fr, <all>, fr,
> false, MAX(very_last_event_time_utc), MAX(very_last_event_time_utc),
> [
> {
> "match_id": 2,
> "rank": 1,
> "count": 2,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> },
> {
> "match_id": 5,
> "rank": 2,
> "count": 1,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> }
> ]
> ]
> +U[<all>#openmatch#fr, <all>, fr,
> false, MAX(very_last_event_time_utc), MAX(very_last_event_time_utc),
> [
> {
> "match_id": 5,
> "rank": 2,
> "count": 1,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> },
> {
> "match_id": 2,
> "rank": 1,
> "count": 3,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> },
> {
> "match_id": 3,
> "rank": 3,
> "count": 2,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> },
> {
> "match_id": 1,
> "rank": 4,
> "count": 3,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> },
> {
> "match_id": 4,
> "rank": 5,
> "count": 1,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> }
> ]
> ]
> ...
> // DUPLICATED VALUES (multiple rank) :
> +U[<all>#openmatch#fr, <all>, fr,
> false, MAX(very_last_event_time_utc), MAX(very_last_event_time_utc),
> [
> {
> "match_id": 3,
> "rank": 1,
> "count": 16,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> },
> {
> "match_id": 5,
> "rank": 3,
> "count": 14,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> },
> {
> "match_id": 2,
> "rank": 2,
> "count": 15,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> },
> {
> "match_id": 4,
> "rank": 4,
> "count": 14,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> },
> {
> "match_id": 1,
> "rank": 5,
> "count": 12,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> },
> {
> "match_id": 2,
> "rank": 2,
> "count": 15,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> },
> {
> "match_id": 1,
> "rank": 5,
> "count": 12,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> },
> {
> "match_id": 5,
> "rank": 3,
> "count": 16,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> },
> {
> "match_id": 3,
> "rank": 1,
> "count": 18,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> },
> {
> "match_id": 4,
> "rank": 4,
> "count": 15,
> "combo_id": "123",
> "offer_catalog": "fr",
> "live":
> }
> ]
> ]
> ... {code}
>
> *Is this a bug or did we misunderstand how to implement the ttl?*
> Flink is also sending to this warning:
>
> !image-2024-04-23-15-34-32-860.png!
>
> We actually found 2 workaround:
> * set a verry large ttl (1year) This will eliminate most of the problems thanks to our business logic
> * Use sliding window (but if we do that we need window of 50days sliding each minute)
> We're afraid that these two solutions will be very expensive
> Can you tell us if this is a normal behavior or if we miss understand something about the TTL?
>
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)