From e44b988d76e328fe60f041d7d7bb0737e86b1056 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Fri, 6 Feb 2026 22:10:51 +0300 Subject: [PATCH] feat(data): add error handling for ODS layer and improve partial data support Add materialized views to capture parsing errors from browser, location, device, and geo raw staging tables and route them to dedicated error tables in the ODS layer. Refactor DDS refresh logic to handle partial data arrivals where device and geo events may arrive independently by using a unified click_id source with LEFT JOINs. Add TRUNCATE command to prevent duplicate data accumulation in DQ summary table. --- ddl/20_ods.sql | 101 ++++++++++++++++++++++++++++++++++++++++ jobs/30_dds_refresh.sql | 31 +++++++++--- jobs/40_dm_refresh.sql | 2 + 3 files changed, 128 insertions(+), 6 deletions(-) diff --git a/ddl/20_ods.sql b/ddl/20_ods.sql index f19318b..0b2c0c6 100644 --- a/ddl/20_ods.sql +++ b/ddl/20_ods.sql @@ -65,6 +65,34 @@ SELECT FROM stg.browser_raw WHERE event_id IS NOT NULL; -- Filter NULL keys to error table +CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_browser_raw_to_ods_errors +TO ods.browser_event_errors +AS +WITH + toUUIDOrNull(JSONExtractString(raw, 'event_id')) AS event_id, + parseDateTime64BestEffortOrNull(JSONExtractString(raw, 'event_timestamp'), 6) AS event_ts, + toUUIDOrNull(JSONExtractString(raw, 'click_id')) AS click_id, + arrayFilter(x -> x != '', [ + if(event_id IS NULL, 'bad_event_id', ''), + if(event_ts IS NULL, 'bad_event_timestamp', ''), + if(click_id IS NULL, 'bad_click_id', '') + ]) AS parse_errors +SELECT + ingest_ts, + kafka_topic, + kafka_partition, + kafka_offset, + kafka_ts, + raw, + arrayStringConcat(parse_errors, ',') AS error_reason +FROM stg.browser_raw +WHERE length(parse_errors) > 0 + AND ( + event_id IS NULL + OR event_ts IS NULL + OR click_id IS NULL + ); + -- ODS: location_events CREATE TABLE IF NOT EXISTS ods.location_event ( @@ -123,6 +151,26 @@ SELECT FROM stg.location_raw WHERE event_id IS NOT NULL; +CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_location_raw_to_ods_errors +TO ods.location_event_errors +AS +WITH + toUUIDOrNull(JSONExtractString(raw, 'event_id')) AS event_id, + arrayFilter(x -> x != '', [ + if(event_id IS NULL, 'bad_event_id', '') + ]) AS parse_errors +SELECT + ingest_ts, + kafka_topic, + kafka_partition, + kafka_offset, + kafka_ts, + raw, + arrayStringConcat(parse_errors, ',') AS error_reason +FROM stg.location_raw +WHERE length(parse_errors) > 0 + AND event_id IS NULL; + -- ODS: device_events CREATE TABLE IF NOT EXISTS ods.device_by_click ( @@ -182,6 +230,31 @@ SELECT FROM stg.device_raw WHERE click_id IS NOT NULL; +CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_device_raw_to_ods_errors +TO ods.device_by_click_errors +AS +WITH + toUUIDOrNull(JSONExtractString(raw, 'click_id')) AS click_id, + toUUIDOrNull(JSONExtractString(raw, 'user_domain_id')) AS user_domain_id, + arrayFilter(x -> x != '', [ + if(click_id IS NULL, 'bad_click_id', ''), + if(user_domain_id IS NULL, 'bad_user_domain_id', '') + ]) AS parse_errors +SELECT + ingest_ts, + kafka_topic, + kafka_partition, + kafka_offset, + kafka_ts, + raw, + arrayStringConcat(parse_errors, ',') AS error_reason +FROM stg.device_raw +WHERE length(parse_errors) > 0 + AND ( + click_id IS NULL + OR user_domain_id IS NULL + ); + -- ODS: geo_events CREATE TABLE IF NOT EXISTS ods.geo_by_click ( @@ -239,3 +312,31 @@ SELECT ]) AS parse_errors FROM stg.geo_raw WHERE click_id IS NOT NULL; + +CREATE MATERIALIZED VIEW IF NOT EXISTS stg.mv_geo_raw_to_ods_errors +TO ods.geo_by_click_errors +AS +WITH + toUUIDOrNull(JSONExtractString(raw, 'click_id')) AS click_id, + toFloat64OrNull(JSONExtractString(raw, 'geo_latitude')) AS geo_latitude, + toFloat64OrNull(JSONExtractString(raw, 'geo_longitude')) AS geo_longitude, + arrayFilter(x -> x != '', [ + if(click_id IS NULL, 'bad_click_id', ''), + if(geo_latitude IS NULL, 'bad_geo_latitude', ''), + if(geo_longitude IS NULL, 'bad_geo_longitude', '') + ]) AS parse_errors +SELECT + ingest_ts, + kafka_topic, + kafka_partition, + kafka_offset, + kafka_ts, + raw, + arrayStringConcat(parse_errors, ',') AS error_reason +FROM stg.geo_raw +WHERE length(parse_errors) > 0 + AND ( + click_id IS NULL + OR geo_latitude IS NULL + OR geo_longitude IS NULL + ); diff --git a/jobs/30_dds_refresh.sql b/jobs/30_dds_refresh.sql index aa4dcb3..1597035 100644 --- a/jobs/30_dds_refresh.sql +++ b/jobs/30_dds_refresh.sql @@ -6,7 +6,7 @@ -- Strategy: full rebuild for demo; incremental for production INSERT INTO dds.click SELECT - d.click_id, + c.click_id, d.user_domain_id, d.user_custom_id, d.device_type, @@ -22,14 +22,33 @@ SELECT g.ip_address, now64(3) AS dds_update_ts, arrayFilter(x -> x != '', arrayConcat( - d.parse_errors, + ifNull(d.parse_errors, []), + if(d.click_id IS NULL, ['device_not_found'], []), if(g.click_id IS NULL, ['geo_not_found'], []), if(g.geo_country IS NULL, ['geo_country_missing'], []) )) AS ods_parse_errors FROM ( + -- Union of all click ids to keep geo-only/device-only arrivals + SELECT click_id + FROM ( + SELECT assumeNotNull(click_id) AS click_id + FROM ods.device_by_click + WHERE click_id IS NOT NULL + GROUP BY click_id + ) + UNION DISTINCT + SELECT click_id + FROM ( + SELECT assumeNotNull(click_id) AS click_id + FROM ods.geo_by_click + WHERE click_id IS NOT NULL + GROUP BY click_id + ) +) AS c +LEFT JOIN ( -- Latest device snapshot from ODS SELECT - click_id, + assumeNotNull(click_id) AS click_id, argMax(user_domain_id, src_ingest_ts) AS user_domain_id, argMax(user_custom_id, src_ingest_ts) AS user_custom_id, argMax(device_type, src_ingest_ts) AS device_type, @@ -41,11 +60,11 @@ FROM ( FROM ods.device_by_click WHERE click_id IS NOT NULL GROUP BY click_id -) AS d +) AS d ON d.click_id = c.click_id LEFT JOIN ( -- Latest geo snapshot from ODS SELECT - click_id, + assumeNotNull(click_id) AS click_id, argMax(geo_country, src_ingest_ts) AS geo_country, argMax(geo_region_name, src_ingest_ts) AS geo_region_name, argMax(geo_timezone, src_ingest_ts) AS geo_timezone, @@ -55,7 +74,7 @@ LEFT JOIN ( FROM ods.geo_by_click WHERE click_id IS NOT NULL GROUP BY click_id -) AS g ON g.click_id = d.click_id; +) AS g ON g.click_id = c.click_id; -- Refresh DDS.event (from browser + location) INSERT INTO dds.event diff --git a/jobs/40_dm_refresh.sql b/jobs/40_dm_refresh.sql index 9637286..d2f0c42 100644 --- a/jobs/40_dm_refresh.sql +++ b/jobs/40_dm_refresh.sql @@ -40,6 +40,8 @@ PARTITION BY toYYYYMM(check_date) ORDER BY (check_date, layer, table_name, check_name); -- Truncate and refill DQ summary +TRUNCATE TABLE dm.dq_summary; + INSERT INTO dm.dq_summary SELECT today() AS check_date,