Tutorial: Build a Data Pipeline

1. Source and destination

using Beep.Skia;
using Beep.Skia.ETL;

var manager = new DrawingManager();

var source = new ETLSource { X = 40, Y = 60, Width = 160, Height = 70, DisplayText = "Orders" };
var target = new ETLTarget { X = 640, Y = 60, Width = 160, Height = 70, DisplayText = "DW.Orders" };
manager.AddComponent(source);
manager.AddComponent(target);

2. Add a lookup

ETLLookup joins the stream against a reference set. Set the key and the returned columns; FailOnNoMatch decides whether an unmatched row fails the pipeline or passes through.

var lookup = new ETLLookup
{
    X = 260, Y = 60, Width = 180, Height = 80,
    MatchColumns = "CustomerId",
    ReturnColumns = "Name,Segment,Region",
    CacheMode = CacheMode.Full,
    FailOnNoMatch = false
};
manager.AddComponent(lookup);

manager.ConnectComponents(source, lookup);
manager.ConnectComponents(lookup, target);

3. Track changes with a dimension

ETLScd implements the three classic dimension strategies. Type 2 keeps history with effective dates and a current flag.

var dimension = new ETLScd
{
    X = 260, Y = 220, Width = 180, Height = 90,
    Type = ScdType.Type2,
    BusinessKeys = "CustomerId",
    ChangeTrackingColumns = "Segment,Region",
    EffectiveFromColumn = "ValidFrom",
    EffectiveToColumn = "ValidTo",
    CurrentFlagColumn = "IsCurrent",
    TableName = "DimCustomer"
};
manager.AddComponent(dimension);
manager.ConnectComponents(lookup, dimension);

4. Derived columns and filtering

Derived columns and filters use the expression engine, the same syntax the expression tester dialog evaluates.

var derived = new ETLDerivedColumn
{
    X = 480, Y = 220, Width = 180, Height = 80
};
manager.AddComponent(derived);
manager.ConnectComponents(dimension, derived);

var engine = new ExpressionEngine();
var row = new Dictionary<string, object>
{
    ["Total"] = 250m,
    ["Discount"] = 0.1
};
Console.WriteLine(engine.Evaluate("Total * (1 - Discount)", row));   // 225
Console.WriteLine(engine.EvaluateBoolean("Total > 100 AND Discount > 0", row));

5. Profile the data

Before running a pipeline for real, profile the source to see nulls, distinct counts and ranges; the same statistics are useful in data-quality reports.

var profiler = new DataProfiler();
// Feed rows as they arrive; inspect the profile per column:
//   profile.Columns[i].Count / NullCount / DistinctCount / Min / Max / Average
//   profile.Columns[i].NullPercentage(rowCount)

6. Run and measure

Record per-node metrics while the pipeline runs, then report throughput:

var metrics = new PipelineMetrics();
metrics.Nodes.Add(new NodeRunMetrics
{
    NodeName = "LookupCustomers",
    RowsIn = 12000,
    RowsOut = 11840,
    Elapsed = TimeSpan.FromMilliseconds(184)
});

foreach (var node in metrics.Nodes)
    Console.WriteLine($"{node.NodeName}: {node.RowsIn} in / {node.RowsOut} out, {node.RowsPerSecond:0} rows/s");

To execute the graph as a workflow instead of by hand, see the Workflow Engine.