/data-flow

allowed-tools: Read, Write, Bash(fd:*), Bash(rg:*), Bash(jq:*), Bash(git:*), Task
name: "Data Flow"
description: "Analyze and design data processing pipelines with automatic source detection and transformation logic"
author: "wcygan"
tags: ["analyze","data"]
version: "1.0.0"
created_at: "2025-07-14T00:00:00Z"
updated_at: "2025-07-14T00:00:00Z"

Context

• Session ID: !`gdate +%s%N`
• Current directory: !`pwd`
• Project structure: !`fd . -t d -d 3 | head -20`
• Data-related files: !`fd -e json -e csv -e sql -e py -e rs -e go -e java . | rg "(data|etl|pipeline|stream|batch)" | head -10 || echo "No data processing files detected"`
• Configuration files: !`fd "(config|\.env|docker-compose)" . -t f | head -5 || echo "No config files found"`
• Database connections: !`rg -i "(database_url|db_host|mongodb|postgres|mysql|redis)" . | head -5 || echo "No database configs found"`
• Git status: !`git status --porcelain | head -5 || echo "Not a git repository"`

Your Task

Think deeply about the optimal data flow analysis approach for this project. Consider performance, scalability, and architectural patterns.

STEP 1: Initialize analysis session

• CREATE session state file: `/tmp/data-flow-analysis-$SESSION_ID.json`
• SET initial state:

  {
    "sessionId": "$SESSION_ID",
    "phase": "discovery",
    "timestamp": "$CURRENT_TIME",
    "sources": [],
    "destinations": [],
    "transformations": [],
    "recommendations": []
  }

STEP 2: Determine analysis scope

IF $ARGUMENTS contains specific source/destination:

• FOCUS on targeted pipeline analysis
• SET scope to "targeted"

ELSE IF project size > 1000 files:

• USE sub-agent delegation for parallel discovery
• SET scope to "comprehensive"

ELSE:

• PERFORM sequential analysis
• SET scope to "standard"

STEP 3: Data source discovery

FOR comprehensive scope:

• DELEGATE to 5 parallel sub-agents:

1. **Database Discovery Agent**: Analyze all database connections and schemas 2. **File Source Agent**: Catalog structured data files (CSV, JSON, Parquet, XML) 3. **API Source Agent**: Discover REST/GraphQL endpoints and streaming APIs 4. **Log Analysis Agent**: Identify log files and extraction patterns 5. **Stream Source Agent**: Find message queues and real-time data streams

FOR standard scope:

• EXECUTE sequential discovery:
• Scan database configuration files
• Inventory structured data files
• Check for API endpoint definitions
• Identify log file patterns

STEP 4: Pipeline pattern analysis

• ANALYZE existing data processing code:
• ETL vs ELT patterns
• Batch vs stream processing
• Error handling strategies
• Monitoring and observability

• IDENTIFY transformation requirements:
• Data quality validation
• Schema transformation needs
• Aggregation patterns
• Business rule applications

STEP 5: Architecture recommendations

• EVALUATE current tech stack compatibility
• SUGGEST optimal pipeline architecture:
• Processing framework recommendations
• Destination storage strategies
• Monitoring and alerting setup
• Scalability considerations

STEP 6: Generate implementation artifacts

TRY:

• CREATE pipeline design document
• GENERATE sample transformation code
• PRODUCE monitoring configuration
• BUILD deployment scripts

CATCH (missing dependencies):

• DOCUMENT required framework installations
• SUGGEST alternative implementations
• PROVIDE fallback strategies

STEP 7: State management and cleanup

• UPDATE session state with final results
• SAVE analysis artifacts to project directory
• CHECKPOINT final recommendations
• CLEAN UP temporary session files

Sub-Agent Delegation Pattern

FOR large-scale codebases (>1000 files), delegate to parallel agents:

Database Discovery Agent

• Scan for database connection strings and configurations
• Analyze table schemas and relationships
• Identify data volume and update patterns
• Map existing database-to-database flows

File Source Agent

• Catalog all structured data files by type and location
• Sample file contents for schema inference
• Identify file naming patterns and partitioning schemes
• Assess data quality and completeness

API Source Agent

• Discover REST endpoints through OpenAPI specs or code analysis
• Analyze GraphQL schemas and query patterns
• Identify authentication and rate limiting requirements
• Map API data models and response structures

Log Analysis Agent

• Find application and system log files
• Identify log formats and parsing requirements
• Analyze log volume and retention patterns
• Suggest structured logging improvements

Stream Source Agent

• Discover message queue configurations (Kafka, RabbitMQ, etc.)
• Analyze stream schemas and partitioning strategies
• Identify real-time processing requirements
• Map event-driven architecture patterns

Data Source Discovery Patterns

Database Sources

**Relational Database Analysis**

-- PostgreSQL/MySQL schema discovery
SELECT 
    table_name, 
    column_name, 
    data_type,
    is_nullable,
    column_default
FROM information_schema.columns 
WHERE table_schema = 'public'
ORDER BY table_name, ordinal_position;

-- Table size and row count analysis
SELECT 
    schemaname,
    tablename,
    pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) as size,
    pg_stat_user_tables.n_tup_ins + pg_stat_user_tables.n_tup_upd + pg_stat_user_tables.n_tup_del as total_changes
FROM pg_tables 
JOIN pg_stat_user_tables ON pg_tables.tablename = pg_stat_user_tables.relname;

**NoSQL Database Analysis**

// MongoDB collection analysis
db.runCommand("listCollections").cursor.firstBatch.forEach(
  function (collection) {
    print("Collection: " + collection.name);
    var sample = db[collection.name].findOne();
    if (sample) {
      print("Sample document schema:");
      printjson(Object.keys(sample));
    }
  },
);

// Redis key pattern analysis
const redis = require("redis");
const client = redis.createClient