AI Data Pipeline

Overview

This workflow automates the entire data pipeline: extraction from multiple sources, validation, transformation, AI-powered analysis, loading to warehouse, and automated reporting. It combines traditional ETL with AI agents for intelligent data processing.

Difficulty

Hard - Requires infrastructure setup and data engineering expertise.

Tools Required

• LangGraph: Complex workflow orchestration with state management
• Python (pandas, polars): Data manipulation and analysis
• Airflow: Workflow scheduling and monitoring
• dbt: Data transformation in warehouse
• Snowflake / BigQuery: Data warehouse
• Great Expectations: Data quality validation

Workflow Steps

Step 1: Extractor Agent

Pulls data from multiple sources.

from sqlalchemy import create_engine
import pandas as pd
from datetime import datetime, timedelta

class ExtractorAgent:
    """
    Extract data from multiple sources with incremental loading.
    """
    
    def __init__(self, sources: dict):
        self.sources = sources
        self.extracted_data = {}
    
    def extract(self, source_name: str, config: dict) -> pd.DataFrame:
        """
        Extract data from a configured source.
        
        Args:
            source_name: Name of the source
            config: Extraction configuration
            
        Returns:
            Extracted DataFrame
        """
        source = self.sources[source_name]
        
        if source["type"] == "postgresql":
            return self._extract_postgresql(source, config)
        elif source["type"] == "api":
            return self._extract_api(source, config)
        elif source["type"] == "s3":
            return self._extract_s3(source, config)
        elif source["type"] == "stripe":
            return self._extract_stripe(source, config)
        else:
            raise ValueError(f"Unknown source type: {source['type']}")
    
    def _extract_postgresql(self, source: dict, config: dict) -> pd.DataFrame:
        """Extract from PostgreSQL with incremental loading."""
        engine = create_engine(source["connection_string"])
        
        # Get last extraction timestamp
        last_extract = self._get_last_extract(source_name)
        
        query = f"""
        SELECT * FROM {config['table']}
        WHERE updated_at > '{last_extract}'
        ORDER BY updated_at ASC
        """
        
        df = pd.read_sql(query, engine)
        
        # Update last extract timestamp
        self._save_last_extract(source_name, datetime.utcnow())
        
        return df
    
    def _extract_api(self, source: dict, config: dict) -> pd.DataFrame:
        """Extract from REST API with pagination."""
        all_records = []
        page = 1
        
        while True:
            response = requests.get(
                source["base_url"] + config["endpoint"],
                params={
                    **source["auth"],
                    **config["params"],
                    "page": page,
                    "per_page": 100
                }
            )
            
            data = response.json()
            records = data.get("data", [])
            
            if not records:
                break
            
            all_records.extend(records)
            page += 1
            
            if len(records) < 100:
                break
        
        return pd.DataFrame(all_records)
    
    def extract_all(self) -> dict:
        """Extract from all configured sources."""
        for source_name, config in self.sources.items():
            print(f"Extracting from {source_name}...")
            self.extracted_data[source_name] = self.extract(source_name, config)
        
        return self.extracted_data

# Example configuration
sources = {
    "postgres_users": {
        "type": "postgresql",
        "connection_string": "postgresql://user:pass@host:5432/db"
    },
    "stripe_payments": {
        "type": "stripe",
        "api_key": "sk_live_..."
    },
    "api_events": {
        "type": "api",
        "base_url": "https://api.example.com",
        "auth": {"Authorization": "Bearer xxx"}
    }
}

extractor = ExtractorAgent(sources)
data = extractor.extract_all()

Step 2: Validator Agent

Checks data quality and completeness.

import great_expectations as gx
from great_expectations.core import ExpectationSuite

