Day 13: Data Processing

Week 3: Workflow Automation Expert Lesson Duration: 45 minutes Difficulty: Intermediate to Advanced

Learning Objectives

By the end of this lesson, you will:

• Master advanced data transformation techniques

• Build robust data validation systems

• Create real-time data processing pipelines

• Handle multiple data formats seamlessly

Prerequisites

• Completed Day 12: Essential Node Library

• Understanding of data structures (JSON, CSV, XML)

• Basic knowledge of data quality concepts

Lesson Overview

Data is the fuel of automation. Today you'll learn to transform, enrich, and validate data like a pro. We'll explore AgenticFlow's powerful data processing capabilities that can handle everything from simple CSV files to complex API responses.

The Data Processing Philosophy

Core Principles:

• Garbage In, Garbage Out: Always validate input data

• Transform Early: Clean data as soon as it enters your system

• Fail Fast: Catch errors before they propagate

• Document Everything: Track data lineage and transformations

🎬 Video Resources

Featured Tutorial

Web Scraping Tutorial (7:59) - Master data extraction and processing from web sources using advanced parsing techniques.

Data Processing Architecture

The 4-Layer Processing Model

Core Data Processing Techniques

1. Data Ingestion Patterns

Multi-Source Data Collection

Implementation Example:

// Multi-source configuration
{
  "sources": {
    "crm": {
      "type": "api",
      "url": "https://api.salesforce.com/v1/contacts",
      "format": "json",
      "schedule": "hourly"
    },
    "marketing": {
      "type": "csv",  
      "location": "/uploads/leads.csv",
      "delimiter": ",",
      "encoding": "utf-8"
    },
    "website": {
      "type": "webhook",
      "endpoint": "/intake/web-leads",
      "validation": "strict"
    }
  }
}

Real-Time Stream Processing

2. Advanced Validation Systems

Schema-Based Validation

{
  "customer_schema": {
    "required": ["email", "name"],
    "properties": {
      "email": {
        "type": "string",
        "format": "email",
        "maxLength": 254
      },
      "name": {
        "type": "string",
        "minLength": 2,
        "maxLength": 100,
        "pattern": "^[a-zA-Z\\s\\-']+$"
      },
      "phone": {
        "type": "string", 
        "pattern": "^\\+?[1-9]\\d{1,14}$"
      },
      "age": {
        "type": "integer",
        "minimum": 18,
        "maximum": 120
      }
    }
  }
}

Quality Score Calculation

// Data quality scoring algorithm
function calculateQualityScore(record) {
  let score = 100;
  
  // Completeness (40% weight)
  const completeness = (filledFields / totalFields) * 40;
  
  // Accuracy (30% weight) 
  const accuracy = validateFormats(record) * 30;
  
  // Consistency (20% weight)
  const consistency = checkConsistency(record) * 20;
  
  // Freshness (10% weight)
  const freshness = checkFreshness(record.updated_at) * 10;
  
  return completeness + accuracy + consistency + freshness;
}

3. Transform & Enrich Operations

Data Standardization Pipeline

Implementation Steps:

1. Email Standardization: Lowercase, trim, validate format

2. Phone Formatting: International format, area code validation

3. Address Validation: Postal service verification, geocoding

4. Name Standardization: Proper case, remove extra spaces

5. Company Enrichment: Industry lookup, size classification

6. Score Calculation: Lead scoring based on enriched data

Advanced Transformations

Date/Time Processing:

// Comprehensive date handling
{
  "transformations": {
    "parse_dates": {
      "input_formats": ["MM/DD/YYYY", "YYYY-MM-DD", "DD-MM-YYYY"],
      "output_format": "ISO 8601",
      "timezone": "UTC"
    },
    "calculate_age": {
      "from": "birth_date",
      "to": "current_age",
      "unit": "years"
    },
    "business_days": {
      "exclude_weekends": true,
      "exclude_holidays": "US",
      "calculation": "working_days_until_deadline"
    }
  }
}

Text Processing Operations:

// Advanced text transformations
{
  "text_operations": {
    "clean_text": {
      "remove_html": true,
      "normalize_unicode": true,
      "trim_whitespace": true
    },
    "extract_entities": {
      "emails": "\\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Z|a-z]{2,}\\b",
      "phones": "\\b\\d{3}-\\d{3}-\\d{4}\\b",
      "urls": "https?://[^\\s]+"
    },
    "sentiment_analysis": {
      "model": "ai_content_node",
      "output": "positive|negative|neutral|score"
    }
  }
}

Hands-On Workshop: Customer Data Platform

Project Overview: 360° Customer View System

Build a comprehensive data processing system that creates unified customer profiles from multiple sources.

Implementation Phase 1: Data Collection (15 minutes)

Step 1: Multi-Source Data Ingestion

# Webhook configuration for real-time data
webhook_endpoints:
  - name: "ecommerce_events"
    path: "/data/ecommerce"
    validation_schema: "customer_interaction"
    
  - name: "email_engagement"  
    path: "/data/email"
    validation_schema: "email_activity"
    
  - name: "support_tickets"
    path: "/data/support" 
    validation_schema: "support_interaction"

Step 2: API Data Collection

// Scheduled data collection from various APIs
const data_sources = [
  {
    name: "Shopify Orders",
    endpoint: "https://shop.myshopify.com/admin/api/2023-01/orders.json",
    schedule: "every 15 minutes",
    auth: "bearer_token"
  },
  {
    name: "Mailchimp Campaigns", 
    endpoint: "https://us1.api.mailchimp.com/3.0/campaigns",
    schedule: "hourly",
    auth: "api_key"
  }
];

