ETL & Workflow Engine

Overview

The Beep Data Management Engine includes a powerful ETL (Extract, Transform, Load) and Workflow system that provides enterprise-grade data processing capabilities. This system supports complex data transformations, automated workflows, and multi-source data synchronization.

ETL Operations

Extract, Transform, and Load data between different sources with validation and error handling

? Workflow Automation

Define and execute complex business workflows with conditional logic and branching

Data Import

Advanced data import capabilities with mapping, validation, and progress reporting

Data Synchronization

Keep data sources synchronized with conflict resolution and change detection

Custom Actions

Extensible framework for creating custom workflow actions and transformations

Rule Engine

Flexible rule system for data validation, transformation, and business logic

ETLEditor

The ETLEditor is the core component for managing Extract, Transform, Load operations.

ETLEditor Interface

public interface IETL
{
    IDMEEditor DMEEditor { get; set; }
    List<ETLScriptDet> Script { get; set; }
    List<ETLScriptHDR> Scripts { get; set; }
    
    IErrorsInfo CreateScript();
    IErrorsInfo RunScript(IProgress<PassedArgs> progress, CancellationToken token);
    IErrorsInfo StopScript();
    IErrorsInfo SaveScript(ETLScriptHDR script);
    IErrorsInfo LoadScript();
    
    List<EntityDataMap> GetEntityDataMaps(string SourceEntityName, string DestEntityName);
    IErrorsInfo GenerateMappingFromTwoEntity(EntityStructure srcentity, EntityStructure destentity);
    IErrorsInfo CreateEntityAs(EntityStructure srcentity, string destdatasourcename, string destentityname);
}

Basic ETL Operations

// Initialize ETL Editor
var etlEditor = dmeEditor.ETL;

// Create a new ETL script
var script = etlEditor.CreateScript();

// Configure source and destination
var mapping = etlEditor.GetEntityDataMaps("SourceTable", "DestinationTable");

// Execute ETL operation with progress reporting
var progress = new Progress<PassedArgs>(args => Console.WriteLine(args.Messege));
var cancellationToken = new CancellationToken();

var result = await Task.Run(() => etlEditor.RunScript(progress, cancellationToken));

if (result.Flag == Errors.Ok)
{
    Console.WriteLine("ETL operation completed successfully");
}
else
{
    Console.WriteLine($"ETL operation failed: {result.Message}");
}

? ETL Features

Data Mapping: Automatic field mapping between source and destination
Validation: Built-in data validation and error handling
Progress Tracking: Real-time progress reporting and cancellation support
Script Management: Save and load ETL scripts for reuse

? WorkFlowEditor

The WorkFlowEditor manages business workflows and automated processes.

WorkFlowEditor Interface

public interface IWorkFlowEditor
{
    IDMEEditor DMEEditor { get; set; }
    List<IWorkFlow> WorkFlows { get; set; }
    List<IWorkFlowAction> Actions { get; set; }
    List<IWorkFlowRule> Rules { get; set; }
    
    IErrorsInfo RunWorkFlow(string WorkFlowName, IProgress<PassedArgs> progress, CancellationToken token);
    IErrorsInfo StopWorkFlow();
    IErrorsInfo SaveWorkFlow(string WorkFlowName);
    IErrorsInfo LoadWorkFlow(string WorkFlowName);
}

Creating and Running Workflows

// Access the workflow editor
var workflowEditor = dmeEditor.WorkFlowEditor;

// Create a new workflow
var workflow = new WorkFlow
{
    Name = "DataProcessingWorkflow",
    Description = "Process customer data and generate reports"
};

// Add actions to the workflow
workflow.Actions.Add(new ExportDataToFileAction
{
    SourceDataSource = "CustomerDB",
    SourceEntity = "Customers",
    DestinationFile = "CustomerExport.csv"
});

workflow.Actions.Add(new CleanseDataAction
{
    DataSource = "CustomerDB",
    Entity = "Customers",
    CleansingRules = new[] { "RemoveNulls", "TrimWhitespace", "ValidateEmails" }
});

// Save the workflow
workflowEditor.SaveWorkFlow(workflow.Name);

// Execute the workflow
var progress = new Progress<PassedArgs>(args => 
{
    Console.WriteLine($"Workflow Progress: {args.Messege}");
});

var result = workflowEditor.RunWorkFlow("DataProcessingWorkflow", progress, CancellationToken.None);

if (result.Flag == Errors.Ok)
{
    Console.WriteLine("Workflow completed successfully");
}
else
{
    Console.WriteLine($"Workflow failed: {result.Message}");
}

