DataflowBuilder

Documentation

Full documentation, architecture guides, and API references are available on our GitHub Pages Documentation Portal:

• 📖 Getting Started Guide
• 🧩 Pipeline Concepts
• 🔀 Multi-Source Pipelines
• ⚡ Parallelism & Performance
• 🛡️ Error Handling & Fault Tolerance
• 🧱 Custom Blocks
• 🏗️ Architecture
• 🗺️ Roadmap
• 📝 Changelog

Overview

DataflowBuilder simplifies the creation of dataflow pipelines, making it easier to build, manage, and test complex data processing workflows. With its fluent API, you can quickly set up pipelines that handle various data processing tasks efficiently.

One of the key features of DataflowBuilder is its support for parallel asynchronous tasks executions.

By leveraging dataflow blocks, you can process data concurrently, making efficient use of system resources and improving the performance of your applications. This is especially useful for CPU-intensive or I/O-bound operations where tasks can benefit from parallelism.

Features

• Type-safe fluent builder
• Process huge streaming datasets effectively with zero memory leaks
• Implement any asynchronous or synchronous operations (database access, API calls, cloud resource integrations)
• Time-windowed and size-based micro-batching (TimeBatchBlock)
• Keyed batch grouping (GroupBlock)
• Multi-source extension to implement data streaming joins
• Extensible: add your own block extensions via BasePropagatorBlock

Available blocks

• FromSource<T> / FromSources<T1, T2> => Entry Source Blocks
• Process and ProcessAsync => 1-to-1 Transformation blocks (T => T or T => U)
• ProcessMany and ProcessManyAsync => 1-to-N Expansion blocks (T => IEnumerable<U>)
• Batch => Size-based or time-windowed batching (BatchBlock / TimeBatchBlock<T>, T => T[])
• Group => Keyed batch grouping (GroupBlock<TKey, T>, T => T[])
• ToTarget and ToTargetAsync => Terminal action consumer blocks

What You Can Do with DataflowBuilder ?

DataflowBuilder turns complex asynchronous streaming, parallel execution, and backpressure coordination into concise, type-safe pipelines:

Use Case	Solution	Real Example
High-Throughput ETL	Filter invalid data, enrich in parallel, batch into chunks, and perform bulk writes.	DataFlowPipelineBuilder .FromSource<RawRecord>() .Filter(r => r.IsValid) .ProcessAsync(EnrichAsync, maxDegreeOfParallelism: 8) .Batch(batchSize: 100) .ToTargetAsync(BulkInsertAsync) .Build();
Resilient API Calls	Process items asynchronously with retry resilience, delay backoff, and fallback callbacks.	DataFlowPipelineBuilder .FromSource<Order>() .ProcessAsync(ChargeAsync, maxDegreeOfParallelism: 4, retryCount: 3, retryDelay: TimeSpan.FromSeconds(1), onFailed: LogError) .ToTargetAsync(SendReceiptAsync) .Build();
Multi-Stream Reconciliation	Join two independent streams with JoinBlock, pair items into tuples, and detect discrepancies.	DataFlowPipelineBuilder .FromSources<Transaction, Ledger>() .Process(pair => Reconciler.Match(pair.Item1, pair.Item2)) .Filter(diff => !diff.IsBalanced) .ToTarget(RecordDiscrepancy) .Build();
Time-Windowed Micro-Batching	Buffer high-frequency events and flush upon reaching batch capacity OR when timeout expires.	DataFlowPipelineBuilder .FromSource<Metric>() .Batch(batchSize: 250, timeout: TimeSpan.FromSeconds(5)) .ToTargetAsync(FlushMetricsAsync) .Build();
Tenant / Keyed Partitioning	Group streaming elements by key and emit individual batches when a key reaches capacity.	DataFlowPipelineBuilder .FromSource<AuditEvent>() .Group(evt => evt.TenantId, boundedCapacity: 50) .ToTargetAsync(WriteTenantAuditAsync) .Build();
Channel & Stream Ingestion	Ingest bounded/unbounded ChannelReader<T> or IAsyncEnumerable<T> with auto-completion.	var pipeline = DataFlowPipelineBuilder .FromSource<Log>() .Process(Sanitize) .ToChannel(writer) .Build(); await pipeline.SendFromChannelAsync(reader);

Getting Started

Installation

To use DataflowBuilder in your project, install it via NuGet Package Manager DotnetKit.DataflowBuilder NUGET

Install-Package DotnetKit.DataflowBuilder

Or using the .NET CLI:

dotnet add package DotnetKit.DataflowBuilder

How it works

• Define typed item source and build another blocks with fluent API builder
• Build the pipeline
• Send item

blocks could transform, enrich, group each data (item) sent to the pipeline

Sequential pipeline flow

