Ingesting Into Data Lake

from awsglue.data_quality import DataQualityEvaluationOptions, DataQualityEvaluator
from awsglue.dynamicframe import DynamicFrame

# Define data quality rules
rules = """
    Rules = [
        RowCount > 0,
        ColumnCount == <expected_count>,
        ColumnValues "<column_name>" Completeness > 0.95,
        ColumnValues "<numeric_column>" between <min> and <max>,
        IsPrimaryKey "<id_column>",
        Uniqueness "<id_column>" > 0.99
    ]
"""

# Evaluate data quality
evaluator = DataQualityEvaluator(
    glueContext,
    rules,
    DynamicFrame.fromDF(transformed_df, glueContext, "check")
)
result = evaluator.evaluate()

# Fail job if quality checks don't pass
if result.overallResult != "PASS":
    raise Exception(f"Data quality check failed: {result}")

from awsglue.data_quality import DataQualityEvaluationOptions, DataQualityEvaluator
from awsglue.dynamicframe import DynamicFrame

# Define comprehensive data quality rules
rules = """
    Rules = [
        # Basic structure checks
        RowCount > 100,
        ColumnCount == 8,

        # Completeness checks
        IsComplete "customer_id",
        IsComplete "order_date",
        ColumnValues "email" Completeness > 0.90,

        # Uniqueness checks
        IsPrimaryKey "order_id",
        Uniqueness "customer_id" > 0.80,

        # Value range checks
        ColumnValues "quantity" between 1 and 1000,
        ColumnValues "price" between 0.01 and 10000.00,
        ColumnValues "order_date" >= "2023-01-01",

        # Statistical checks
        Mean "price" between 10.0 and 500.0,
        StandardDeviation "quantity" < 100.0,

        # Categorical checks
        ColumnValues "status" in ["pending", "completed", "cancelled"],
        DistinctValuesCount "status" == 3
    ]
"""

# Convert DataFrame to DynamicFrame for evaluation
dynamic_frame = DynamicFrame.fromDF(transformed_df, glueContext, "quality_check")

# Create evaluation options
eval_options = DataQualityEvaluationOptions(
    publishCloudWatchMetrics=True,
    publishResultsToCloudWatch=True
)

# Evaluate data quality
evaluator = DataQualityEvaluator(glueContext, rules, dynamic_frame, eval_options)
result = evaluator.evaluate()

# Check results
if result.overallResult != "PASS":
    # Log failed rules
    for rule_result in result.ruleResults:
        if rule_result.result == "FAIL":
            print(f"Failed rule: {rule_result.rule}")
            print(f"Failure reason: {rule_result.failureReason}")

    # Fail the job
    raise Exception(f"Data quality check failed: {result.overallResult}")
else:
    print("All data quality checks passed!")

# Define critical vs warning rules
critical_rules = """
    Rules = [
        IsPrimaryKey "order_id",
        IsComplete "customer_id",
        RowCount > 0
    ]
"""

warning_rules = """
    Rules = [
        ColumnValues "email" Completeness > 0.90,
        Mean "price" between 10.0 and 500.0
    ]
"""

# Evaluate critical rules (fail on failure)
critical_result = DataQualityEvaluator(glueContext, critical_rules, dynamic_frame).evaluate()
if critical_result.overallResult != "PASS":
    raise Exception(f"Critical data quality check failed")

# Evaluate warning rules (log but don't fail)
warning_result = DataQualityEvaluator(glueContext, warning_rules, dynamic_frame).evaluate()
if warning_result.overallResult != "PASS":
    print(f"Warning: Non-critical data quality issues detected")
    for rule_result in warning_result.ruleResults:
        if rule_result.result == "FAIL":
            print(f"  - {rule_result.rule}: {rule_result.failureReason}")

-- Count rows in target table
SELECT COUNT(*) as row_count
FROM "<catalog>"."<namespace>"."<table>"

