Data Quality & Validation

1import pandas as pd
2import pandas_profiling as pp
3
4# Load a dataset
5df = pd.read_csv('customer_data.csv')
6
7# Generate a profile report
8profile = pp.ProfileReport(df)
9
10# Save the report
11profile.to_file('customer_data_profile.html')
12
13# Alternatively, perform manual profiling
14column_stats = {
15    'column_name': [],
16    'data_type': [],
17    'count': [],
18    'null_count': [],
19    'null_percentage': [],
20    'unique_count': [],
21    'unique_percentage': [],
22    'min': [],
23    'max': [],
24    'mean': []
25}
26
27for col in df.columns:
28    column_stats['column_name'].append(col)
29    column_stats['data_type'].append(str(df[col].dtype))
30    column_stats['count'].append(df[col].count())
31    null_count = df[col].isnull().sum()
32    column_stats['null_count'].append(null_count)
33    column_stats['null_percentage'].append(null_count / len(df) * 100)
34    unique_count = df[col].nunique()
35    column_stats['unique_count'].append(unique_count)
36    column_stats['unique_percentage'].append(unique_count / df[col].count() * 100)
37
38    # Only calculate statistics for numeric columns
39    if pd.api.types.is_numeric_dtype(df[col]):
40        column_stats['min'].append(df[col].min())
41        column_stats['max'].append(df[col].max())
42        column_stats['mean'].append(df[col].mean())
43    else:
44        column_stats['min'].append(None)
45        column_stats['max'].append(None)
46        column_stats['mean'].append(None)
47
48# Create a DataFrame with the profiling results
49profile_df = pd.DataFrame(column_stats)
50print(profile_df)

