"""Asset checks for ddb_clip_meta_heavy to validate data quality.""" import dagster as dg from dagster_snowflake import SnowflakeResource from src.assets.snowflake.raw_table.dynamodb.clip_meta_heavy.assets import DDB_CLIP_META_HEAVY_TABLE_NAME from src.utils.snowflake.partition_utils import parse_hourly_partition_key from src.utils.snowflake.constants import Warehouse # Constants for validation MIN_ROW_COUNT = 0 # Minimum row count per partition (0 allows empty partitions) MAX_VOLUME_CHANGE_RATE = 0.30 # Maximum allowed day-over-day volume change (10%) @dg.asset_check(asset="ddb_clip_meta_heavy", blocking=False) def check_minimum_row_count(context: dg.AssetCheckExecutionContext, snowflake: SnowflakeResource) -> dg.AssetCheckResult: """Ensure minimum row count threshold is met.""" partition_key = context.op_execution_context.partition_key p_date, p_hour = parse_hourly_partition_key(partition_key) with snowflake.get_connection() as conn: cursor = conn.cursor() # Use the appropriate warehouse cursor.execute(f"USE WAREHOUSE {Warehouse.DYNAMODB_EVENTS_SMALL.value}") query = f""" SELECT COUNT(*) as row_count FROM {DDB_CLIP_META_HEAVY_TABLE_NAME} WHERE p_date = '{p_date}' and p_hour = {p_hour} """ result = cursor.execute(query).fetchone() row_count = result[0] if result else 0 passed = row_count >= MIN_ROW_COUNT return dg.AssetCheckResult( passed=passed, metadata={ "partition_key": partition_key, "p_date": p_date, "p_hour": p_hour, "row_count": row_count, "min_threshold": MIN_ROW_COUNT, }, description=f"Row count: {row_count:,} (threshold: {MIN_ROW_COUNT:,})" ) @dg.asset_check(asset="ddb_clip_meta_heavy", blocking=False) def check_id_uniqueness(context: dg.AssetCheckExecutionContext, snowflake: SnowflakeResource) -> dg.AssetCheckResult: """Verify that ID is unique within each partition.""" partition_key = context.op_execution_context.partition_key p_date, p_hour = parse_hourly_partition_key(partition_key) with snowflake.get_connection() as conn: cursor = conn.cursor() # Use the appropriate warehouse cursor.execute(f"USE WAREHOUSE {Warehouse.DYNAMODB_EVENTS_SMALL.value}") # Check for duplicate IDs query = f""" SELECT COUNT(*) as total_rows, COUNT(DISTINCT id) as unique_ids, COUNT(*) - COUNT(DISTINCT id) as duplicate_count FROM {DDB_CLIP_META_HEAVY_TABLE_NAME} WHERE p_date = '{p_date}' and p_hour = {p_hour} """ result = cursor.execute(query).fetchone() if result: total_rows = result[0] unique_ids = result[1] duplicate_count = result[2] else: total_rows = 0 unique_ids = 0 duplicate_count = 0 passed = duplicate_count == 0 return dg.AssetCheckResult( passed=passed, metadata={ "partition_key": partition_key, "p_date": p_date, "p_hour": p_hour, "total_rows": total_rows, "unique_ids": unique_ids, "duplicate_count": duplicate_count, }, description=f"Found {duplicate_count} duplicate IDs out of {total_rows} total rows" ) @dg.asset_check(asset="ddb_clip_meta_heavy", blocking=False) def check_day_over_day_volume(context: dg.AssetCheckExecutionContext, snowflake: SnowflakeResource) -> dg.AssetCheckResult: """Check that daily data volume doesn't decrease more than 20% from previous day.""" partition_key = context.op_execution_context.partition_key p_date, p_hour = parse_hourly_partition_key(partition_key) # Maximum allowed decrease rate (20%) MAX_DECREASE_RATE = 0.20 with snowflake.get_connection() as conn: cursor = conn.cursor() # Use the appropriate warehouse cursor.execute(f"USE WAREHOUSE {Warehouse.DYNAMODB_EVENTS_SMALL.value}") # Get both today's and yesterday's count in one query using conditional aggregation query = f""" SELECT SUM(CASE WHEN p_date = '{p_date}' THEN 1 ELSE 0 END) as today_count, SUM(CASE WHEN p_date = DATEADD(day, -1, '{p_date}'::date) THEN 1 ELSE 0 END) as yesterday_count FROM {DDB_CLIP_META_HEAVY_TABLE_NAME} WHERE p_date IN ('{p_date}', DATEADD(day, -1, '{p_date}'::date)) AND p_hour = {p_hour} """ result = cursor.execute(query).fetchone() today_count = result[0] if result and result[0] else 0 yesterday_count = result[1] if result and result[1] else 0 # Calculate change if yesterday_count == 0: # If no data yesterday, skip this check return dg.AssetCheckResult( passed=True, description=f"No data for yesterday's partition ({p_date} {p_hour}:00), skipping volume comparison.", metadata={ "partition_key": partition_key, "today_count": today_count, "yesterday_count": yesterday_count, } ) # Check if volume increased or stayed same - always pass if today_count >= yesterday_count: change_rate = (today_count - yesterday_count) / yesterday_count return dg.AssetCheckResult( passed=True, metadata={ "partition_key": partition_key, "p_date": p_date, "p_hour": p_hour, "today_count": today_count, "yesterday_count": yesterday_count, "change_rate": round(change_rate, 4), "change_percentage": f"+{round(change_rate * 100, 2)}%", }, description=f"Volume increased or stable: +{round(change_rate * 100, 2)}%. Today: {today_count:,}, Yesterday: {yesterday_count:,}" ) # Volume decreased - check if decrease is within acceptable range decrease_rate = (yesterday_count - today_count) / yesterday_count passed = decrease_rate <= MAX_DECREASE_RATE return dg.AssetCheckResult( passed=passed, metadata={ "partition_key": partition_key, "p_date": p_date, "p_hour": p_hour, "today_count": today_count, "yesterday_count": yesterday_count, "decrease_rate": round(decrease_rate, 4), "decrease_percentage": f"-{round(decrease_rate * 100, 2)}%", "max_allowed_decrease": f"{MAX_DECREASE_RATE * 100}%", }, description=f"Volume decreased: -{round(decrease_rate * 100, 2)}% (threshold: -{MAX_DECREASE_RATE * 100}%). Today: {today_count:,}, Yesterday: {yesterday_count:,}" ) # Export all checks as a list asset_checks = [ check_minimum_row_count, check_id_uniqueness, check_day_over_day_volume, ] __all__ = ["asset_checks"]