ETL Engine

Overview

The modern ETL Engine replaces the legacy ETLEditor with a plugin-based pipeline architecture:

• Plugin-based - Sources, transformers, validators, and sinks are plugins discovered at runtime
• Streaming - Records flow through steps in configurable batches
• Observable - Metrics, alerting, and lineage tracking built-in
• Resilient - Checkpointing, retries, and error sinks for fault tolerance
• Scheduled - Cron-based scheduling with watermark tracking

Architecture

PipelineManager (persistence, CRUD, orchestration)
    |
    +-- PipelineEngine (execution)
    |       +-- PipelinePluginRegistry (discovery)
    |       +-- PipelineStepRunner (step execution)
    |       +-- PipelineCheckpointManager (fault tolerance)
    |       +-- PipelineLineageTracker (data lineage)
    |
    +-- SchedulerHost (scheduling)
    |       +-- PipelineRunQueue (queueing)
    |       +-- WatermarkTracker (incremental sync)
    |       +-- BackfillManager (backfill operations)
    |
    +-- ObservabilityStore (metrics, alerting)

PipelineDefinition

Full pipeline specification replacing ETLScriptHDR:

public class PipelineDefinition
{
    // Identity
    public string Id { get; set; }
    public string Name { get; set; }
    public string Description { get; set; }
    public string Category { get; set; }
    public string Tags { get; set; }
    public int Version { get; set; }
    public bool IsEnabled { get; set; }

    // Plugins
    public string SourcePluginId { get; set; }
    public string SinkPluginId { get; set; }
    public string ErrorSinkPluginId { get; set; }

    // Steps
    public List<PipelineStepDef> Steps { get; set; }

    // Parameters
    public Dictionary<string, object> Parameters { get; set; }

    // Scheduling
    public string? SchedulerPluginId { get; set; }
    public Dictionary<string, object> SchedulerParameters { get; set; }

    // Execution Policy
    public int BatchSize { get; set; } = 500;
    public int MaxParallelBatches { get; set; } = 4;
    public int MaxRetries { get; set; } = 3;
    public int StopOnErrorCount { get; set; } = 0;
    public bool EnableCheckpointing { get; set; } = true;
    public bool EnableLineageTracking { get; set; } = true;

    // Workload
    public string WorkloadClass { get; set; } = "standard";
    public int MemoryCapMB { get; set; }
    public double CostBudgetPerRun { get; set; }

    // Security
    public DataClassification DataClassification { get; set; }
    public List<string> SensitiveFields { get; set; }
    public MaskingStrategy MaskingStrategy { get; set; }
    public List<string> RequiredPermissions { get; set; }

    // Migration
    public PipelineTier Tier { get; set; }
    public bool IsCanaryEnabled { get; set; }
    public bool IsShadowRunEnabled { get; set; }
    public bool CompatibilityMode { get; set; }

    // History
    public DateTime? LastRunAt { get; set; }
    public string? LastRunStatus { get; set; }
}

PipelineEngine

Core pipeline runner that executes PipelineDefinitions:

public class PipelineEngine
{
    public PipelineEngine(IDMEEditor editor);

    // Observability hooks
    public ObservabilityStore? ObservabilityStore { get; set; }
    public MetricsEngine? Metrics { get; set; }
    public AlertingEngine? Alerting { get; set; }
    public SecurityPolicyEngine? SecurityPolicy { get; set; }

    // Execution
    public Task<PipelineRunResult> RunAsync(
        PipelineDefinition def,
        IProgress<PassedArgs>? progress = null,
        CancellationToken token = default,
        Dictionary<string, object>? overrideParams = null,
        SecurityContext? securityContext = null);

    // Resume from checkpoint
    public Task<PipelineRunResult> ResumeAsync(
        string checkpointId,
        IProgress<PassedArgs>? progress = null,
        CancellationToken token = default);
}

Execution Flow

1. Resolve plugins via PipelinePluginRegistry
2. Configure source, transformers, validators, sinks
3. Stream records through steps in batches
4. Write accepted records to main sink
5. Route rejected records to error sink
6. Save checkpoints after committed batches
7. Record column-level lineage

PipelineManager

Persists, loads, and runs PipelineDefinitions:

public class PipelineManager
{
    public PipelineManager(IDMEEditor editor);

