NPipeline 0.52.0

NPipeline

High-performance, streaming data pipelines for .NET

NPipeline is a modern .NET library that makes building complex data processing workflows simple and efficient. Whether you're building AI-driven data pipelines, ETL workflows, processing real-time data streams, or crafting sophisticated data transformation workflows, NPipeline provides the tools you need with minimal ceremony.

Why Choose NPipeline?

Built for Performance

Stream data efficiently using IAsyncEnumerable<T> without loading entire datasets into memory. Perfect for processing large files, database results, or real-time data streams. NPipeline is designed with performance in mind:

• Streaming architecture minimizes memory allocations and GC pressure

• Zero-reflection execution model with pre-compiled delegates and cached execution plans

• Optimized for high-throughput using ValueTask<T> to eliminate heap allocations for synchronous operations

Graph-Based Architecture

Create sophisticated workflows with multiple data sources, parallel processing, joins, and branching operations. Build complex DAGs (Directed Acyclic Graphs) with ease.

Type-Safe by Design

Strongly-typed nodes with compile-time validation prevent common runtime errors. Your IDE will catch type mismatches before you run your code.

Build-Time Guardrails

Roslyn analyzers flag misconfigured retries, blocking calls, silent data loss, and performance anti-patterns before your pipeline ever runs.

Production-Ready Resilience

Built-in error handling with retry strategies, circuit breakers, and dead letter queues. Create robust pipelines that handle failures gracefully with intelligent backoff and jitter patterns to prevent thundering herd problems.

Designed for Testing

Test each node independently. Pass in test data, assert on outputs. Use the testing extensions to run entire pipelines in memory. First-class support for FluentAssertions and AwesomeAssertions.

Use Cases

NPipeline excels in scenarios where you need to process data efficiently and reliably:

• ETL Pipelines - Extract data from databases, APIs, or files; transform it; load it into data warehouses

• Data Migration - Move and transform data between systems with validation and error handling

• Real-time Processing - Process streaming data from message queues, IoT devices, or APIs

• File Processing - Parse, validate, and transform large CSV, JSON, or XML files

• API Integration - Fetch data from multiple APIs, combine results, and sync to your systems

• Data Validation - Clean and validate data with complex business rules and error reporting

• Report Generation - Aggregate data from multiple sources and generate reports or exports

• Batch Processing - Process large datasets efficiently with parallel execution and state management

• AI Data Enrichment & Intelligent Routing - Augment records with LLM-generated classifications, summaries, or entity extractions, then route to the right downstream system based on content, confidence scores, or business rules

Key Concepts

Nodes are the building blocks of your pipeline:

• Sources generate or fetch data (SourceNode<TOut>)
• Transforms process data item by item (TransformNode<TIn, TOut>)
• Sinks consume data and perform final operations (SinkNode<TIn>)

Data Streams transport data between nodes as strongly-typed async streams (IDataStream<T>).

Pipeline Context provides logging, cancellation, error handling, and shared state without carrying data payloads.

Execution Strategies control how nodes process data: sequential, parallel, or batched processing. Mix strategies within the same pipeline.

Prerequisites

NPipeline requires .NET 8.0, .NET 9.0 or .NET 10.0 to take advantage of the latest performance improvements and language features.

The core library is dependency-free.

Quick Start

Install the core package to get started:

dotnet add package NPipeline
dotnet add package NPipeline.Extensions.DependencyInjection

Additionally, install the Analyzer package to enable Roslyn analyzers that provide compile-time checks and guardrails:

dotnet add package NPipeline.Analyzers

Simple Example

Here's a basic pipeline that processes customer orders:

// 1. Define your data models
public record Order(int Id, string Customer, decimal Amount);
public record ProcessedOrder(int Id, string Customer, decimal Amount, decimal Tax, decimal Total);

// 2. Create source, transform and sink nodes
public class OrderSource : SourceNode<Order>
{
    public override IDataStream<Order> OpenStream(PipelineContext context, CancellationToken cancellationToken)
    {
        var orders = new[] { new Order(1, "Alice", 100m), new Order(2, "Bob", 250m), new Order(3, "Carol", 75m) };
        return new DataStream<Order>(orders.ToAsyncEnumerable());
    }
}

public class TaxCalculator : TransformNode<Order, ProcessedOrder>
{
    public override Task<ProcessedOrder> TransformAsync(Order order, PipelineContext context, CancellationToken cancellationToken)
    {
        var tax = order.Amount * 0.08m;
        var total = order.Amount + tax;
        return Task.FromResult(new ProcessedOrder(order.Id, order.Customer, order.Amount, tax, total));
    }
}

public class OrderSink : SinkNode<ProcessedOrder>
{
    public override async Task ConsumeAsync(IDataStream<ProcessedOrder> input, PipelineContext context, CancellationToken cancellationToken)
    {
        await foreach (var order in input.WithCancellation(cancellationToken))
        {
            Console.WriteLine($"Order {order.Id}: {order.Customer} - Total: ${order.Total:F2}");
        }
    }
}

// 3. Define the pipeline
public class OrderPipeline : IPipelineDefinition
{
    public void Define(PipelineBuilder builder, PipelineContext context)
    {
        var source = builder.AddSource<OrderSource, Order>("Order Source");
        var calculator = builder.AddTransform<TaxCalculator, Order, ProcessedOrder>("Tax Calculator");
        var sink = builder.AddSink<OrderSink, ProcessedOrder>("Order Sink");

        builder.Connect(source, calculator);
        builder.Connect(calculator, sink);
    }
}

// 4. Run the pipeline
var services = new ServiceCollection()
    .AddNPipeline(typeof(Program).Assembly)
    .BuildServiceProvider();

await services.RunPipelineAsync<OrderPipeline>();

Available Packages

NPipeline is modular - install only what you need.

Core Packages

Key Extensions

Connectors

... And More

For the complete list of all available packages see the Extensions Documentation and Connectors Documentation.

What's Next?

1. Installation - Set up NPipeline in your project
2. Your First Pipeline - Build a complete pipeline step-by-step
3. Key Concepts - Learn the fundamentals

Contributing

Contributions are welcome! Please see our Contributing Guide for details.

License

MIT License - see LICENSE file for details.