DataImportManager

The DataImportManager provides advanced data import capabilities with mapping and validation.

DataImportManager Usage

// Initialize the import manager
var importManager = new DataImportManager(dmeEditor);

// Configure source and destination
importManager.LoadSourceEntityStructure("SourceTable", "SourceDataSource");
importManager.LoadDestEntityStructure("DestinationTable", "DestinationDataSource");

// Set up field mapping
var mapping = new EntityDataMap
{
    EntityName = "CustomerImport",
    MappedEntities = new List<EntityDataMap_DTL>
    {
        new EntityDataMap_DTL 
        { 
            FromFieldName = "CustomerName", 
            ToFieldName = "Name",
            Fieldtype = "string"
        },
        new EntityDataMap_DTL 
        { 
            FromFieldName = "CustomerEmail", 
            ToFieldName = "EmailAddress",
            Fieldtype = "string"
        }
    }
};

importManager.Mapping = mapping;

// Set up filters if needed
importManager.SourceFilters.Add(new AppFilter
{
   FieldName = "Status",
    Operator = "=",
    FilterValue = "Active"
});

// Execute the import
var result = await importManager.ExecuteImportAsync();

if (result.Flag == Errors.Ok)
{
    Console.WriteLine($"Import completed. {importManager.ImportLogData.Count} records processed");
}
else
{
    Console.WriteLine($"Import failed: {result.Message}");
    
    // Review import logs
    foreach (var logEntry in importManager.ImportLogData)
    {
        Console.WriteLine($"{logEntry.Timestamp}: {logEntry.Message} (Record {logEntry.RecordNumber})");
    }
}

Import Features

Field Mapping: Map fields between source and destination with type conversion
Data Validation: Validate data before import with custom rules
Progress Tracking: Real-time progress updates and detailed logging
Error Handling: Continue processing with detailed error reporting

BeepSyncManager

The BeepSyncManager handles schema-driven synchronization between data systems.

Signature-Verified Snapshot (BeepSync)

public partial class BeepSyncManager : IDisposable
{
    public ObservableBindingList<DataSyncSchema> SyncSchemas { get; set; }
    public List<ConflictEvidence> LastRunConflicts { get; private set; }
    public SyncCheckpoint LastRunCheckpoint { get; private set; }

    public BeepSyncManager(IDMEEditor editor);
    public BeepSyncManager(IDMEEditor editor, SyncIntegrationContext integrationContext);

    public void AddSyncSchema(DataSyncSchema schema);
    public Task<IErrorsInfo> SyncDataAsync(DataSyncSchema schema, IProgress<PassedArgs> progress = null, CancellationToken cancellationToken = default);
    public void SyncData(DataSyncSchema schema);
}

BeepSyncManager Usage

// Current sync entrypoint (Editor/BeepSync/BeepSyncManager.Core.cs)
var syncManager = new BeepSyncManager(dmeEditor);

// Build a schema-driven sync definition
var schema = new DataSyncSchema
{
    Name = "ProductionToReporting",
    SourceDataSourceName = "ProductionDB",
    DestinationDataSourceName = "ReportingDB"
};

// Execute synchronization
var result = await syncManager.SyncDataAsync(schema, CancellationToken.None);

// Run metadata surfaces
var conflicts = syncManager.LastRunConflicts;
var checkpoint = syncManager.LastRunCheckpoint;

Console.WriteLine($"Sync status: {result.Status}");
Console.WriteLine($"Conflicts: {conflicts.Count}");

Sync Considerations

Conflict Resolution: Choose appropriate strategy for data conflicts
Performance: Use batch processing for large datasets
Monitoring: Implement proper logging and alerting
Recovery: Plan for sync failures and resumption

Workflow Actions

Built-in workflow actions for common data processing tasks.

ExportDataToFileAction

Export data from any source to files (CSV, Excel, JSON)

new ExportDataToFileAction
{
    SourceDataSource = "CustomerDB",
    SourceEntity = "Customers",
    DestinationFile = "customers.csv",
    Format = ExportFormat.CSV,
    IncludeHeaders = true
}

CleanseDataAction

Clean and standardize data using predefined or custom rules

new CleanseDataAction
{
    DataSource = "CustomerDB",
    Entity = "Customers",
    CleansingRules = new[]
    {
                        "TrimWhitespace",
                        "RemoveNulls",
                        "ValidateEmails",
                        "StandardizePhoneNumbers"
                    }
}

ConditionalExecutionAction

Execute actions based on conditions and business rules