class ValidatorAgent:
    """
    Validate data quality using Great Expectations.
    """
    
    def __init__(self, context: gxDataContext):
        self.context = context
    
    def create_suite(self, name: str) -> ExpectationSuite:
        """Create a validation suite."""
        suite = self.context.suites.add(
            ExpectationSuite(name=name)
        )
        return suite
    
    def validate(self, df: pd.DataFrame, suite_name: str) -> dict:
        """
        Validate DataFrame against expectation suite.
        
        Returns:
            Validation result with pass/fail and details
        """
        # Create in-memory data asset
        batch_request = self.context.get_batch_request(
            dataframe=df,
            data_asset_name="temp_asset"
        )
        
        # Run validation
        validator = self.context.get_validator(
            batch_request=batch_request,
            expectation_suite_name=suite_name
        )
        
        results = validator.validate()
        
        return {
            "success": results["success"],
            "statistics": results["statistics"],
            "results": [
                {
                    "expectation": r["expectation_config"]["type"],
                    "success": r["success"],
                    "message": r.get("result", {}).get("element_count", "N/A")
                }
                for r in results["results"]
            ],
            "validation_time": datetime.utcnow().isoformat()
        }
    
    def common_validations(self, df: pd.DataFrame, columns: list) -> dict:
        """
        Run common data quality validations.
        """
        suite = self.create_suite(f"validation_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}")
        
        # Add common expectations
        for col in columns:
            suite.add_expectation(
                expectation_type="expect_column_values_to_not_be_null",
                kwargs={"column": col}
            )
            suite.add_expectation(
                expectation_type="expect_column_values_to_be_unique",
                kwargs={"column": col}
            )
        
        return self.validate(df, suite.name)

# Example validation
validator = ValidatorAgent(context)
validation_result = validator.validate(df, "users_validation")

"""
{
  "success": true,
  "statistics": {
    "elements_evaluated": 10000,
    "successful_expectations": 24,
    "unsuccessful_expectations": 0
  },
  "results": [
    {
      "expectation": "expect_column_values_to_not_be_null",
      "success": true,
      "message": 10000
    },
    {
      "expectation": "expect_column_values_to_be_unique",
      "success": true,
      "message": 10000
    }
  ]
}
"""

Step 3: Transformer Agent

Cleans and transforms data.

class TransformerAgent:
    """
    Transform and clean data with AI-powered decisions.
    """
    
    def transform(self, df: pd.DataFrame, config: dict) -> pd.DataFrame:
        """
        Apply transformations based on config.
        
        Args:
            df: Input DataFrame
            config: Transformation configuration
            
        Returns:
            Transformed DataFrame
        """
        df = df.copy()
        
        # Standard transformations
        for transform in config.get("transforms", []):
            df = self._apply_transform(df, transform)
        
        # AI-powered transformations
        if config.get("ai_enrichment"):
            df = self._ai_enrichment(df, config["ai_enrichment"])
        
        return df
    
    def _apply_transform(self, df: pd.DataFrame, transform: dict) -> pd.DataFrame:
        """Apply a single transformation."""
        if transform["type"] == "rename":
            df = df.rename(columns=transform["mapping"])
        
        elif transform["type"] == "drop":
            df = df.drop(columns=transform["columns"])
        
        elif transform["type"] == "fill_null":
            df[transform["column"]] = df[transform["column"]].fillna(transform["value"])
        
        elif transform["type"] == "convert_type":
            df[transform["column"]] = df[transform["column"]].astype(transform["target_type"])
        
        elif transform["type"] == "extract":
            df[transform["new_column"]] = df[transform["column"]].str.extract(transform["pattern"])
        
        elif transform["type"] == "aggregate":
            df = df.groupby(transform["group_by"]).agg(transform["aggregations"]).reset_index()
        
        return df
    
    def _ai_enrichment(self, df: pd.DataFrame, config: dict) -> pd.DataFrame:
        """
        Use AI to enrich data (e.g., categorize, sentiment, entities).
        """
        column = config["column"]
        task = config["task"]
        
        if task == "categorize":
            df[f"{column}_category"] = df[column].apply(
                lambda x: self._categorize_with_ai(x, config["categories"])
            )
        
        elif task == "sentiment":
            df[f"{column}_sentiment"] = df[column].apply(
                lambda x: self._analyze_sentiment(x)
            )
        
        elif task == "extract_entities":
            df[f"{column}_entities"] = df[column].apply(
                lambda x: self._extract_entities(x)
            )
        
        return df
    
    def _categorize_with_ai(self, text: str, categories: list) -> str:
        """Categorize text using AI."""
        prompt = f"""
        Categorize this text into one of: {', '.join(categories)}
        
        Text: {text[:200]}
        
        Return only the category name.
        """
        return call_claude(prompt).strip()
    
    def _analyze_sentiment(self, text: str) -> str:
        """Analyze sentiment of text."""
        prompt = f"""
        Analyze sentiment: positive, negative, or neutral
        
        Text: {text[:200]}
        
        Return only: positive, negative, or neutral
        """
        return call_claude(prompt).strip()

