Migration Accelerator¶

The Migration Accelerator scans existing infrastructure, discovers data feeds, samples data, parses operational logs, and generates candidate PatternOps pipeline definitions with confidence scoring and risk assessment.

Architecture¶

graph TD
    A[Signal Sources] --> S[DefaultMigrationAccelerator]
    S --> MC[MetadataCatalogScanner]
    S --> CF[ConfigFileScanner]
    S --> SR[SchemaRegistryScanner]
    S --> LF[LogFileScanner]
    MC --> DR[DiscoveryResult]
    CF --> DR
    SR --> DR
    LF --> OP[OperationalProfile]
    DR --> PG[PipelineGenerator]
    DR --> RA[RiskAssessor]
    PG --> GP[Generated Pipelines + Confidence]
    RA --> MA[Migration Assessment]

Scanners¶

Scanner	Source Type	Discovers
`MetadataCatalogScanner`	`metadata_catalog`	Tables, schemas, ETL/streaming patterns, storage infra
`ConfigFileScanner`	`config_file`	Jobs, stages, connections, orchestrator type
`SchemaRegistryScanner`	`schema_registry`	Topics/subjects, event pipelines, CDC patterns
`LogFileScanner`	`log_file`	Schedules, error patterns, throughput, dependencies

Complete Workflow¶

Step 1: Scan Signal Sources¶

MigrationAccelerator accelerator = new DefaultMigrationAccelerator();

DiscoveryResult discovery = accelerator.scan(List.of(
    new SignalSource("metadata_catalog", "hive://metastore:9083",
        Map.of("databases", List.of("warehouse", "analytics"),
               "tables", List.of("raw_orders", "curated_orders", "orders_agg"))),
    new SignalSource("config_file", "/dags/daily_etl.py",
        Map.of("config_type", "airflow",
               "jobs", List.of(Map.of(
                   "name", "daily_sales_etl",
                   "stages", List.of("extract", "validate", "transform", "load"))))),
    new SignalSource("schema_registry", "http://registry:8081",
        Map.of("subjects", List.of(
            Map.of("name", "orders-value",
                   "fields", Map.of("order_id", "BIGINT", "amount", "DOUBLE")),
            Map.of("name", "users-value",
                   "fields", Map.of("user_id", "BIGINT", "email", "STRING")))))
));

// discovery.sources() → discovered data sources with inferred schemas
// discovery.pipelines() → discovered pipeline patterns
// discovery.infrastructure() → discovered components (catalog:hive, messaging:kafka, etc.)
// discovery.scanDuration() → elapsed milliseconds

Step 2: Sample Data¶

for (DiscoveredSource source : discovery.sources()) {
    SampleResult sample = accelerator.sample(source, 1000);

    // Inferred schema with types and nullability
    Map<String, Object> schema = sample.schema();
    // Per-field statistics
    Map<String, Object> stats = sample.statistics();

    // Each field has: null_rate, distinct_count, sample_size
    // Numeric fields also have: min, max, mean, stddev
    // String fields also have: min_length, max_length, avg_length
}

Step 3: Parse Operational Logs¶

OperationalProfile profile = accelerator.parseLogs(List.of(
    new LogSource("scheduler", "/var/log/airflow/scheduler-daily.log", "text"),
    new LogSource("application", "/var/log/spark/app.log", "text"),
    new LogSource("infrastructure", "/var/log/cluster/nodes.json", "json")
));

// profile.schedules() → ["0 * * * *", "0 0 * * *", ...]
// profile.errorPatterns() → ["NullPointerException in transformation stage", ...]
// profile.throughput() → {records_per_second=10000.0, jobs_per_hour=24.0, ...}
// profile.dependencies() → ["apache_spark", "apache_airflow", "aws_s3", ...]

Step 4: Generate Pipelines¶

List<GeneratedPipeline> pipelines = accelerator.generatePipelines(discovery);

for (GeneratedPipeline generated : pipelines) {
    Pipeline pipeline = generated.definition();
    ConfidenceScore score = generated.confidenceScore();

    System.out.printf("Pipeline: %s (%s mode)%n", pipeline.name(), pipeline.executionMode());
    System.out.printf("  Stages: %d%n", pipeline.stages().size());
    System.out.printf("  Confidence: %.0f%% (review: %s)%n",
        score.overall() * 100, score.requiresHumanReview());
    System.out.printf("  Components: %s%n", score.components());

    for (ReviewItem item : generated.humanReviewRequired()) {
        System.out.printf("  ⚠ %s: %s → %s (%.0f%%)%n",
            item.component(), item.reason(), item.suggestion(), item.confidence() * 100);
    }
}

Step 5: Assess Migration Risk¶

MigrationAssessment assessment = accelerator.assessMigration(discovery);

System.out.printf("Risk: %s%n", assessment.overallRisk());        // LOW, MEDIUM, HIGH, CRITICAL
System.out.printf("Complexity: %.2f%n", assessment.complexity());  // 0.0 to 1.0
System.out.printf("Effort: %.1f person-days%n", assessment.estimatedEffort());

System.out.println("Risks:");
assessment.risks().forEach(r -> System.out.println("  • " + r));

System.out.println("Recommendations:");
assessment.recommendations().forEach(r -> System.out.println("  • " + r));

Risk Assessment Factors¶

The RiskAssessor computes complexity from three weighted dimensions:

Dimension	Weight	Factors
Pipeline complexity	45%	Stage count, technology diversity, streaming/CDC presence
Source complexity	35%	Source count, type diversity, schema field count
Infrastructure complexity	20%	Component count, category diversity, multi-cloud

Risk level thresholds:

Level	Complexity	Additional Triggers
LOW	< 0.3	—
MEDIUM	0.3 – 0.6	—
HIGH	0.6 – 0.8	Streaming/CDC technology detected
CRITICAL	> 0.8	>10 stages, >50 sources, or >15 infra components

Confidence Scoring¶

Every generated pipeline gets scored on 4 dimensions:

Component	What It Measures
`pattern_match`	How well the discovered pattern maps to stages
`technology_mapping`	Whether the technology maps to a known execution mode
`schema_inference`	Quality of schema discovery from sources
`stage_mapping`	How confidently stage names map to StageTypes

Pipelines below the 0.5 threshold are flagged requiresHumanReview = true.

Signal Source Types¶

Type	Scanner	What It Discovers
`metadata_catalog`	MetadataCatalogScanner	Tables, schemas, ETL/streaming patterns
`config_file`	ConfigFileScanner	Airflow DAGs, Spark configs, dbt projects
`schema_registry`	SchemaRegistryScanner	Topics, subjects, CDC patterns
`log_file`	LogFileScanner	Schedules, errors, throughput, dependencies