flowchart LR
    Source --> Process1[Enrich] --> Process2[Transform] --> Target

Loading

Pipeline flow with parallelization

flowchart LR
    Source --> Process1[Enrich] --> Process2[Parallel processing] --> Store1
    Process2[Parallel processing] --> Store2
    Process2[Parallel processing] --> Store3
    Store1 --> Target
    Store2 --> Target
    Store3 --> Target

Loading

Basic Usage

Here is a simple example to get you started with DataflowBuilder.

Create a Pipeline:

using DotnetKit.DataflowBuilder;
using System.Threading.Tasks;

public async Task RunPipeline()
{
    // build the pipeline
    var pipeline = DataFlowPipelineBuilder.FromSource<int>()
        .Process(a => a * 2)
        .ToTarget(a =>
        {
            Console.WriteLine(a);
        })
        .Build();

    // run the pipeline
    for(var i=1;i<=5;i++)
        {
            await pipeline.SendAsync(1);
        }

    await pipeline.CompleteAsync();
}

Result:

2
4
6
8
10

Advanced Usage

DataflowBuilder supports advanced operations like batching and grouping.

Size-Based Batching

var pipeline = DataFlowPipelineBuilder.FromSource<int>()
    .Batch(5)
    .ToTarget(batch =>
    {
        Console.WriteLine($"Batch received: {string.Join(",", batch)}");
    })
    .Build();

for (int i = 1; i <= 10; i++)
{
    await pipeline.SendAsync(i);
}

await pipeline.CompleteAsync();

Time-Windowed Batching (TimeBatchBlock)

Buffer streaming events and automatically flush when batchSize is reached OR when the timeout duration expires:

var pipeline = DataFlowPipelineBuilder.FromSource<MetricEvent>()
    .Batch(batchSize: 100, timeout: TimeSpan.FromSeconds(2))
    .ToTargetAsync(async batch =>
    {
        await metricsClient.SendBatchAsync(batch);
    })
    .Build();

await pipeline.SendAsync(new MetricEvent("cpu_load", 0.75));
// Automatically flushes after 2 seconds even if batchSize has not been reached
await pipeline.CompleteAsync();

Grouping

var pipeline = DataFlowPipelineBuilder.FromSource<char[]>()
    .ProcessMany(chars => chars.GroupBy(c => c).Select(g => g.Key))
    .ToTarget(c =>
    {
        Console.Write(c);
    })
    .Build();

await pipeline.SendAsync("hello world".ToCharArray());
await pipeline.CompleteAsync();

Multi-source pipeline (JoinBlock implementation)

        var expected = new List<string>() { $"HelloWorld-10", $"PingPong-8" };

        var result = new List<string>();
        var pipeline = DataFlowPipelineBuilder.FromSources<string, int>()
            .Process(items => $"{items.Item1}-{items.Item2}")
            .ToTargetAsync(item =>
            {
                result.Add(item);
                return Task.CompletedTask;
            })

            .Build();

        //Order of sending is important for sources
        await pipeline.Send1Async("HelloWorld");
        await pipeline.Send1Async("PingPong");

        await pipeline.Send2Async("HelloWorld".Length);
        await pipeline.Send2Async("PingPong".Length);

        await pipeline.CompleteAsync();

        result.Should().BeEquivalentTo(expected);

Real world example with parallelization

flowchart TD
    A[Sensor item] -->|Process| B(Enriched sensor)
    B --> |batch| C(process each 1000 items with 3 // tasks )
    C -->|processing task1| D[Bulk insert 1000 sensors]
    C -->|processing task2| E[Bulk insert 1000 sensors]
    C -->|processing task3| F[Bulk insert 1000 sensors]
    D --> |task1 processed| G[Target log event]
    E --> |task2 processed| G[Target ]
    F --> |task3 processed| G[Publish notification]

Loading

//Configuration stage (DI, startup process, or specific lifetime )
//Building sensor bulk insertion pipeline
var pipeline = DataFlowPipelineBuilder.FromSource<SensorEntity>()
    .Process(sensor => sensor.EnrichAsync())
    .Batch(1000)
    .ProcessAsync(async enrichedSensors => {
       await mongoDBClient.BulkInsertAsync(enrichedSensors)
    }, maxDegreeOfParallelism: 3)
    .ToTargetAsync(enrichedSensors =>
    {
     await servicebusDomainTopic.PublishCreatedSensorsNotificationAsync();
    })
    .Build();

//Execution stage (API or Event trigger )
//it could be billion of sensors
await foreach(var sensorPage in GetAllSensorsAsync())
    {
        await _pipeline.SendAsync(sensor);
    }
await _pipeline.CompleteAsync();

Contributing

Contributions are welcome! Feel free to open issues or submit pull requests.

License

This project is licensed under the MIT License.