Implementation Phase 2: Validation & Cleaning (15 minutes)

Advanced Validation Rules

{
  "validation_rules": {
    "email_quality": {
      "format": "valid_email",
      "domain_check": true,
      "disposable_check": true,
      "role_account_check": true
    },
    "phone_validation": {
      "format": "international",
      "carrier_lookup": true,
      "line_type": ["mobile", "landline"],
      "region_validation": true
    },
    "address_validation": {
      "postal_service": true,
      "geocoding": true,
      "standardization": "USPS"
    }
  }
}

Data Cleaning Pipeline

// Multi-stage cleaning process
const cleaning_pipeline = [
  {
    stage: "normalize",
    operations: ["trim", "lowercase_email", "title_case_name"]
  },
  {
    stage: "deduplicate", 
    criteria: ["email_fuzzy_match", "phone_exact_match"],
    merge_strategy: "most_complete_record"
  },
  {
    stage: "standardize",
    operations: ["format_phone", "validate_postal", "normalize_company"]
  }
];

Implementation Phase 3: Enrichment & Analysis (15 minutes)

External API Enrichment

Enrichment Configuration:

enrichment_apis:
  company_data:
    provider: "Clearbit"
    endpoint: "https://company.clearbit.com/v2/companies/find"
    rate_limit: "600/hour"
    fields: ["industry", "employee_count", "revenue", "technology_stack"]
    
  demographic_data:
    provider: "Experian"
    fields: ["age_range", "income_bracket", "lifestyle_segment"]
    
  social_profiles:
    provider: "FullContact" 
    fields: ["social_handles", "interests", "influence_score"]

AI-Powered Profile Analysis

// AI analysis for customer insights
{
  "ai_analysis": {
    "personality_analysis": {
      "input": "communication_history + purchase_behavior",
      "output": "personality_traits + communication_preferences"
    },
    "churn_prediction": {
      "model": "gradient_boosting",
      "features": ["engagement_score", "purchase_frequency", "support_interactions"],
      "output": "churn_probability + risk_factors"
    },
    "lifetime_value": {
      "calculation": "purchase_history + engagement_trend + demographic_factors",
      "prediction_horizon": "12_months"
    }
  }
}

Advanced Data Processing Patterns

Pattern 1: The Data Lake Architecture

Pattern 2: Real-Time Data Processing

Pattern 3: Batch Processing Optimization

Error Handling & Recovery

Robust Error Handling Strategy

{
  "error_handling": {
    "validation_errors": {
      "action": "quarantine",
      "notification": "data_team",
      "retry": false
    },
    "api_failures": {
      "action": "retry",
      "max_attempts": 3,
      "backoff": "exponential",
      "fallback": "cached_data"
    },
    "transformation_errors": {
      "action": "log_and_continue",
      "partial_processing": true,
      "alert_threshold": "10_errors_per_hour"
    }
  }
}

Data Recovery Mechanisms

Performance Optimization

Processing Speed Optimization

Parallel Processing Strategy:

// Optimal batch processing configuration
{
  "batch_config": {
    "batch_size": 1000, // Optimal for most APIs
    "parallel_workers": 4, // Based on system capacity
    "processing_timeout": 300, // 5 minutes per batch
    "memory_limit": "2GB"
  }
}

Caching Strategy:

Memory Management

Large Dataset Handling:

// Streaming processing for large files
{
  "streaming_config": {
    "chunk_size": "10MB",
    "buffer_size": "100MB", 
    "garbage_collection": "after_each_chunk",
    "temp_storage": "/tmp/processing"
  }
}

Quality Monitoring & Metrics

Data Quality Dashboard

Automated Quality Alerts

{
  "quality_thresholds": {
    "completeness": {
      "warning": 85,
      "critical": 70
    },
    "accuracy": {
      "warning": 90, 
      "critical": 80
    },
    "processing_speed": {
      "warning": "5min_delay",
      "critical": "15min_delay"
    }
  }
}

Resource Library

Essential Documentation

• Node Reference Library

• Extract Website Content

• Workflow Building Blocks

Advanced Resources

• Data validation schemas library

• Transformation pattern templates

• Error handling best practices

• Performance tuning guides

What's Next

Tomorrow (Day 14): Logic and Control Flow

• Advanced conditional logic

• Complex decision trees

• Loop optimization

• Error handling patterns

Week Progress Check

• Day 15: Integration and deployment strategies

• Week 3 Capstone: Complete marketing automation platform

Homework Challenge

Build a Data Quality Engine (60 minutes)

Create a comprehensive data processing system that:

Requirements:

1. Ingests data from at least 3 different sources

2. Validates data against custom schemas

3. Cleans and standardizes all incoming data

4. Enriches records with external API data

5. Scores data quality automatically

6. Alerts on quality issues

7. Generates quality reports

8. Handles errors gracefully

Bonus Challenges:

• Implement real-time processing

• Add data lineage tracking

• Create automated data profiling

• Build a data quality dashboard

Success Criteria:

• Processes 1000+ records without errors

• Achieves 95%+ data quality scores

• Completes full processing in under 5 minutes

• Handles various data formats seamlessly

---

Master data processing, master automation. Tomorrow we'll add intelligent logic and control flow to create truly smart automation systems.