🤖 AI Applications in Data Engineering

import pandas as pd
from sklearn.ensemble import IsolationForest

def detect_data_anomalies(df, contamination=0.1):
    """
    Detect anomalous records using Isolation Forest
    """
    # Select numeric columns
    numeric_cols = df.select_dtypes(include=['float64', 'int64']).columns
    X = df[numeric_cols]
    
    # Train Isolation Forest
    iso_forest = IsolationForest(
        contamination=contamination, 
        random_state=42
    )
    anomalies = iso_forest.fit_predict(X)
    
    # Mark anomalies
    df['is_anomaly'] = anomalies == -1
    return df[df['is_anomaly']]

# Usage
anomalous_records = detect_data_anomalies(sales_data)
print(f"Found {len(anomalous_records)} anomalous records")

from great_expectations import dataset

# Use machine learning to learn expected patterns
def create_adaptive_expectations(df_historical):
    """
    Learn expected patterns from historical data
    """
    dataset_obj = dataset.PandasDataset(df_historical)
    
    # Stats from historical data
    mean_salary = df_historical['salary'].mean()
    std_salary = df_historical['salary'].std()
    
    # Create dynamic expectations
    expectations = {
        'salary_range': (mean_salary - 3*std_salary, 
                        mean_salary + 3*std_salary),
        'expected_columns': list(df_historical.columns)
    }
    
    return expectations

from difflib import SequenceMatcher
import numpy as np

def find_column_mappings(source_schema, target_schema):
    """
    Use similarity matching to suggest column mappings
    """
    mappings = {}
    
    for source_col in source_schema:
        max_similarity = 0
        best_match = None
        
        for target_col in target_schema:
            # Calculate similarity score
            similarity = SequenceMatcher(
                None, 
                source_col.lower(), 
                target_col.lower()
            ).ratio()
            
            if similarity > max_similarity:
                max_similarity = similarity
                best_match = target_col
        
        if max_similarity > 0.6:  # Confidence threshold
            mappings[source_col] = {
                'target': best_match,
                'confidence': max_similarity
            }
    
    return mappings

# Usage
source = ['customer_id', 'cust_email', 'created_at']
target = ['customer_id', 'email', 'creation_date', 'status']
mappings = find_column_mappings(source, target)

from sklearn.linear_model import LinearRegression

def predict_pipeline_duration(historical_data):
    """
    Predict how long a data pipeline will take
    """
    # Features: input rows, transformations, joins
    X = historical_data[['input_rows', 'transformations', 'joins']]
    y = historical_data['duration_minutes']
    
    model = LinearRegression()
    model.fit(X, y)
    
    return model

# Usage
model = predict_pipeline_duration(pipeline_metrics)
new_pipeline = [[1000000, 5, 2]]  # 1M rows, 5 transforms, 2 joins
predicted_time = model.predict(new_pipeline)[0]
print(f"Predicted duration: {predicted_time:.0f} minutes")

import re
from typing import Dict, List

def extract_metadata_ai(table_name: str, sample_data: pd.DataFrame) -> Dict:
    """
    Automatically extract metadata using data profiling
    """
    metadata = {
        'table_name': table_name,
        'row_count': len(sample_data),
        'column_info': []
    }
    
    for col in sample_data.columns:
        col_metadata = {
            'name': col,
            'data_type': str(sample_data[col].dtype),
            'null_percentage': sample_data[col].isnull().sum() / len(sample_data),
            'unique_values': sample_data[col].nunique(),
            'estimated_cardinality': sample_data[col].nunique() / len(sample_data)
        }
        
        # Infer column category
        if 'id' in col.lower() or 'code' in col.lower():
            col_metadata['likely_purpose'] = 'Identifier'
        elif 'date' in col.lower() or 'time' in col.lower():
            col_metadata['likely_purpose'] = 'Temporal'
        elif col_metadata['unique_values'] < 50:
            col_metadata['likely_purpose'] = 'Dimension/Category'
        else:
            col_metadata['likely_purpose'] = 'Measure/Fact'
        
        metadata['column_info'].append(col_metadata)
    
    return metadata

# Usage
metadata = extract_metadata_ai('sales_transactions', sales_df)
print(f"Table: {metadata['table_name']}")
for col_info in metadata['column_info']:
    print(f"  - {col_info['name']}: {col_info['likely_purpose']}")