1-- Calculate completeness for customer data
2SELECT
3    COUNT(*) AS total_records,
4    SUM(CASE WHEN customer_id IS NOT NULL THEN 1 ELSE 0 END) AS customer_id_populated,
5    SUM(CASE WHEN customer_name IS NOT NULL THEN 1 ELSE 0 END) AS customer_name_populated,
6    SUM(CASE WHEN email IS NOT NULL THEN 1 ELSE 0 END) AS email_populated,
7    SUM(CASE WHEN phone IS NOT NULL THEN 1 ELSE 0 END) AS phone_populated,
8    SUM(CASE WHEN address IS NOT NULL THEN 1 ELSE 0 END) AS address_populated,
9
10    -- Calculate percentages
11    ROUND(SUM(CASE WHEN customer_id IS NOT NULL THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2) AS customer_id_completeness,
12    ROUND(SUM(CASE WHEN customer_name IS NOT NULL THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2) AS customer_name_completeness,
13    ROUND(SUM(CASE WHEN email IS NOT NULL THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2) AS email_completeness,
14    ROUND(SUM(CASE WHEN phone IS NOT NULL THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2) AS phone_completeness,
15    ROUND(SUM(CASE WHEN address IS NOT NULL THEN 1 ELSE 0 END) * 100.0 / COUNT(*), 2) AS address_completeness
16FROM customer_dimension;
17
18-- Calculate validity for email addresses
19SELECT
20    COUNT(*) AS total_emails,
21    SUM(CASE WHEN email LIKE '%@%.%' AND email NOT LIKE '%@%@%' THEN 1 ELSE 0 END) AS valid_emails,
22    ROUND(SUM(CASE WHEN email LIKE '%@%.%' AND email NOT LIKE '%@%@%' THEN 1 ELSE 0 END) * 100.0 /
23          COUNT(email), 2) AS email_validity_percentage
24FROM customer_dimension
25WHERE email IS NOT NULL;
26
27-- Check for duplicates
28SELECT
29    COUNT(*) AS total_records,
30    COUNT(DISTINCT customer_id) AS distinct_customer_ids,
31    ROUND((COUNT(DISTINCT customer_id) * 100.0 / COUNT(*)), 2) AS uniqueness_percentage,
32    (COUNT(*) - COUNT(DISTINCT customer_id)) AS duplicate_count
33FROM customer_dimension;

1-- SQL constraint examples for syntactic validation
2CREATE TABLE customers (
3    customer_id INTEGER PRIMARY KEY,
4    email VARCHAR(255) CHECK (email LIKE '%@%.%'),
5    age INTEGER CHECK (age BETWEEN 0 AND 120),
6    phone_number VARCHAR(20) CHECK (phone_number ~ '^\\+?[0-9]{10,15}$')
7);
8
9-- Python validation example
10def validate_customer(customer):
11    errors = []
12
13    # Email validation
14    if not re.match(r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$', customer['email']):
15        errors.append("Invalid email format")
16
17    # Age validation
18    if not (0 <= customer['age'] <= 120):
19        errors.append("Age must be between 0 and 120")
20
21    # Phone validation
22    if not re.match(r'^\\+?[0-9]{10,15}$', customer['phone_number']):
23        errors.append("Invalid phone number format")
24
25    return errors

1-- SQL constraint examples for semantic validation
2ALTER TABLE orders
3ADD CONSTRAINT order_dates_valid CHECK (shipping_date > order_date),
4ADD CONSTRAINT order_total_check CHECK (order_total =
5    (SELECT SUM(quantity * price) FROM order_items WHERE order_id = orders.order_id));
6
7-- Python semantic validation example
8def validate_order_semantics(order, order_items, product_catalog):
9    errors = []
10
11    # Shipping date must be after order date
12    if order['shipping_date'] <= order['order_date']:
13        errors.append("Shipping date must be after order date")
14
15    # All products must exist in catalog
16    for item in order_items:
17        if item['product_id'] not in product_catalog:
18            errors.append(f"Product ID {item['product_id']} does not exist in catalog")
19
20    # Order total must match sum of items
21    calculated_total = sum(item['quantity'] * item['price'] for item in order_items)
22    if abs(order['order_total'] - calculated_total) > 0.01:  # Allow for small rounding differences
23        errors.append("Order total does not match sum of items")
24
25    return errors

1-- SQL constraint examples for referential integrity
2ALTER TABLE orders
3ADD CONSTRAINT fk_customer
4FOREIGN KEY (customer_id) REFERENCES customers(customer_id);
5
6ALTER TABLE order_items
7ADD CONSTRAINT fk_order
8FOREIGN KEY (order_id) REFERENCES orders(order_id) ON DELETE CASCADE;
9
10-- Python referential integrity validation
11def validate_references(data_tables):
12    errors = []
13
14    # Check if all order customer_ids exist in customers table
15    customer_ids = set(c['customer_id'] for c in data_tables['customers'])
16    for order in data_tables['orders']:
17        if order['customer_id'] not in customer_ids:
18            errors.append(f"Order {order['order_id']} references non-existent customer {order['customer_id']}")
19
20    # Check if all order_items reference valid orders
21    order_ids = set(o['order_id'] for o in data_tables['orders'])
22    for item in data_tables['order_items']:
23        if item['order_id'] not in order_ids:
24            errors.append(f"Order item {item['item_id']} references non-existent order {item['order_id']}")
25
26    return errors

1import great_expectations as ge
2
3# Load data as a Great Expectations DataFrame
4df = ge.read_csv("customer_data.csv")
5
6# Define expectations
7df.expect_column_values_to_not_be_null("customer_id")
8df.expect_column_values_to_be_unique("email")
9df.expect_column_values_to_match_regex("email", r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\\.[a-zA-Z]{2,}$')
10df.expect_column_values_to_be_between("age", min_value=0, max_value=120)
11
12# Validate expectations
13results = df.validate()
14print(results)

1import com.amazon.deequ.{VerificationSuite, VerificationResult}
2import com.amazon.deequ.constraints.{ConstraintStatus, ConstraintResult}
3import com.amazon.deequ.checks.{Check, CheckLevel, CheckStatus}
4import org.apache.spark.sql.{SparkSession, DataFrame}
5
6val spark = SparkSession.builder().master("local").getOrCreate()
7
8// Sample dataframe
9val df: DataFrame = spark.read.parquet("customer_data.parquet")
10
11// Define validation
12val verificationResult: VerificationResult = VerificationSuite()
13  .onData(df)
14  .addCheck(
15    Check(CheckLevel.Error, "Data Quality Check")
16      .hasSize(_ >= 10)
17      .isComplete("customer_id")
18      .isUnique("email")
19      .isContainedIn("gender", Array("M", "F", "Non-binary"))
20      .isNonNegative("age")
21  )
22  .run()
23
24// Inspect verification results
25if (verificationResult.status == CheckStatus.Success) {
26  println("Data quality checks passed!")
27} else {
28  println("Data quality checks failed!")
29
30  val resultsForAllConstraints = verificationResult.checkResults
31    .flatMap { case (_, checkResult) => checkResult.constraintResults }
32
33  resultsForAllConstraints
34    .filter { _.status != ConstraintStatus.Success }
35    .foreach { result =>
36      println(s"Constraint '\${result.constraint}' failed: \${result.message.getOrElse("")}")
37    }
38}

1import pandas as pd
2import numpy as np
3import json
4import matplotlib.pyplot as plt
5from datetime import datetime
6import smtplib
7from email.mime.multipart import MIMEMultipart
8from email.mime.text import MIMEText
9
10class DataQualityPipeline:
11    """A comprehensive data quality pipeline implementation"""
12
13    def __init__(self, config_path):
14        """Initialize with configuration from a JSON file"""
15        with open(config_path, 'r') as f:
16            self.config = json.load(f)
17
18        # Store quality metrics history
19        self.metrics_history = []
20
21        # Initialize validation rules from config
22        self.rules = self.config['validation_rules']
23
24        # Initialize thresholds
25        self.thresholds = self.config['quality_thresholds']
26
27    def profile_data(self, df):
28        """Profile the data to understand its characteristics"""
29        profile = {
30            'record_count': len(df),
31            'column_stats': {},
32            'timestamp': datetime.now().isoformat()
33        }
34
35        for column in df.columns:
36            col_profile = {
37                'data_type': str(df[column].dtype),
38                'null_count': df[column].isnull().sum(),
39                'null_percentage': df[column].isnull().sum() / len(df) * 100,
40                'unique_count': df[column].nunique(),
41                'unique_percentage': df[column].nunique() / len(df) * 100 if len(df) > 0 else 0
42            }
43
44            # Add numeric statistics if applicable
45            if pd.api.types.is_numeric_dtype(df[column]):
46                col_profile.update({
47                    'min': df[column].min(),
48                    'max': df[column].max(),
49                    'mean': df[column].mean(),
50                    'std': df[column].std(),
51                    'quartiles': [
52                        df[column].quantile(0.25),
53                        df[column].quantile(0.5),
54                        df[column].quantile(0.75)
55                    ]
56                })
57
58            # Add string statistics if applicable
59            elif pd.api.types.is_string_dtype(df[column]):
60                col_profile.update({
61                    'empty_count': (df[column] == '').sum(),
62                    'max_length': df[column].str.len().max(),
63                    'min_length': df[column].str.len().min(),
64                    'avg_length': df[column].str.len().mean()
65                })
66
67            profile['column_stats'][column] = col_profile
68
69        return profile
70
71    def validate_data(self, df):
72        """Validate data against defined rules"""
73        validation_results = {
74            'passed_rules': [],
75            'failed_rules': [],
76            'validation_time': datetime.now().isoformat()
77        }
78
79        for rule in self.rules:
80            rule_result = {
81                'rule_id': rule['id'],
82                'rule_name': rule['name'],
83                'rule_type': rule['type'],
84                'description': rule['description']
85            }
86
87            # Apply the appropriate validation based on rule type
88            if rule['type'] == 'not_null':
89                column = rule['column']
90                null_count = df[column].isnull().sum()
91                rule_result['null_count'] = null_count
92                rule_result['total_count'] = len(df)
93                rule_result['null_percentage'] = (null_count / len(df) * 100) if len(df) > 0 else 0
94                rule_result['passed'] = null_count == 0
95
96            elif rule['type'] == 'unique':
97                column = rule['column']
98                duplicates = len(df) - df[column].nunique()
99                rule_result['duplicate_count'] = duplicates
100                rule_result['total_count'] = len(df)
101                rule_result['duplicate_percentage'] = (duplicates / len(df) * 100) if len(df) > 0 else 0
102                rule_result['passed'] = duplicates == 0
103
104            elif rule['type'] == 'value_range':
105                column = rule['column']
106                min_value = rule.get('min_value', float('-inf'))
107                max_value = rule.get('max_value', float('inf'))
108                out_of_range = ((df[column] < min_value) | (df[column] > max_value)).sum()
109                rule_result['out_of_range_count'] = out_of_range
110                rule_result['total_count'] = len(df)
111                rule_result['out_of_range_percentage'] = (out_of_range / len(df) * 100) if len(df) > 0 else 0
112                rule_result['passed'] = out_of_range == 0
113
114            elif rule['type'] == 'regex_pattern':
115                column = rule['column']
116                pattern = rule['pattern']
117                # Count values that don't match the pattern (excluding nulls)
118                non_matching = df[df[column].notnull()][column].str.match(pattern).value_counts().get(False, 0)
119                rule_result['non_matching_count'] = non_matching
120                rule_result['total_non_null_count'] = df[column].notnull().sum()
121                rule_result['non_matching_percentage'] = (non_matching / df[column].notnull().sum() * 100) if df[column].notnull().sum() > 0 else 0
122                rule_result['passed'] = non_matching == 0
123
124            elif rule['type'] == 'referential_integrity':
125                source_column = rule['source_column']
126                reference_data = rule.get('reference_data', [])
127                invalid_refs = df[~df[source_column].isin(reference_data)][source_column].count()
128                rule_result['invalid_references'] = invalid_refs
129                rule_result['total_count'] = len(df)
130                rule_result['invalid_percentage'] = (invalid_refs / len(df) * 100) if len(df) > 0 else 0
131                rule_result['passed'] = invalid_refs == 0
132
133            elif rule['type'] == 'custom_function':
134                # Execute custom validation function
135                # This would typically be a function reference or code that gets evaluated
136                # For simplicity, we'll just use a placeholder
137                rule_result['passed'] = True  # Placeholder
138
139            # Store the rule result
140            if rule_result['passed']:
141                validation_results['passed_rules'].append(rule_result)
142            else:
143                validation_results['failed_rules'].append(rule_result)
144
145        return validation_results
146
147    def calculate_quality_score(self, validation_results):
148        """Calculate an overall data quality score"""
149        # Simple scoring: percentage of rules that passed
150        total_rules = len(validation_results['passed_rules']) + len(validation_results['failed_rules'])
151        if total_rules == 0:
152            return 100.0  # Default perfect score if no rules
153
154        score = len(validation_results['passed_rules']) / total_rules * 100
155        return score
156
157    def identify_issues(self, df, validation_results):
158        """Identify specific records with quality issues"""
159        issue_records = {}
160
161        for rule in validation_results['failed_rules']:
162            rule_id = rule['rule_id']
163            issue_records[rule_id] = {
164                'rule_name': rule['rule_name'],
165                'issue_type': rule['rule_type'],
166                'description': rule['description'],
167                'problematic_records': []
168            }
169
170            # Find problematic records based on rule type
171            if rule['rule_type'] == 'not_null':
172                column = next(r['column'] for r in self.rules if r['id'] == rule_id)
173                problematic_indices = df[df[column].isnull()].index.tolist()
174                issue_records[rule_id]['problematic_records'] = problematic_indices
175
176            elif rule['rule_type'] == 'unique':
177                column = next(r['column'] for r in self.rules if r['id'] == rule_id)
178                # Find duplicated values
179                duplicated = df[df[column].duplicated(keep=False)]
180                problematic_indices = duplicated.index.tolist()
181                issue_records[rule_id]['problematic_records'] = problematic_indices
182
183            elif rule['rule_type'] == 'value_range':
184                column = next(r['column'] for r in self.rules if r['id'] == rule_id)
185                rule_config = next(r for r in self.rules if r['id'] == rule_id)
186                min_value = rule_config.get('min_value', float('-inf'))
187                max_value = rule_config.get('max_value', float('inf'))
188                out_of_range = df[(df[column] < min_value) | (df[column] > max_value)]
189                problematic_indices = out_of_range.index.tolist()
190                issue_records[rule_id]['problematic_records'] = problematic_indices
191
192            # Add other rule types as needed
193
194        return issue_records
195
196    def remediate_issues(self, df, issues):
197        """Apply automatic data cleansing/remediation where possible"""
198        remediated_df = df.copy()
199        remediation_report = {
200            'remediated_count': 0,
201            'details': {}
202        }
203
204        for rule_id, issue in issues.items():
205            rule_config = next(r for r in self.rules if r['id'] == rule_id)
206
207            # Only attempt remediation if the rule has a remediation strategy
208            if 'remediation' in rule_config:
209                remediation_strategy = rule_config['remediation']
210                column = rule_config.get('column', '')
211                problematic_indices = issue['problematic_records']
212
213                remediation_report['details'][rule_id] = {
214                    'rule_name': rule_config['name'],
215                    'strategy': remediation_strategy,
216                    'affected_records': len(problematic_indices)
217                }
218
219                # Apply remediation based on strategy
220                if remediation_strategy == 'impute_mean' and column:
221                    if pd.api.types.is_numeric_dtype(df[column]):
222                        mean_value = df[column].mean()
223                        remediated_df.loc[problematic_indices, column] = mean_value
224                        remediation_report['remediated_count'] += len(problematic_indices)
225
226                elif remediation_strategy == 'impute_median' and column:
227                    if pd.api.types.is_numeric_dtype(df[column]):
228                        median_value = df[column].median()
229                        remediated_df.loc[problematic_indices, column] = median_value
230                        remediation_report['remediated_count'] += len(problematic_indices)
231
232                elif remediation_strategy == 'impute_mode' and column:
233                    mode_value = df[column].mode()[0]
234                    remediated_df.loc[problematic_indices, column] = mode_value
235                    remediation_report['remediated_count'] += len(problematic_indices)
236
237                elif remediation_strategy == 'remove_duplicates' and column:
238                    # For duplicate records, keep the first occurrence
239                    duplicated = df[df[column].duplicated(keep='first')].index
240                    remediated_df = remediated_df.drop(index=duplicated)
241                    remediation_report['remediated_count'] += len(duplicated)
242
243                elif remediation_strategy == 'clip_outliers' and column:
244                    if pd.api.types.is_numeric_dtype(df[column]):
245                        rule_config = next(r for r in self.rules if r['id'] == rule_id)
246                        min_value = rule_config.get('min_value', float('-inf'))
247                        max_value = rule_config.get('max_value', float('inf'))
248
249                        # Clip values to the specified range
250                        remediated_df[column] = remediated_df[column].clip(min_value, max_value)
251                        remediation_report['remediated_count'] += len(problematic_indices)
252
253        return remediated_df, remediation_report
254
255    def monitor_quality_over_time(self, quality_score):
256        """Track data quality metrics over time"""
257        self.metrics_history.append({
258            'timestamp': datetime.now().isoformat(),
259            'quality_score': quality_score
260        })
261
262    def generate_report(self, profile, validation_results, quality_score, remediation_report=None):
263        """Generate a comprehensive data quality report"""
264        report = {
265            'timestamp': datetime.now().isoformat(),
266            'data_profile_summary': {
267                'record_count': profile['record_count'],
268                'column_count': len(profile['column_stats'])
269            },
270            'validation_summary': {
271                'total_rules': len(self.rules),
272                'passed_rules': len(validation_results['passed_rules']),
273                'failed_rules': len(validation_results['failed_rules'])
274            },
275            'quality_score': quality_score,
276            'detailed_validation': validation_results
277        }
278
279        if remediation_report:
280            report['remediation_summary'] = remediation_report
281
282        return report
283
284    def alert_if_needed(self, quality_score, report):
285        """Send alerts if quality score falls below threshold"""
286        if quality_score < self.thresholds.get('alert_threshold', 90):
287            # In a real implementation, this would send an email or trigger another notification system
288            alert_message = f"DATA QUALITY ALERT: Score {quality_score} below threshold {self.thresholds.get('alert_threshold')}"
289            print(alert_message)
290
291            # Example of email notification (not actually sending)
292            email_body = f"""
293            Data Quality Alert
294            ------------------
295            Quality Score: {quality_score}
296            Threshold: {self.thresholds.get('alert_threshold')}
297
298            Failed Rules: {report['validation_summary']['failed_rules']}
299
300            Please check the detailed report for more information.
301            """
302
303            # self.send_email_alert(email_body)  # Uncomment in real implementation
304
305    def send_email_alert(self, body):
306        """Send an email alert (placeholder implementation)"""
307        # This is a placeholder - in a real implementation, configure SMTP details
308        pass
309
310    def visualize_quality_metrics(self):
311        """Generate visualizations of data quality metrics"""
312        if not self.metrics_history:
313            return None
314
315        # Extract data for plotting
316        timestamps = [entry['timestamp'] for entry in self.metrics_history]
317        scores = [entry['quality_score'] for entry in self.metrics_history]
318
319        # Create a simple time series plot
320        plt.figure(figsize=(10, 6))
321        plt.plot(timestamps, scores, marker='o')
322        plt.title('Data Quality Score Over Time')
323        plt.xlabel('Time')
324        plt.ylabel('Quality Score (%)')
325        plt.grid(True)
326        plt.ylim(0, 100)
327        plt.xticks(rotation=45)
328        plt.tight_layout()
329
330        # Save the plot
331        plot_path = f"quality_trend_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png"
332        plt.savefig(plot_path)
333        return plot_path
334
335    def run_pipeline(self, data):
336        """Run the complete data quality pipeline"""
337        # Step 1: Profile the data
338        profile = self.profile_data(data)
339
340        # Step 2: Validate the data
341        validation_results = self.validate_data(data)
342
343        # Step 3: Calculate quality score
344        quality_score = self.calculate_quality_score(validation_results)
345
346        # Step 4: Identify specific issues
347        issues = self.identify_issues(data, validation_results)
348
349        # Step 5: Remediate issues if configured
350        remediation_report = None
351        if self.config.get('auto_remediation', False):
352            remediated_data, remediation_report = self.remediate_issues(data, issues)
353        else:
354            remediated_data = data
355
356        # Step 6: Monitor quality over time
357        self.monitor_quality_over_time(quality_score)
358
359        # Step 7: Generate report
360        report = self.generate_report(profile, validation_results, quality_score, remediation_report)
361
362        # Step 8: Send alerts if needed
363        self.alert_if_needed(quality_score, report)
364
365        # Return the results
366        return {
367            'original_data': data,
368            'remediated_data': remediated_data,
369            'profile': profile,
370            'validation_results': validation_results,
371            'quality_score': quality_score,
372            'issues': issues,
373            'report': report
374        }
375
376# Example usage
377if __name__ == "__main__":
378    # Sample configuration
379    config = {
380        "validation_rules": [
381            {
382                "id": "rule1",
383                "name": "Customer ID Not Null",
384                "type": "not_null",
385                "column": "customer_id",
386                "description": "Customer ID must not be null"
387            },
388            {
389                "id": "rule2",
390                "name": "Email Uniqueness",
391                "type": "unique",
392                "column": "email",
393                "description": "Email addresses must be unique"
394            },
395            {
396                "id": "rule3",
397                "name": "Age Range",
398                "type": "value_range",
399                "column": "age",
400                "min_value": 0,
401                "max_value": 120,
402                "description": "Age must be between 0 and 120",
403                "remediation": "clip_outliers"
404            }
405        ],
406        "quality_thresholds": {
407            "alert_threshold": 90,
408            "blocking_threshold": 80
409        },
410        "auto_remediation": True
411    }
412
413    # Save config to a temporary file
414    with open("data_quality_config.json", "w") as f:
415        json.dump(config, f)
416
417    # Sample data
418    data = {
419        "customer_id": [1, 2, None, 4, 5],
420        "email": ["[email protected]", "[email protected]", "[email protected]", "[email protected]", "[email protected]"],
421        "age": [25, 30, 150, -5, 45]
422    }
423    df = pd.DataFrame(data)
424
425    # Create and run the pipeline
426    pipeline = DataQualityPipeline("data_quality_config.json")
427    results = pipeline.run_pipeline(df)
428
429    print(f"Quality Score: {results['quality_score']:.2f}%")
430    print(f"Issues found: {len(results['issues'])}")
431
432    if config['auto_remediation']:
433        print("Data after remediation:")
434        print(results['remediated_data'])

1import pandas as pd
2import numpy as np
3import matplotlib.pyplot as plt
4import seaborn as sns
5import plotly.express as px
6import plotly.graph_objects as go
7from datetime import datetime, timedelta
8import json
9
10# Sample data for demonstration
11def generate_sample_quality_data(days=30):
12    """Generate sample data quality metrics over time"""
13    date_range = pd.date_range(end=datetime.now(), periods=days)
14
15    # Generate quality scores for different dimensions
16    np.random.seed(42)  # For reproducibility
17
18    # Overall scores gradually improving with some fluctuations
19    overall_scores = np.clip(
20        np.linspace(82, 95, days) + np.random.normal(0, 3, days),
21        0, 100
22    )
23
24    # Dimension-specific scores
25    accuracy_scores = np.clip(
26        np.linspace(78, 94, days) + np.random.normal(0, 4, days),
27        0, 100
28    )
29
30    completeness_scores = np.clip(
31        np.linspace(85, 97, days) + np.random.normal(0, 2, days),
32        0, 100
33    )
34
35    consistency_scores = np.clip(
36        np.linspace(80, 93, days) + np.random.normal(0, 5, days),
37        0, 100
38    )
39
40    timeliness_scores = np.clip(
41        np.linspace(90, 98, days) + np.random.normal(0, 1, days),
42        0, 100
43    )
44
45    # Entity-specific scores
46    customer_scores = np.clip(
47        np.linspace(75, 92, days) + np.random.normal(0, 3, days),
48        0, 100
49    )
50
51    product_scores = np.clip(
52        np.linspace(88, 96, days) + np.random.normal(0, 2, days),
53        0, 100
54    )
55
56    order_scores = np.clip(
57        np.linspace(83, 94, days) + np.random.normal(0, 4, days),
58        0, 100
59    )
60
61    # Create DataFrame
62    data = {
63        'date': date_range,
64        'overall_score': overall_scores,
65        'accuracy_score': accuracy_scores,
66        'completeness_score': completeness_scores,
67        'consistency_score': consistency_scores,
68        'timeliness_score': timeliness_scores,
69        'customer_score': customer_scores,
70        'product_score': product_scores,
71        'order_score': order_scores
72    }
73
74    return pd.DataFrame(data)
75
76def generate_issue_data():
77    """Generate sample data quality issue data"""
78    issues = [
79        {
80            'issue_id': 'DQ-2023-001',
81            'entity': 'customer',
82            'attribute': 'email',
83            'issue_type': 'validity',
84            'description': 'Invalid email format',
85            'severity': 'high',
86            'affected_records': 157,
87            'status': 'resolved',
88            'detected_date': '2023-10-05',
89            'resolved_date': '2023-10-08'
90        },
91        {
92            'issue_id': 'DQ-2023-002',
93            'entity': 'product',
94            'attribute': 'price',
95            'issue_type': 'range',
96            'description': 'Negative prices found',
97            'severity': 'critical',
98            'affected_records': 12,
99            'status': 'resolved',
100            'detected_date': '2023-10-10',
101            'resolved_date': '2023-10-10'
102        },
103        {
104            'issue_id': 'DQ-2023-003',
105            'entity': 'order',
106            'attribute': 'customer_id',
107            'issue_type': 'referential_integrity',
108            'description': 'Orders with non-existent customer IDs',
109            'severity': 'medium',
110            'affected_records': 34,
111            'status': 'in_progress',
112            'detected_date': '2023-10-20',
113            'resolved_date': None
114        },
115        {
116            'issue_id': 'DQ-2023-004',
117            'entity': 'customer',
118            'attribute': 'address',
119            'issue_type': 'completeness',
120            'description': 'Missing address information',
121            'severity': 'low',
122            'affected_records': 289,
123            'status': 'in_progress',
124            'detected_date': '2023-10-25',
125            'resolved_date': None
126        },
127        {
128            'issue_id': 'DQ-2023-005',
129            'entity': 'order',
130            'attribute': 'order_date',
131            'issue_type': 'timeliness',
132            'description': 'Orders with future dates',
133            'severity': 'medium',
134            'affected_records': 8,
135            'status': 'open',
136            'detected_date': '2023-11-01',
137            'resolved_date': None
138        }
139    ]
140
141    return pd.DataFrame(issues)
142
143def create_data_quality_dashboard(quality_data, issue_data):
144    """Create a comprehensive data quality dashboard"""
145    # Set style for matplotlib plots
146    plt.style.use('ggplot')
147
148    # Create figure for the dashboard
149    fig = plt.figure(figsize=(20, 16))
150
151    # Define grid layout
152    gs = fig.add_gridspec(3, 2)
153
154    # Plot 1: Overall Data Quality Score Trend
155    ax1 = fig.add_subplot(gs[0, :])
156    ax1.plot(quality_data['date'], quality_data['overall_score'], marker='o', linewidth=2)
157    ax1.set_title('Overall Data Quality Score Trend', fontsize=16)
158    ax1.set_ylim(0, 100)
159    ax1.axhline(y=90, color='g', linestyle='--', alpha=0.7, label='Target')
160    ax1.axhline(y=80, color='y', linestyle='--', alpha=0.7, label='Warning')
161    ax1.axhline(y=70, color='r', linestyle='--', alpha=0.7, label='Critical')
162    ax1.legend()
163    ax1.set_ylabel('Quality Score (%)')
164
165    # Plot 2: Data Quality by Dimension
166    ax2 = fig.add_subplot(gs[1, 0])
167    dimensions = ['accuracy_score', 'completeness_score', 'consistency_score', 'timeliness_score']
168    dim_names = ['Accuracy', 'Completeness', 'Consistency', 'Timeliness']
169    latest_scores = quality_data[dimensions].iloc[-1].values
170
171    ax2.bar(dim_names, latest_scores, color=sns.color_palette("viridis", len(dimensions)))
172    ax2.set_title('Current Data Quality by Dimension', fontsize=16)
173    ax2.set_ylim(0, 100)
174    ax2.axhline(y=90, color='g', linestyle='--', alpha=0.5)
175    ax2.axhline(y=80, color='y', linestyle='--', alpha=0.5)
176    ax2.axhline(y=70, color='r', linestyle='--', alpha=0.5)
177
178    for i, v in enumerate(latest_scores):
179        ax2.text(i, v + 2, f"{v:.1f}%", ha='center')
180
181    ax2.set_ylabel('Quality Score (%)')
182
183    # Plot 3: Data Quality by Entity
184    ax3 = fig.add_subplot(gs[1, 1])
185    entities = ['customer_score', 'product_score', 'order_score']
186    entity_names = ['Customer', 'Product', 'Order']
187    latest_entity_scores = quality_data[entities].iloc[-1].values
188
189    ax3.bar(entity_names, latest_entity_scores, color=sns.color_palette("magma", len(entities)))
190    ax3.set_title('Current Data Quality by Entity', fontsize=16)
191    ax3.set_ylim(0, 100)
192    ax3.axhline(y=90, color='g', linestyle='--', alpha=0.5)
193    ax3.axhline(y=80, color='y', linestyle='--', alpha=0.5)
194    ax3.axhline(y=70, color='r', linestyle='--', alpha=0.5)
195
196    for i, v in enumerate(latest_entity_scores):
197        ax3.text(i, v + 2, f"{v:.1f}%", ha='center')
198
199    ax3.set_ylabel('Quality Score (%)')
200
201    # Plot 4: Open Issues by Severity
202    ax4 = fig.add_subplot(gs[2, 0])
203    open_issues = issue_data[issue_data['status'] != 'resolved']
204    severity_counts = open_issues['severity'].value_counts()
205
206    # Define colors based on severity
207    severity_colors = {'critical': 'darkred', 'high': 'red', 'medium': 'orange', 'low': 'gold'}
208    colors = [severity_colors.get(sev, 'gray') for sev in severity_counts.index]
209
210    ax4.pie(severity_counts, labels=severity_counts.index, autopct='%1.1f%%', startangle=90, colors=colors)
211    ax4.set_title('Open Issues by Severity', fontsize=16)
212
213    # Plot 5: Issue Status Table
214    ax5 = fig.add_subplot(gs[2, 1])
215    ax5.axis('off')
216
217    # Create a table with issue data
218    issue_table_data = [
219        ['Issue ID', 'Entity', 'Type', 'Severity', 'Affected', 'Status']
220    ]
221
222    for _, issue in issue_data.iterrows():
223        issue_table_data.append([
224            issue['issue_id'],
225            issue['entity'].capitalize(),
226            issue['issue_type'].replace('_', ' ').capitalize(),
227            issue['severity'].capitalize(),
228            str(issue['affected_records']),
229            issue['status'].replace('_', ' ').capitalize()
230        ])
231
232    table = ax5.table(
233        cellText=issue_table_data,
234        loc='center',
235        cellLoc='center'
236    )
237
238    table.auto_set_font_size(False)
239    table.set_fontsize(12)
240    table.scale(1, 2)
241
242    # Style the header row
243    for i in range(len(issue_table_data[0])):
244        table[(0, i)].set_facecolor('#4472C4')
245        table[(0, i)].set_text_props(color='white')
246
247    # Style the status cells
248    for i in range(1, len(issue_table_data)):
249        status = issue_table_data[i][5]
250        if status == 'Resolved':
251            table[(i, 5)].set_facecolor('#90EE90')  # Light green
252        elif status == 'In progress':
253            table[(i, 5)].set_facecolor('#FFD700')  # Gold
254        else:
255            table[(i, 5)].set_facecolor('#FFA07A')  # Light salmon
256
257    ax5.set_title('Recent Data Quality Issues', fontsize=16)
258
259    # Add dashboard title and timestamp
260    fig.suptitle('Data Warehouse Quality Dashboard', fontsize=20)
261    fig.text(0.5, 0.01, f"Last updated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
262            ha='center', fontsize=12)
263
264    plt.tight_layout(rect=[0, 0.03, 1, 0.95])
265
266    # Save the dashboard to a file
267    dashboard_file = "data_quality_dashboard.png"
268    plt.savefig(dashboard_file, dpi=150, bbox_inches='tight')
269
270    return dashboard_file
271
272# Generate sample data
273quality_data = generate_sample_quality_data()
274issue_data = generate_issue_data()
275
276# Create and save dashboard
277dashboard_file = create_data_quality_dashboard(quality_data, issue_data)
278
279print(f"Data quality dashboard created: {dashboard_file}")