import dagster as dg from dagster_snowflake import SnowflakeResource from datetime import timedelta, datetime @dg.asset_check( asset="dim_user", blocking=True ) def user_id_has_no_nulls(snowflake: SnowflakeResource) -> dg.AssetCheckResult: """ Asset check that verifies that user_id has no null values in dim_user table. Returns True if no nulls found, False if nulls exist. """ with snowflake.get_connection() as conn: cursor = conn.cursor() # Query to check for null user_id values check_query = """ SELECT COUNT(*) as null_count FROM DIM_USER WHERE user_id IS NULL """ cursor.execute(check_query) result = cursor.fetchone() null_count = result[0] if result else 0 if null_count == 0: return dg.AssetCheckResult( passed=True, description=f"No null user_id values found in dim_user table" ) else: return dg.AssetCheckResult( passed=False, description=f"Found {null_count} null user_id values in dim_user table" ) @dg.asset_check( asset="dim_user", blocking=True ) def dim_user_has_rows(snowflake: SnowflakeResource) -> dg.AssetCheckResult: """ Asset check that verifies that dim_user table has rows (row count > 0). Returns True if table has rows, False if table is empty. """ with snowflake.get_connection() as conn: cursor = conn.cursor() # Query to count total rows in the table count_query = """ SELECT COUNT(*) as total_rows FROM DIM_USER """ cursor.execute(count_query) result = cursor.fetchone() row_count = result[0] if result else 0 if row_count > 0: return dg.AssetCheckResult( passed=True, description=f"dim_user table has {row_count} rows" ) else: return dg.AssetCheckResult( passed=False, description="dim_user table is empty (0 rows)" ) @dg.asset_check( asset="dim_user", blocking=False ) def dim_user_execution_duration_check(context: dg.AssetCheckExecutionContext) -> dg.AssetCheckResult: """ Asset check that verifies the dim_user asset execution duration is within acceptable limits. Alerts if the job took longer than the specified threshold (default: 30 minutes). """ # Get the asset key - it's available through the selected_asset_check_keys # The asset key is part of the AssetCheckKey asset_check_keys = context.selected_asset_check_keys if not asset_check_keys: return dg.AssetCheckResult( passed=False, description="No asset check keys found in context" ) # Get the asset key from the first (and only) asset check key asset_key = list(asset_check_keys)[0].asset_key # Get the latest materialization event events = context.instance.get_event_records( event_records_filter=dg.EventRecordsFilter( event_type=dg.DagsterEventType.ASSET_MATERIALIZATION, asset_key=asset_key, ), limit=1, ) if not events: return dg.AssetCheckResult( passed=False, description="No materialization record found for dim_user asset" ) event = events[0] run_id = event.event_log_entry.run_id # Find the run start event (RUN_START) and materialization event run_record = context.instance.get_run_record_by_id(run_id) if not run_record: return dg.AssetCheckResult( passed=False, description=f"Could not find run record for run_id: {run_id}" ) # Calculate duration from run start to materialization start_time = run_record.start_time end_time = event.event_log_entry.timestamp if not start_time or not end_time: return dg.AssetCheckResult( passed=False, description="Could not determine execution timestamps" ) duration = timedelta(seconds=end_time - start_time) # Set threshold (30 minutes by default) threshold_minutes = 30 threshold = timedelta(minutes=threshold_minutes) passed = duration <= threshold return dg.AssetCheckResult( passed=passed, metadata={ "duration_minutes": duration.total_seconds() / 60, "threshold_minutes": threshold_minutes, "run_id": run_id, }, description=f"dim_user execution completed in {duration.total_seconds()/60:.2f} minutes (threshold: {threshold_minutes} minutes)" ) @dg.asset_check( asset="dim_user", blocking=False ) def dim_user_volume_comparison_check( context: dg.AssetCheckExecutionContext, snowflake: SnowflakeResource ) -> dg.AssetCheckResult: """ Asset check that compares a recent complete hour's dim_user row count with the same hour from the previous day. Skips the most recent partition to avoid comparing incomplete data. Alerts if volume drops by more than 30%. """ with snowflake.get_connection() as conn: cursor = conn.cursor() # Get the most recent TWO partitions and their counts # We'll use the 2nd one to avoid incomplete current hour data recent_partitions_query = """ SELECT p_date, p_hour, COUNT(*) as row_count FROM DIM_USER GROUP BY p_date, p_hour ORDER BY p_date DESC, p_hour DESC LIMIT 2 """ cursor.execute(recent_partitions_query) results = cursor.fetchall() if len(results) < 2: return dg.AssetCheckResult( passed=True, description="Insufficient partition history in dim_user for comparison" ) # Use the 2nd most recent partition (skip potentially incomplete current hour) current_date, current_hour, current_count = results[1] # Calculate same hour from previous day current_datetime = datetime.combine(current_date, datetime.min.time()) + timedelta(hours=current_hour) yesterday_datetime = current_datetime - timedelta(days=1) yesterday_date = yesterday_datetime.date() yesterday_hour = yesterday_datetime.hour # Get row count for same hour yesterday yesterday_hour_query = """ SELECT COUNT(*) as row_count FROM dim_user WHERE p_date = %s AND p_hour = %s """ cursor.execute(yesterday_hour_query, (yesterday_date, yesterday_hour)) yesterday_result = cursor.fetchone() yesterday_count = yesterday_result[0] if yesterday_result else 0 # If yesterday's same hour had no data, we can't compare if yesterday_count == 0: return dg.AssetCheckResult( passed=True, metadata={ "current_hour_count": current_count, "yesterday_same_hour_count": yesterday_count, "current_date": str(current_date), "current_hour": current_hour, "yesterday_date": str(yesterday_date), "yesterday_hour": yesterday_hour, "note": "skipped_most_recent_partition" }, description=f"No data found for yesterday's same hour ({yesterday_date} at {yesterday_hour}:00), cannot compare volumes" ) # Calculate percentage change percent_change = ((current_count - yesterday_count) / yesterday_count) * 100 # Set threshold for acceptable decrease (30% by default) acceptable_decrease_percent = -30.0 # Check passes if volume is not significantly lower passed = percent_change >= acceptable_decrease_percent return dg.AssetCheckResult( passed=passed, metadata={ "current_hour_count": current_count, "yesterday_same_hour_count": yesterday_count, "percent_change": round(percent_change, 2), "threshold_percent": acceptable_decrease_percent, "current_date": str(current_date), "current_hour": current_hour, "yesterday_date": str(yesterday_date), "yesterday_hour": yesterday_hour, "note": "skipped_most_recent_partition" }, description=f"Latest complete hour ({current_date} at {current_hour}:00) volume: {current_count:,} vs Same hour yesterday ({yesterday_date} at {yesterday_hour}:00): {yesterday_count:,} ({percent_change:+.2f}% change, threshold: {acceptable_decrease_percent}%)" ) @dg.asset_check( asset="dim_user", blocking=False ) def dim_user_signup_country_null_percentage_check( context: dg.AssetCheckExecutionContext, snowflake: SnowflakeResource ) -> dg.AssetCheckResult: """ Asset check that monitors the count of null values in signup_country column by platform. Compares the null count from a recent complete hour with the same hour from the previous day, broken down by registered_platform. Skips the most recent partition to avoid comparing incomplete data. Alerts if ANY platform's null count increases by more than 10% compared to yesterday's same hour. """ # Get the most recent materialization events from Dagster (limit to 2 to get 2nd most recent) asset_key = dg.AssetKey("dim_user") recent_events = context.instance.get_event_records( event_records_filter=dg.EventRecordsFilter( event_type=dg.DagsterEventType.ASSET_MATERIALIZATION, asset_key=asset_key, ), limit=2, # Get the 2 most recent materializations ) if len(recent_events) < 2: return dg.AssetCheckResult( passed=True, description="Insufficient partition history in dim_user for null count comparison" ) # Use the 2nd most recent partition (skip potentially incomplete current hour) # Get partition key from the materialization event second_most_recent_event = recent_events[1].event_log_entry if not second_most_recent_event.dagster_event or not second_most_recent_event.dagster_event.partition: return dg.AssetCheckResult( passed=True, description="Could not find partition information in materialization event" ) current_partition_key = second_most_recent_event.dagster_event.partition # Parse partition key (format: YYYY-MM-DD-HH:00) # Example: "2025-10-03-14:00" partition_parts = current_partition_key.split("-") current_date = f"{partition_parts[0]}-{partition_parts[1]}-{partition_parts[2]}" current_hour = int(partition_parts[3].split(":")[0]) current_datetime = datetime.strptime(current_partition_key, "%Y-%m-%d-%H:00") # Calculate same hour from previous day yesterday_datetime = current_datetime - timedelta(days=1) yesterday_date = yesterday_datetime.strftime("%Y-%m-%d") yesterday_hour = yesterday_datetime.hour with snowflake.get_connection() as conn: cursor = conn.cursor() # Get null counts grouped by platform for current hour current_hour_query = """ SELECT COALESCE(registered_platform, 'unknown') as platform, COUNT(*) as total_count, SUM(CASE WHEN signup_country IS NULL THEN 1 ELSE 0 END) as null_count FROM dim_user WHERE p_date = %s AND p_hour = %s GROUP BY registered_platform """ cursor.execute(current_hour_query, (current_date, current_hour)) current_results = cursor.fetchall() # Build dictionary of current null counts by platform current_platform_data = {} for row in current_results: platform = row[0] total_count = row[1] null_count = row[2] current_platform_data[platform] = { 'total_count': total_count, 'null_count': null_count } # Get null counts grouped by platform for yesterday's same hour yesterday_hour_query = """ SELECT COALESCE(registered_platform, 'unknown') as platform, COUNT(*) as total_count, SUM(CASE WHEN signup_country IS NULL THEN 1 ELSE 0 END) as null_count FROM dim_user WHERE p_date = %s AND p_hour = %s GROUP BY registered_platform """ cursor.execute(yesterday_hour_query, (yesterday_date, yesterday_hour)) yesterday_results = cursor.fetchall() # Build dictionary of yesterday's null counts by platform yesterday_platform_data = {} for row in yesterday_results: platform = row[0] total_count = row[1] null_count = row[2] yesterday_platform_data[platform] = { 'total_count': total_count, 'null_count': null_count } # Get all platforms (union of current and yesterday) all_platforms = set(current_platform_data.keys()) | set(yesterday_platform_data.keys()) if not all_platforms: return dg.AssetCheckResult( passed=True, metadata={ "current_date": current_date, "current_hour": current_hour, "yesterday_date": yesterday_date, "yesterday_hour": yesterday_hour, "note": "skipped_most_recent_partition" }, description=f"No platform data found for comparison" ) # Compare each platform and track results platform_comparisons = {} failed_platforms = [] threshold_percent = 10.0 for platform in sorted(all_platforms): current_nulls = current_platform_data.get(platform, {}).get('null_count', 0) yesterday_nulls = yesterday_platform_data.get(platform, {}).get('null_count', 0) current_total = current_platform_data.get(platform, {}).get('total_count', 0) yesterday_total = yesterday_platform_data.get(platform, {}).get('total_count', 0) # Skip if no data yesterday for this platform if yesterday_nulls == 0: platform_comparisons[platform] = { 'current_nulls': current_nulls, 'current_total': current_total, 'yesterday_nulls': yesterday_nulls, 'yesterday_total': yesterday_total, 'pct_change': None, 'passed': True, 'note': 'no_baseline_data' } continue # Calculate percentage change pct_change = ((current_nulls - yesterday_nulls) / yesterday_nulls * 100) if yesterday_nulls > 0 else 0 # Check if current null count exceeds 10% threshold null_count_threshold = yesterday_nulls * (1 + threshold_percent / 100) platform_passed = current_nulls <= null_count_threshold platform_comparisons[platform] = { 'current_nulls': current_nulls, 'current_total': current_total, 'yesterday_nulls': yesterday_nulls, 'yesterday_total': yesterday_total, 'pct_change': round(pct_change, 2), 'passed': platform_passed, 'threshold_exceeded': not platform_passed } if not platform_passed: failed_platforms.append(platform) # Overall check passes only if ALL platforms pass current_check_passes = len(failed_platforms) == 0 # Calculate overall totals for summary total_current_nulls = sum(p.get('null_count', 0) for p in current_platform_data.values()) total_yesterday_nulls = sum(p.get('null_count', 0) for p in yesterday_platform_data.values()) overall_pct_change = ((total_current_nulls - total_yesterday_nulls) / total_yesterday_nulls * 100) if total_yesterday_nulls > 0 else 0 # Determine pass/fail status (simplified - no degraded period for now) passed = current_check_passes status_note = "normal_evaluation" # Build metadata with platform breakdown metadata = { "current_date": current_date, "current_hour": current_hour, "yesterday_date": yesterday_date, "yesterday_hour": yesterday_hour, "threshold_percent": threshold_percent, "total_current_nulls": total_current_nulls, "total_yesterday_nulls": total_yesterday_nulls, "overall_pct_change": round(overall_pct_change, 2), "platforms_checked": len(all_platforms), "platforms_failed": len(failed_platforms), "failed_platforms": failed_platforms if failed_platforms else None, "status_note": status_note, "note": "skipped_most_recent_partition" } # Add per-platform breakdown to metadata for platform, data in platform_comparisons.items(): # Create safe metadata key (replace special chars) safe_platform_name = platform.replace(' ', '_').replace('-', '_').lower() if platform else 'unknown' metadata[f"platform_{safe_platform_name}_current_nulls"] = data['current_nulls'] metadata[f"platform_{safe_platform_name}_yesterday_nulls"] = data['yesterday_nulls'] if data['pct_change'] is not None: metadata[f"platform_{safe_platform_name}_pct_change"] = data['pct_change'] metadata[f"platform_{safe_platform_name}_passed"] = data['passed'] # Build description if failed_platforms: failed_summary = ", ".join([f"{p} ({platform_comparisons[p]['pct_change']:+.2f}%)" for p in failed_platforms]) description = f"signup_country nulls by platform - {len(failed_platforms)} platform(s) FAILED: {failed_summary} | Overall: {total_current_nulls:,} nulls vs yesterday: {total_yesterday_nulls:,} nulls ({overall_pct_change:+.2f}% change, threshold: +{threshold_percent}%)" return dg.AssetCheckResult( passed=passed, severity=dg.AssetCheckSeverity.ERROR, metadata=metadata, description=description ) else: description = f"signup_country nulls by platform - All {len(all_platforms)} platform(s) PASSED | Overall: {total_current_nulls:,} nulls vs yesterday: {total_yesterday_nulls:,} nulls ({overall_pct_change:+.2f}% change, threshold: +{threshold_percent}%)" return dg.AssetCheckResult( passed=passed, metadata=metadata, description=description ) asset_checks = [ user_id_has_no_nulls, dim_user_has_rows, dim_user_execution_duration_check, dim_user_volume_comparison_check, dim_user_signup_country_null_percentage_check, ] __all__ = ["asset_checks"]