-- Check null percentages for critical columns
SELECT
    COUNT(*) as total_rows,
    SUM(CASE WHEN customer_id IS NULL THEN 1 ELSE 0 END) as null_customer_id,
    SUM(CASE WHEN order_date IS NULL THEN 1 ELSE 0 END) as null_order_date,
    SUM(CASE WHEN amount IS NULL THEN 1 ELSE 0 END) as null_amount,
    -- Calculate percentages
    CAST(SUM(CASE WHEN customer_id IS NULL THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*) * 100 as pct_null_customer_id,
    CAST(SUM(CASE WHEN order_date IS NULL THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*) * 100 as pct_null_order_date,
    CAST(SUM(CASE WHEN amount IS NULL THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(*) * 100 as pct_null_amount
FROM "<catalog>"."<namespace>"."<table>"

-- Sample data to verify types
SELECT *
FROM "<catalog>"."<namespace>"."<table>"
LIMIT 100

-- Find duplicate order_ids
SELECT
    order_id,
    COUNT(*) as duplicate_count
FROM "<catalog>"."<namespace>"."<table>"
GROUP BY order_id
HAVING COUNT(*) > 1
ORDER BY duplicate_count DESC
LIMIT 100

-- Check value ranges
SELECT
    MIN(order_date) as min_date,
    MAX(order_date) as max_date,
    MIN(amount) as min_amount,
    MAX(amount) as max_amount,
    MIN(quantity) as min_quantity,
    MAX(quantity) as max_quantity
FROM "<catalog>"."<namespace>"."<table>"

-- Check distinct values in status column
SELECT
    status,
    COUNT(*) as count
FROM "<catalog>"."<namespace>"."<table>"
GROUP BY status
ORDER BY count DESC

-- Calculate basic statistics
SELECT
    COUNT(*) as total_rows,
    AVG(amount) as avg_amount,
    STDDEV(amount) as stddev_amount,
    APPROX_PERCENTILE(amount, 0.5) as median_amount,
    APPROX_PERCENTILE(amount, 0.95) as p95_amount
FROM "<catalog>"."<namespace>"."<table>"

Data Load Validation Report:
✓ Row count: 1,234,567 rows loaded
✓ Null checks:
  - customer_id: 0% null (expected: 0%)
  - order_date: 0.1% null (acceptable)
  - amount: 2.3% null (within threshold)
✓ Duplicates: No duplicate order_ids found
✓ Value ranges:
  - order_date: 2023-01-01 to 2024-12-31 (expected)
  - amount: $0.01 to $9,999.99 (valid range)
  - quantity: 1 to 500 (valid range)
✓ Categorical values:
  - status: pending (45%), completed (50%), cancelled (5%)
⚠ Warning: email column has 10% null values (target: < 5%)

Overall: PASS with warnings

# In Glue job script
if result.overallResult != "PASS":
    failure_summary = []
    for rule_result in result.ruleResults:
        if rule_result.result == "FAIL":
            failure_summary.append(f"  - {rule_result.rule}: {rule_result.failureReason}")

    error_message = "Data quality validation failed:\n" + "\n".join(failure_summary)
    print(error_message)

    # Optionally send notification or write to error table
    # Then fail the job
    raise Exception(error_message)

# Sample source data
sample_df = spark.read.format("csv").option("header", "true").load(source_path).limit(1000)

# Check structure
print(f"Row count: {sample_df.count()}")
print(f"Column count: {len(sample_df.columns)}")
print(f"Columns: {sample_df.columns}")
print(f"Schema: {sample_df.printSchema()}")

# Check for issues
null_counts = sample_df.select([
    (col(c).isNull().cast("int")).alias(c) for c in sample_df.columns
]).groupBy().sum()

print("Null counts in sample:")
null_counts.show()

# Count source rows
source_count = spark.read.format("csv").option("header", "true").load(source_path).count()

# Count target rows
target_count = spark.sql(f"SELECT COUNT(*) FROM {target_table}").collect()[0][0]

# Verify match
if source_count != target_count:
    print(f"Row count mismatch: source={source_count}, target={target_count}")
    raise Exception("Row count mismatch detected")
else:
    print(f"Row count validation passed: {target_count} rows")