new ConditionalExecutionAction
{
    Condition = "CustomerType = 'Premium'",
    TrueActions = new[] { sendWelcomeEmail },
    FalseActions = new[] { sendStandardEmail }
}

ArchiveOldDataAction

Archive old data based on date criteria

new ArchiveOldDataAction
{
    SourceDataSource = "TransactionDB",
    SourceEntity = "Transactions",
    ArchiveDataSource = "ArchiveDB",
    DateField = "TransactionDate",
    ArchiveThreshold = TimeSpan.FromDays(365)
}

Workflow Rules

Built-in workflow rules for data validation and transformation.

FormatDate

Format dates according to specified patterns

new FormatDate
{
                    InputFormat = "MM/dd/yyyy",
                    OutputFormat = "yyyy-MM-dd",
                    HandleInvalidDates = true
}

GetGUID

Generate unique identifiers for records

new GetGUID
{
                    TargetField = "RecordID",
                    GuidFormat = GuidFormat.Standard
}

GetNow

Set current timestamp for audit fields

new GetNow
{
                    TargetField = "LastModified",
                    DateTimeFormat = "yyyy-MM-dd HH:mm:ss"
}

GetRecordCount

Count records in entities for reporting

new GetRecordCount
{
                    DataSource = "CustomerDB",
                    Entity = "Customers",
                    FilterCondition = "Status = 'Active'"
}

Usage Examples

Complete Data Processing Workflow

public class DataProcessingWorkflow
{
    private readonly IDMEEditor _dmeEditor;
    
    public DataProcessingWorkflow(IDMEEditor dmeEditor)
    {
        _dmeEditor = dmeEditor;
    }
    
    public async Task<IErrorsInfo> ProcessCustomerDataAsync()
    {
        try
        {
            // Step 1: Import data from external source
            var importManager = new DataImportManager(_dmeEditor);
            importManager.LoadSourceEntityStructure("Customers", "ExternalSystem");
            importManager.LoadDestEntityStructure("Customers", "MainDatabase");
            
            var importResult = await importManager.ExecuteImportAsync();
            if (importResult.Flag != Errors.Ok)
                return importResult;
            
            // Step 2: Run data cleansing workflow
            var workflowEditor = _dmeEditor.WorkFlowEditor;
            var cleansingResult = workflowEditor.RunWorkFlow("CustomerDataCleansing", 
                new Progress<PassedArgs>(), CancellationToken.None);
            
            if (cleansingResult.Flag != Errors.Ok)
                return cleansingResult;
            
            // Step 3: Synchronize with reporting database
            var syncManager = new BeepSyncManager(_dmeEditor);
            var syncConfig = new DataSyncConfiguration
            {
                SourceDataSource = "MainDatabase",
                TargetDataSource = "ReportingDB",
                SyncMode = SyncMode.Incremental
            };
            
            var syncResult = await syncService.SynchronizeAsync(syncConfig, 
                new Progress<SyncProgressArgs>(), CancellationToken.None);
            
            // Step 4: Generate completion report
            await GenerateProcessingReportAsync(importResult, cleansingResult, syncResult);
            
            return _dmeEditor.ErrorObject;
        }
        catch (Exception ex)
        {
            _dmeEditor.AddLogMessage("DataProcessing", $"Workflow failed: {ex.Message}", 
                DateTime.Now, -1, ex.StackTrace, Errors.Failed);
            return _dmeEditor.ErrorObject;
        }
    }
    
    private async Task GenerateProcessingReportAsync(params IErrorsInfo[] results)
    {
        var reportData = new
        {
            ProcessingDate = DateTime.Now,
            ImportedRecords = results[0].Flag == Errors.Ok ? "Success" : "Failed",
            CleansingStatus = results[1].Flag == Errors.Ok ? "Success" : "Failed",
            SyncStatus = results[2].Flag == Errors.Ok ? "Success" : "Failed",
            OverallStatus = results.All(r => r.Flag == Errors.Ok) ? "Success" : "Failed"
        };
        
        // Export report to file or send notification
        var exportAction = new ExportDataToFileAction
        {
            Data = new[] { reportData },
            DestinationFile = $"ProcessingReport_{DateTime.Now:yyyyMMdd_HHmmss}.json",
            Format = ExportFormat.JSON
        };
        
        await exportAction.ExecuteAsync();
    }
}

Best Practices

Error Handling: Always check return values and handle errors appropriately
Progress Reporting: Use progress callbacks for long-running operations
Logging: Implement comprehensive logging for troubleshooting
Testing: Test workflows with sample data before production
Monitoring: Set up monitoring and alerting for critical workflows