# Example transformation config
transform_config = {
    "transforms": [
        {"type": "rename", "mapping": {"usr_nm": "username"}},
        {"type": "drop", "columns": ["internal_id", "temp_field"]},
        {"type": "fill_null", "column": "email", "value": "unknown@example.com"},
        {"type": "convert_type", "column": "created_at", "target_type": "datetime64[ns]"},
        {"type": "extract", "column": "email", "new_column": "domain", "pattern": "@(.+)$"}
    ],
    "ai_enrichment": {
        "column": "product_description",
        "task": "categorize",
        "categories": ["electronics", "clothing", "home", "sports", "other"]
    }
}

transformer = TransformerAgent()
transformed_df = transformer.transform(raw_df, transform_config)

Step 4: Analyzer Agent

Generates insights and detects anomalies.

class AnalyzerAgent:
    """
    AI-powered data analysis and anomaly detection.
    """
    
    def analyze(self, df: pd.DataFrame, question: str) -> dict:
        """
        Analyze data to answer a natural language question.
        
        Args:
            df: DataFrame to analyze
            question: Natural language question
            
        Returns:
            Analysis results with insights
        """
        # Generate Python code to answer question
        code = self._generate_analysis_code(df, question)
        
        # Execute code safely
        result = self._execute_analysis(df, code)
        
        # Generate natural language insights
        insights = self._generate_insights(result, question)
        
        return {
            "question": question,
            "code": code,
            "result": result,
            "insights": insights,
            "charts": self._generate_charts(df, result)
        }
    
    def _generate_analysis_code(self, df: pd.DataFrame, question: str) -> str:
        """Generate Python code to answer the question."""
        columns = df.columns.tolist()
        dtypes = df.dtypes.to_dict()
        
        prompt = f"""
        Generate Python code to answer this question using pandas:
        
        Question: {question}
        
        Available columns: {columns}
        Data types: {dtypes}
        
        Sample data:
        {df.head(3).to_string()}
        
        Write a function that returns the answer.
        Return only the Python code.
        """
        
        return call_claude(prompt)
    
    def _execute_analysis(self, df: pd.DataFrame, code: str) -> any:
        """Safely execute analysis code."""
        # Use restricted execution environment
        local_vars = {"df": df, "pd": pd, "np": np}
        
        # Execute and return result
        exec(code, {}, local_vars)
        return local_vars.get("result")
    
    def detect_anomalies(self, df: pd.DataFrame, numeric_columns: list) -> dict:
        """
        Detect anomalies in numeric columns.
        """
        anomalies = {}
        
        for col in numeric_columns:
            # Statistical approach
            mean = df[col].mean()
            std = df[col].std()
            threshold = 3  # 3 standard deviations
            
            outlier_mask = abs(df[col] - mean) > threshold * std
            outliers = df[outlier_mask]
            
            if len(outliers) > 0:
                anomalies[col] = {
                    "count": len(outliers),
                    "percentage": len(outliers) / len(df) * 100,
                    "sample_values": outliers[col].head(5).tolist(),
                    "mean": mean,
                    "std": std
                }
        
        return anomalies
    
    def _generate_insights(self, result: any, question: str) -> list:
        """Generate natural language insights from results."""
        prompt = f"""
        Generate 3-5 key insights from this analysis:
        
        Question: {question}
        Result: {result}
        
        Format each insight as a clear, actionable statement.
        """
        
        insights = call_claude(prompt)
        return insights.split("\n")

