Quickstart Guide

This guide walks you through installing FastIngest and executing your first bulk data import pipeline in under 5 minutes.

---

1. Installation

Install the core FastIngest package alongside the database sink of your choice via the .NET CLI or Package Manager Console:

PostgreSQL

dotnet add package FastIngest.Core
dotnet add package FastIngest.PostgreSql

SQL Server

dotnet add package FastIngest.Core
dotnet add package FastIngest.SqlServer

MySQL

dotnet add package FastIngest.Core
dotnet add package FastIngest.MySql

SQLite

dotnet add package FastIngest.Core
dotnet add package FastIngest.Sqlite

MongoDB

dotnet add package FastIngest.Core
dotnet add package FastIngest.MongoDb

Cosmos DB

dotnet add package FastIngest.Core
dotnet add package FastIngest.CosmosDb

Elasticsearch

dotnet add package FastIngest.Core
dotnet add package FastIngest.Elasticsearch

---

2. Define Your Record & Validator

Create a strongly typed record representing each row in your input data, and optionally define validation rules using FluentValidation:

using FluentValidation;

// Define your target model
public record CustomerRecord(int Id, string Email, string FullName, decimal Balance);

// Define validation rules
public class CustomerValidator : AbstractValidator<CustomerRecord>
{
    public CustomerValidator()
    {
        RuleFor(x => x.Email).NotEmpty().EmailAddress();
        RuleFor(x => x.FullName).NotEmpty().MaximumLength(100);
        RuleFor(x => x.Balance).GreaterThanOrEqualTo(0);
    }
}

---

3. Run Ingestion Pipeline (10-Line Example)

Use FastIngestPipeline<TRecord>.Create() to stream the file straight from disk or network into your database:

using FastIngest.Core.Common;
using FastIngest.Core.Pipeline;
using FastIngest.PostgreSql.Extensions;
using Npgsql;

::: code-group

```csharp [CSV Ingestion]
await using var stream = File.OpenRead("customers.csv");
await using var connection = new NpgsqlConnection(connectionString);
await connection.OpenAsync();

var result = await FastIngestPipeline<CustomerRecord>.Create()
    .FromStream(stream, FileType.Csv)
    .WithMapping(m => {
        m.Map(x => x.Id, "customer_id");
        m.Map(x => x.Email, "email");
        m.Map(x => x.FullName, "full_name");
        m.Map(x => x.Balance, "balance");
    })
    .ValidateWith<CustomerValidator>(opt => opt.ErrorStrategy = ErrorStrategy.CollectAndContinue)
    .WithBatchSize(5000)
    .WithChannelCapacity(2)
    .OnProgress(p => Console.WriteLine($"Processed {p.RowsProcessed} rows ({p.PercentComplete:F1}%)..."))
    .WriteToPostgresAsync(connection, "customers");

Console.WriteLine($"Done! Succeeded: {result.TotalSucceeded:N0}, Failed: {result.TotalFailed:N0}");

await using var stream = File.OpenRead("customers.jsonl");
await using var connection = new NpgsqlConnection(connectionString);
await connection.OpenAsync();

// Automatically streams each line via PipeReader & Utf8JsonReader
var result = await FastIngestPipeline<CustomerRecord>.Create()
    .FromStream(stream, FileType.JsonLines) // Or FileType.Ndjson
    .WithJsonOptions(opt => opt.PropertyNameCaseInsensitive = true)
    .ValidateWith<CustomerValidator>(opt => opt.ErrorStrategy = ErrorStrategy.CollectAndContinue)
    .WithBatchSize(5000)
    .WithChannelCapacity(2)
    .OnProgress(p => Console.WriteLine($"Processed {p.RowsProcessed} rows ({p.PercentComplete:F1}%)..."))
    .WriteToPostgresAsync(connection, "customers");

Console.WriteLine($"Done! Succeeded: {result.TotalSucceeded:N0}, Failed: {result.TotalFailed:N0}");

:::

Format Auto-Detection

FastIngest can automatically detect file formats using extensions (.csv, .jsonl, .ndjson) or content inspection (leading { on seekable streams):

pipeline.FromStream(stream, "upload.jsonl", FileType.AutoDetect);
// Or directly from disk:
pipeline.FromFile("customers.jsonl");

For an in-depth dive into line-delimited JSON ingestion, see the NDJSON / JSON Lines Guide.

Channel Tuning & Concurrency

FastIngest runs a decoupled producer-consumer architecture using System.Threading.Channels:

• The Producer task reads, maps, and validates incoming rows.
• The Consumer task streams batches into the database sink.

You can configure the in-flight batch capacity using .WithChannelCapacity(int capacity = 2) or fine-tune channel options with .WithOptions(...):

pipeline
    .WithBatchSize(5000)
    .WithChannelCapacity(3) // Keeps up to 3 batches in flight concurrently
    .WithOptions(opt =>
    {
        opt.BoundedChannelCapacity = 3;
        opt.SingleWriter = true;
        opt.SingleReader = true;
        opt.FullMode = BoundedChannelFullMode.Wait; // Enforces backpressure
    });

---

4. Inspecting Ingestion Results

The pipeline returns an IngestResult object containing comprehensive execution metrics:

if (!result.IsSuccess)
{
    Console.WriteLine($"Encountered {result.TotalFailed} invalid records!");

    // Iterate through validation errors
    foreach (var error in result.Errors.Take(10))
    {
        Console.WriteLine($"Row #{error.RowNumber}: Field '{error.PropertyName}' -> {error.ErrorMessage}");
    }

    // Export all failed rows and errors as a downloadable CSV manifest
    byte[] errorCsvBytes = result.ExportErrorsToCsv();
    await File.WriteAllBytesAsync("ingestion_errors.csv", errorCsvBytes);
}
else
{
    Console.WriteLine($"Successfully ingested {result.TotalSucceeded:N0} rows in {result.Duration.TotalSeconds:F2}s");
}

---

Next Steps

• Explore Dependency Injection & ASP.NET Core API Integration
• Learn about Validation Strategies & Error Manifests
• Check out the specific guide for your database in Database Sinks