    // Observability
    public ObservabilityStore? ObservabilityStore { get; set; }
    public MetricsEngine? Metrics { get; set; }
    public AlertingEngine? Alerting { get; set; }
    public SecurityPolicyEngine? SecurityPolicy { get; set; }
    public SecurityContext? SecurityContext { get; set; }

    // CRUD
    public Task<IErrorsInfo> SaveAsync(PipelineDefinition pipeline);
    public Task<IErrorsInfo> DeleteAsync(string pipelineId);
    public Task<PipelineDefinition?> LoadAsync(string pipelineId);
    public Task<IReadOnlyList<PipelineDefinition>> LoadAllAsync();
    public Task<IReadOnlyList<PipelineDefinition>> FindByTagAsync(string tag);

    // Execution
    public Task<PipelineRunResult> RunAsync(
        string pipelineId,
        IProgress<PassedArgs>? progress = null,
        CancellationToken token = default);

    public Task<PipelineRunResult> RunDefinitionAsync(
        PipelineDefinition def,
        IProgress<PassedArgs>? progress = null,
        CancellationToken token = default);

    public Task<PipelineRunResult> ResumeAsync(
        string checkpointId,
        IProgress<PassedArgs>? progress = null,
        CancellationToken token = default);
}

Storage

• Pipelines: {BeepDataPath}/Pipelines/{id}.pipeline.json
• History: {BeepDataPath}/Pipelines/History/{pipelineId}/{runId}.run.json

PipelinePluginRegistry

Discovers and instantiates pipeline plugins using the same mechanism as [AddinAttribute] connectors:

public class PipelinePluginRegistry
{
    public PipelinePluginRegistry(IDMEEditor editor);

    // Discovery
    public void Discover();

    // Instantiation
    public T Create<T>(string pluginId) where T : IPipelinePlugin;

    // Queries
    public IReadOnlyList<PipelinePluginDescriptor> GetAll();
    public IReadOnlyList<PipelinePluginDescriptor> GetByCategory(string category);
    public bool HasPlugin(string pluginId);
}

Plugin Categories

Scheduling

The SchedulerHost provides cron-based scheduling with watermarks and backfills:

public sealed class SchedulerHost : IAsyncDisposable
{
    public SchedulerHost(IDMEEditor editor);

    // State
    public ScheduleStorage Storage { get; }
    public WatermarkTracker Watermarks { get; }
    public BackfillManager Backfills { get; }
    public ScheduleAuditLog AuditLog { get; }

    // Events
    public event EventHandler<SchedulerRunEventArgs>? RunStarted;
    public event EventHandler<SchedulerRunEventArgs>? RunCompleted;
    public event EventHandler<SchedulerRunEventArgs>? RunFailed;
    public event EventHandler<SchedulerRunEventArgs>? CircuitBreakerChanged;

    // Control
    public Task StartAsync(CancellationToken token = default);
    public Task StopAsync();
    public Task ReloadAsync();

    // Manual trigger
    public Task<string> TriggerManualAsync(
        string pipelineOrWorkflowId,
        IReadOnlyDictionary<string, object>? overrideParams = null);

    // Backfill
    public Task<string> SubmitBackfillAsync(
        string scheduleId, string fromValue, string toValue,
        string reason = "", string requestedBy = "",
        int windowSizeSeconds = 86400);

    // Circuit breaker
    public Task ResetCircuitBreakerAsync(string scheduleId, string resetBy = "");

    // Monitoring
    public IReadOnlyList<(string RunId, string PipelineId, RunStatus Status)> GetActiveRuns();
}

PipelineRunQueue

public sealed class PipelineRunQueue
{
    public int Count { get; }
    public int CountByClass(string workloadClass);
    public void SetClassConcurrencyLimit(string workloadClass, int maxConcurrent);
    public void Enqueue(QueuedRun run);
    public Task<QueuedRun> DequeueAsync(CancellationToken token);
    public void NotifyRunCompleted(string workloadClass);
}

WatermarkTracker

public sealed class WatermarkTracker
{
    public Task<WatermarkState?> LoadAsync(string scheduleId);
    public Task<IErrorsInfo> SaveAsync(string scheduleId, WatermarkState state);
    public WatermarkWindow ComputeWindow(WatermarkConfig config, WatermarkState? currentState);
    public Task<IErrorsInfo> ResetAsync(string scheduleId);
}

Observability

Built-in metrics, alerting, and lineage tracking:

public class MetricsEngine
{
    public void RecordCounter(string metricName, long value, Dictionary<string, string>? tags = null);
    public void RecordGauge(string metricName, double value, Dictionary<string, string>? tags = null);
    public void RecordHistogram(string metricName, double value, Dictionary<string, string>? tags = null);
    public void RecordTimer(string metricName, TimeSpan duration, Dictionary<string, string>? tags = null);
    public IReadOnlyList<MetricSnapshot> GetSnapshots();
}

public class AlertingEngine
{
    public void RegisterRule(AlertRule rule);
    public void Evaluate(PipelineRunResult result);
    public IReadOnlyList<Alert> GetActiveAlerts();
    public void AcknowledgeAlert(string alertId, string acknowledgedBy);
}

public class PipelineLineageTracker
{
    public void RecordFieldLineage(string runId, string fieldName, string source, string transformation);
    public void RecordRecordLineage(string runId, string recordId, List<string> sourceIds);
    public LineageReport GetReport(string runId);
}

Checkpointing

Fault-tolerant execution with resume capability:

public class PipelineCheckpointManager
{
    public Task<Checkpoint> CreateAsync(string runId, long recordsProcessed);
    public Task<Checkpoint?> LoadAsync(string checkpointId);
    public Task SaveAsync(Checkpoint checkpoint);
    public Task<IReadOnlyList<Checkpoint>> GetHistoryAsync(string pipelineId);
    public Task PurgeAsync(string pipelineId, TimeSpan olderThan);
}

Checkpoint Data

Usage Examples

Creating a Pipeline

var pipeline = new PipelineDefinition
{
    Name = "DailyCustomerSync",
    Description = "Sync customers from CRM to Data Warehouse",
    SourcePluginId = "crm-db-source",
    SinkPluginId = "warehouse-db-sink",
    ErrorSinkPluginId = "error-log-sink",
    BatchSize = 1000,
    MaxParallelBatches = 2,
    EnableCheckpointing = true,
    Steps = new List<PipelineStepDef>
    {
        new PipelineStepDef
        {
            StepType = "transform",
            PluginId = "map-fields",
            Parameters = new Dictionary<string, object>
            {
                ["mappings"] = new Dictionary<string, string>
                {
                    ["cust_id"] = "customer_id",
                    ["cust_name"] = "customer_name"
                }
            }
        },
        new PipelineStepDef
        {
            StepType = "validate",
            PluginId = "schema-validator",
            Parameters = new Dictionary<string, object>
            {
                ["required_fields"] = new[] { "customer_id", "email" }
            }
        }
    }
};

// Save
var manager = new PipelineManager(editor);
await manager.SaveAsync(pipeline);

Running a Pipeline

// Run by ID
var result = await manager.RunAsync(
    pipelineId: "daily-customer-sync",
    progress: new Progress<PassedArgs>(p => 
        Console.WriteLine($"Progress: {p.Messege}")),
    token: cancellationToken);

Console.WriteLine($"Records processed: {result.RecordsProcessed}");
Console.WriteLine($"Records rejected: {result.RecordsRejected}");
Console.WriteLine($"Duration: {result.Duration}");

// Run by definition
var result2 = await manager.RunDefinitionAsync(
    pipeline,
    progress: progressReporter);

Scheduling

var host = new SchedulerHost(editor);
await host.StartAsync();

// Manual trigger
var runId = await host.TriggerManualAsync(
    "daily-customer-sync",
    new Dictionary<string, object> { ["full_refresh"] = true });

// Backfill
var backfillId = await host.SubmitBackfillAsync(
    scheduleId: "daily-customer-sync",
    fromValue: "2024-01-01T00:00:00Z",
    toValue: "2024-01-31T23:59:59Z",
    reason: "Initial load",
    requestedBy: "admin");

Plugin Development

[PipelinePlugin(
    pluginId: "custom-transform",
    category: "transform",
    name: "Custom Transform",
    description: "Applies custom business logic")]
public class CustomTransformPlugin : IPipelineTransformer
{
    private readonly IDMEEditor _editor;

    public CustomTransformPlugin(IDMEEditor editor)
    {
        _editor = editor;
    }

    public async Task<IEnumerable<PipelineRecord>> TransformAsync(
        IEnumerable<PipelineRecord> records,
        Dictionary<string, object> parameters,
        CancellationToken token)
    {
        var result = new List<PipelineRecord>();
        foreach (var record in records)
        {
            // Custom transformation logic
            record.Data["processed_at"] = DateTime.UtcNow;
            result.Add(record);
        }
        return result;
    }
}