# Example analysis
analyzer = AnalyzerAgent()
analysis = analyzer.analyze(sales_df, "What are the top 5 products by revenue this month?")

"""
{
  "question": "What are the top 5 products by revenue this month?",
  "insights": [
    "Product A leads with $125K in revenue, 23% above second place",
    "Top 5 products account for 67% of total monthly revenue",
    "Product E shows 45% growth compared to last month",
    "Electronics category dominates the top 5 with 4 products"
  ]
}
"""

Step 5: Loader Agent

Loads data to warehouse.

class LoaderAgent:
    """
    Load transformed data to data warehouse.
    """
    
    def load(self, df: pd.DataFrame, config: dict) -> dict:
        """
        Load DataFrame to warehouse.
        
        Args:
            df: DataFrame to load
            config: Load configuration
            
        Returns:
            Load results
        """
        if config["warehouse"] == "snowflake":
            return self._load_snowflake(df, config)
        elif config["warehouse"] == "bigquery":
            return self._load_bigquery(df, config)
        elif config["warehouse"] == "redshift":
            return self._load_redshift(df, config)
        else:
            raise ValueError(f"Unknown warehouse: {config['warehouse']}")
    
    def _load_snowflake(self, df: pd.DataFrame, config: dict) -> dict:
        """Load to Snowflake with merge strategy."""
        from snowflake.connector.pandas_tools import write_pandas
        
        # Convert DataFrame to appropriate types
        df = self._prepare_for_snowflake(df)
        
        # Write to stage first
        success, nchunks, nrows, _ = write_pandas(
            conn=self._get_snowflake_connection(),
            df=df,
            table_name=config["table"],
            schema=config.get("schema", "PUBLIC"),
            chunk_size=10000,
            compression='gzip',
            auto_create_table=False,
            create_temp_table=False
        )
        
        # Merge if incremental load
        if config.get("merge"):
            self._merge_snowflake(config)
        
        return {
            "success": success,
            "rows_loaded": nrows,
            "chunks": nchunks,
            "timestamp": datetime.utcnow().isoformat()
        }
    
    def _merge_snowflake(self, config: dict):
        """Execute merge statement for incremental loads."""
        merge_sql = f"""
        MERGE INTO {config['schema']}.{config['table']} AS target
        USING {config['schema']}.{config['staging_table']} AS source
        ON target.{config['key_column']} = source.{config['key_column']}
        WHEN MATCHED THEN
            UPDATE SET {self._generate_update_clause(config)}
        WHEN NOT MATCHED THEN
            INSERT ({', '.join(config['columns'])})
            VALUES ({', '.join('source.' + c for c in config['columns'])})
        """
        
        self._execute_sql(merge_sql)

# Example load config
load_config = {
    "warehouse": "snowflake",
    "schema": "RAW",
    "table": "users",
    "key_column": "id",
    "columns": ["id", "email", "name", "created_at", "updated_at"],
    "merge": True,
    "staging_table": "users_staging"
}

loader = LoaderAgent()
load_result = loader.load(transformed_df, load_config)

Step 6: Reporter Agent

Creates automated reports.

