purview-multi-criteria-dq-accelerator/01_generate_sample_data.py at master · vmisirlis010/purview-multi-criteria-dq-accelerator · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
# Fabric Notebook: 01 — Generate Sample Data (Insurance)
# ========================================================
# Creates sample insurance data for the DQ Rules Engine accelerator.
# Deploy this as a Fabric Notebook attached to your Lakehouse.
#
# Tables created:
#   - claims            (target table — the data to validate)
#   - policy_criteria   (reference table — determines which rules apply)
# ========================================================

from pyspark.sql.types import (
    StructType, StructField, StringType, DoubleType, DateType, TimestampType, IntegerType
)
from pyspark.sql.functions import lit, current_timestamp
from datetime import date

# -------------------------------------------------------
# 1. Claims table (target data to validate)
# -------------------------------------------------------

claims_schema = StructType([
    StructField("claim_id", StringType(), False),
    StructField("policy_id", StringType(), False),
    StructField("claim_amount", DoubleType(), True),
    StructField("claim_type", StringType(), True),
    StructField("premium_rate", DoubleType(), True),
    StructField("policy_number", StringType(), True),
    StructField("claim_date", DateType(), True),
    StructField("description", StringType(), True),
])

claims_data = [
    # --- Commercial EU Auto Claims ---
    ("C001", "P100", 35000.0,  "Collision",     0.09, "EU-POL-2026-00100", date(2026, 3, 1),  "Fleet vehicle collision claim"),
    ("C002", "P100", 65000.0,  "Collision",     0.11, "EU-POL-2026-00100", date(2026, 3, 5),  "Fleet vehicle major accident"),           # amount > 50K — should FAIL R001
    ("C003", "P100", 28000.0,  "Property",      0.15, "EU-POL-2026-00100", date(2026, 3, 10), "Commercial property water damage"),        # premium_rate > 0.12 — should FAIL R002

    # --- Individual US Health Claims ---
    ("C004", "P200", 4500.0,   "Medical",       0.06, "US-POL-2026-00200", date(2026, 3, 2),  "Routine surgery coverage"),
    ("C005", "P200", 12000.0,  "Medical",       0.05, "US-POL-2026-00200", date(2026, 3, 8),  "Emergency room visit — high cost"),        # amount > 10K — should FAIL R003
    ("C006", "P200", 3000.0,   "Dental",        0.04, "US-POL-2026-00200", date(2026, 3, 12), "Dental procedure — invalid type"),         # Dental not in allowed types — should FAIL R004

    # --- Commercial EU Auto — High Risk ---
    ("C007", "P300", 75000.0,  "Liability",     0.14, "EU-POL-2026-00300", date(2026, 3, 3),  "High-risk commercial liability claim"),    # FAILS: R001 (>50K), R002 (>0.12), R005 (>0.08 for high-risk)
    ("C008", "P300", 22000.0,  "Collision",     0.07, "EU-POL-2026-00300", date(2026, 3, 7),  "High-risk fleet — compliant claim"),

    # --- Individual EU Home Insurance ---
    ("C009", "P400", 120000.0, "Property",      0.045, "EU-POL-2026-00400", date(2026, 3, 4),  "House fire — full coverage claim"),
    ("C010", "P400", 85000.0,  "Property",      0.045, "EU-POL-2026-00400", date(2026, 3, 14), "Storm damage — partial coverage"),

    # --- Commercial US Life Insurance ---
    ("C011", "P500", 250000.0, "Medical",       None,  "US-POL-2026-00500", date(2026, 3, 6),  "Group life — no premium rate on file"),    # NULL premium_rate — should FAIL R006
    ("C012", "P500", 0.0,      "Medical",       0.10,  "US-POL-2026-00500", date(2026, 3, 11), "Group life — zero amount filed"),          # amount = 0 — should FAIL R007

    # --- Edge cases ---
    ("C013", "P600", 8000.0,   "Comprehensive", 0.07, None,                date(2026, 3, 9),  "Missing policy number"),                   # NULL policy_number — should FAIL R008
    ("C014", "P600", -1500.0,  "Comprehensive", 0.06, "EU-POL-2026-00600", date(2026, 3, 13), "Negative amount — data entry error"),      # negative — should FAIL R007
    ("C015", "P100", 40000.0,  "Collision",     0.09, "EU-POL-2026-00100", date(2026, 3, 15), "Fleet minor incident — all rules pass"),   # all rules should PASS
]

claims_df = spark.createDataFrame(claims_data, schema=claims_schema)

# Write to lakehouse
claims_df.write.format("delta").mode("overwrite").saveAsTable("claims")
print(f"✓ Created 'claims' table with {claims_df.count()} records")


# -------------------------------------------------------
# 2. Policy Criteria table (reference/lookup table)
# -------------------------------------------------------

criteria_schema = StructType([
    StructField("policy_id", StringType(), False),
    StructField("policy_type", StringType(), True),
    StructField("region", StringType(), True),
    StructField("product", StringType(), True),
    StructField("risk_tier", StringType(), True),
    StructField("channel", StringType(), True),
    StructField("coverage_limit", DoubleType(), True),
    StructField("effective_date", DateType(), True),
])

criteria_data = [
    ("P100", "Commercial",  "EU", "Auto",           "Medium", "Broker",  100000.0,  date(2024, 1, 1)),
    ("P200", "Individual",  "US", "Health",          "Low",    "Online",  50000.0,   date(2024, 6, 1)),
    ("P300", "Commercial",  "EU", "Auto",            "High",   "Broker",  150000.0,  date(2023, 1, 1)),
    ("P400", "Individual",  "EU", "Home",            "Low",    "Agent",   500000.0,  date(2025, 1, 1)),
    ("P500", "Commercial",  "US", "Life",            "Medium", "Broker",  1000000.0, date(2024, 3, 1)),
    ("P600", "Individual",  "EU", "Comprehensive",   "Low",    "Online",  75000.0,   date(2025, 6, 1)),
]

criteria_df = spark.createDataFrame(criteria_data, schema=criteria_schema)

criteria_df.write.format("delta").mode("overwrite").saveAsTable("policy_criteria")
print(f"✓ Created 'policy_criteria' table with {criteria_df.count()} records")


# -------------------------------------------------------
# 3. Quick preview
# -------------------------------------------------------

print("\n--- Claims (sample) ---")
spark.sql("SELECT claim_id, policy_id, claim_amount, claim_type, premium_rate FROM claims").show(5, truncate=False)

print("--- Policy Criteria ---")
spark.sql("SELECT * FROM policy_criteria").show(truncate=False)

print("\n✓ Sample data generation complete. Proceed to notebook 02.")