# Using Claude or similar LLM API
def generate_table_documentation(metadata: Dict) -> str:
    """
    Generate human-readable documentation using LLM
    """
    # This would call an LLM API
    # Example output: Detailed description of table purpose and relationships
    
    prompt = f"""
    Generate a concise data dictionary for this table:
    - Name: {metadata['table_name']}
    - Columns: {[c['name'] + ' (' + c['likely_purpose'] + ')' for c in metadata['column_info']]}
    
    Provide:
    1. Table purpose
    2. Column descriptions
    3. Suggested use cases
    """
    
    # Call LLM API (pseudo-code)
    # documentation = llm_client.generate(prompt)
    return "Generated documentation..."

def suggest_query_optimization(query: str, execution_stats: Dict) -> List[str]:
    """
    Analyze query patterns and suggest optimizations
    """
    suggestions = []
    
    # Check for missing indexes
    if execution_stats['full_table_scans'] > 0:
        suggestions.append("Add index on frequently scanned columns")
    
    # Check for inefficient joins
    if execution_stats['join_complexity'] > 3:
        suggestions.append("Consider materializing intermediate results")
    
    # Check for subquery inefficiencies
    if 'SELECT' in query and query.count('SELECT') > 2:
        suggestions.append("Consolidate multiple subqueries using CTEs")
    
    # Check for aggregation patterns
    if 'GROUP BY' in query and execution_stats['rows_processed'] > 10_000_000:
        suggestions.append("Consider pre-aggregating in separate table")
    
    return suggestions

# Usage
stats = {'full_table_scans': 2, 'join_complexity': 4, 'rows_processed': 50_000_000}
suggestions = suggest_query_optimization(query, stats)
for suggestion in suggestions:
    print(f"💡 {suggestion}")

def determine_optimal_partitioning(df: pd.DataFrame, target_partition_size_mb: int = 100):
    """
    Use ML to determine optimal partitioning strategy
    """
    df_size_mb = df.memory_usage(deep=True).sum() / (1024**2)
    
    # Calculate partition column
    numeric_cols = df.select_dtypes(include=['int64','float64']).columns
    
    best_partition_col = None
    best_entropy = -1
    
    for col in numeric_cols:
        # Calculate entropy as proxy for data distribution
        value_counts = df[col].value_counts()
        entropy = -sum((value_counts / len(df)) * np.log2(value_counts / len(df) + 1e-10))
        
        if entropy > best_entropy:
            best_entropy = entropy
            best_partition_col = col
    
    return {
        'partition_column': best_partition_col,
        'estimated_partitions': int(np.ceil(df_size_mb / target_partition_size_mb)),
        'score': best_entropy
    }

from sklearn.ensemble import IsolationForest

def detect_pipeline_anomalies(metrics_history: pd.DataFrame):
    """
    Detect anomalous behavior in pipeline execution metrics
    """
    # Prepare features: duration, error_rate, data_volume
    features = metrics_history[['duration_seconds', 'error_count', 'row_count']]
    
    # Normalize features
    from sklearn.preprocessing import StandardScaler
    scaler = StandardScaler()
    X_scaled = scaler.fit_transform(features)
    
    # Train Isolation Forest
    model = IsolationForest(contamination=0.05, random_state=42)
    anomalies = model.fit_predict(X_scaled)
    
    # Identify anomalous runs
    anomalous_runs = metrics_history[anomalies == -1]
    
    return {
        'anomalies_detected': len(anomalous_runs),
        'anomalous_runs': anomalous_runs,
        'alerts': generate_alerts(anomalous_runs)
    }

def generate_alerts(anomalous_runs: pd.DataFrame) -> List[str]:
    alerts = []
    for idx, run in anomalous_runs.iterrows():
        if run['duration_seconds'] > 300:
            alerts.append(f"Pipeline {run['pipeline_id']} took {run['duration_seconds']}s (expected ~100s)")
        if run['error_count'] > 0:
            alerts.append(f"Pipeline {run['pipeline_id']} had {run['error_count']} errors")
    return alerts

# Pseudo-code for modern data engineering with AI
class SmartDataPipeline:
    def __init__(self):
        self.quality_monitor = GreatExpectations()
        self.optimizer = QueryOptimizer()
        self.catalog = MetadataService()
    
    def execute(self, data_source):
        # 1. Profile and validate data
        quality_report = self.quality_monitor.validate(data_source)
        
        # 2. Optimize query execution
        optimized_query = self.optimizer.optimize(source_query)
        
        # 3. Auto-generate metadata
        metadata = self.catalog.extract_metadata(data_source)
        
        # 4. Detect anomalies
        anomalies = self.detect_anomalies(data_source)
        
        return {
            'data': data_source,
            'quality': quality_report,
            'metadata': metadata,
            'alerts': anomalies
        }