class ReporterAgent:
    """
    Generate automated reports and dashboards.
    """
    
    def generate_report(self, analysis_results: list, config: dict) -> dict:
        """
        Generate a comprehensive report.
        
        Args:
            analysis_results: List of analysis results
            config: Report configuration
            
        Returns:
            Generated report
        """
        report = {
            "title": config["title"],
            "generated_at": datetime.utcnow().isoformat(),
            "period": config.get("period", "last_7_days"),
            "executive_summary": self._generate_executive_summary(analysis_results),
            "sections": [],
            "charts": [],
            "recommendations": []
        }
        
        for result in analysis_results:
            section = {
                "title": result.get("question", "Analysis"),
                "insights": result.get("insights", []),
                "data": result.get("result"),
                "chart": result.get("charts", [])
            }
            report["sections"].append(section)
        
        # Generate recommendations
        report["recommendations"] = self._generate_recommendations(analysis_results)
        
        # Export to format
        if config["format"] == "pdf":
            report["file"] = self._export_pdf(report)
        elif config["format"] == "html":
            report["file"] = self._export_html(report)
        elif config["format"] == "notion":
            report["url"] = self._export_notion(report)
        
        return report
    
    def _generate_executive_summary(self, analysis_results: list) -> str:
        """Generate executive summary from all analyses."""
        all_insights = []
        for result in analysis_results:
            all_insights.extend(result.get("insights", []))
        
        prompt = f"""
        Write a 3-4 sentence executive summary based on these insights:
        
        {chr(10).join(all_insights)}
        
        Focus on key findings and actionable recommendations.
        """
        
        return call_claude(prompt)
    
    def _generate_recommendations(self, analysis_results: list) -> list:
        """Generate actionable recommendations."""
        prompt = f"""
        Based on these analysis results, provide 3-5 actionable recommendations:
        
        {str(analysis_results)[:2000]}
        
        Format each as: [Priority] Recommendation
        """
        
        recommendations = call_claude(prompt)
        return [r.strip() for r in recommendations.split("\n") if r.strip()]

# Example report generation
reporter = ReporterAgent()
report = reporter.generate_report(
    [sales_analysis, user_analysis, anomaly_report],
    {
        "title": "Weekly Business Report",
        "period": "2025-06-01 to 2025-06-07",
        "format": "pdf"
    }
)

# Email report
def send_report(report: dict, recipients: list):
    """Send report via email."""
    email_body = f"""
    Weekly Business Report
    
    Generated: {report['generated_at']}
    Period: {report['period']}
    
    Executive Summary:
    {report['executive_summary']}
    
    Key Recommendations:
    {chr(10).join(f"• {r}" for r in report['recommendations'])}
    
    Full report attached.
    """
    
    send_email(
        to=recipients,
        subject=f"Weekly Business Report - {report['period']}",
        body=email_body,
        attachment=report.get("file")
    )

Example Usage

# Daily pipeline execution
1. Extractor: Pull 500K rows from 5 sources (5 min)
2. Validator: Run 50+ data quality checks (1 min)
3. Transformer: Apply 20 transformations + AI enrichment (3 min)
4. Analyzer: Run 10 natural language analyses (2 min)
5. Loader: Merge 500K rows to Snowflake (2 min)
6. Reporter: Generate PDF report (1 min)

# Total: ~14 minutes, fully automated

Pros

• ✅ End-to-end automation of data pipelines
• ✅ AI-powered insights without SQL knowledge
• ✅ Automatic anomaly detection
• ✅ Consistent data quality validation
• ✅ Automated reporting saves hours weekly

Cons

• ❌ Complex infrastructure setup
• ❌ Requires data engineering expertise
• ❌ AI analysis can be slow for large datasets
• ❌ Cost of AI API calls for enrichment
• ❌ Debugging AI-generated code can be challenging

When to Use

Use this workflow when:

• You have multiple data sources to consolidate
• You need regular automated reporting
• Your team lacks SQL/data analysis skills
• You want AI-powered insights from data

Consider alternatives when:

• You have a single, simple data source
• Your team has strong data engineering capabilities
• You need real-time (sub-second) analysis
• Data volume is very small (< 10K rows)

Resources

• LangGraph Documentation
• Great Expectations
• dbt
• Airflow
• Snowflake