NPipeline.Connectors.DataLake 0.65.1

NPipeline Data Lake Connector

NPipeline Data Lake Connector provides table abstractions for building data lakes on top of the Parquet connector. This package enables partitioned writes, manifest-based table management, time travel queries, and small-file compaction with Parquet as the default storage format.

About NPipeline

NPipeline is a high-performance, extensible data processing framework for .NET that enables developers to build scalable and efficient pipeline-based applications. It provides a rich set of components for data transformation, aggregation, branching, and parallel processing, with built-in support for resilience patterns and error handling.

Installation

dotnet add package NPipeline.Connectors.DataLake

Requirements

• .NET 8.0, 9.0, or 10.0
• NPipeline.Connectors.Parquet (automatically included as a dependency)
• NPipeline.Connectors (automatically included as a dependency)
• NPipeline.StorageProviders (automatically included as a dependency)

Relationship to Parquet Connector

This package builds on NPipeline.Connectors.Parquet and uses Parquet as its default file format. The Data Lake connector adds:

• Hive-style partitioning: Automatic directory structure with column=value/ patterns
• Manifest tracking: NDJSON-based file inventory with snapshot IDs
• Time travel: Read table state as of a specific timestamp or snapshot
• Compaction: Merge small files into larger, query-optimized files
• Format adapters: Extensibility for Iceberg, Delta, or custom table formats

Why this separation: The Parquet connector handles single-file I/O with full Parquet feature support. The Data Lake connector adds table-level semantics ( partitioning, snapshots, time travel) without duplicating the Parquet implementation. This allows using either package independently or together.

Features

• Partition Specification: Fluent API for defining multi-level partition schemes
• Hive-Style Paths: Standard column=value/ directory structure for query engine compatibility
• DataLakeTableWriter: Append and snapshot APIs for writing partitioned data
• DataLakeTableSourceNode: Read all files in a table via manifest lookup
• Time Travel: Query historical table state by timestamp or snapshot ID
• Manifest Tracking: NDJSON manifest with per-snapshot files for auditability
• Small-File Compaction: Consolidate files below size thresholds
• Format Adapter Interface: Extend to Iceberg, Delta Lake, or custom formats

Usage

Defining Partition Specifications

Use PartitionSpec<T> to define how records map to partition directories:

using NPipeline.Connectors.DataLake.Partitioning;

public class SalesRecord
{
    public long Id { get; set; }
    public string ProductName { get; set; } = string.Empty;
    public decimal Amount { get; set; }
    public DateTime EventDate { get; set; }  // Partition column
    public string Region { get; set; } = string.Empty;  // Partition column
}

// Single-level partitioning
var spec = PartitionSpec<SalesRecord>.By(x => x.EventDate);

// Multi-level partitioning (produces: event_date=2025-01-15/region=EU/)
var spec = PartitionSpec<SalesRecord>
    .By(x => x.EventDate)
    .ThenBy(x => x.Region);

// Custom column names
var spec = PartitionSpec<SalesRecord>
    .By(x => x.EventDate, "date")
    .ThenBy(x => x.Region, "geo_region");

// No partitioning (all files in table root)
var spec = PartitionSpec<SalesRecord>.None();

Why fluent builder: The fluent API makes partition schemes readable and discoverable. Property expressions are compiled to delegates for efficient runtime evaluation. Column names default to snake_case (e.g., EventDate → event_date) following Hive conventions.

Writing Partitioned Data

Use DataLakeTableWriter<T> to write partitioned Parquet files:

using NPipeline.Connectors.DataLake;
using NPipeline.Connectors.DataLake.Partitioning;
using NPipeline.Connectors.Parquet;
using NPipeline.StorageProviders;
using NPipeline.StorageProviders.Models;
using NPipeline.DataFlow.DataStreams;

var resolver = StorageProviderFactory.CreateResolver();
var provider = StorageProviderFactory.GetProviderOrThrow(
    resolver,
    StorageUri.Parse("file:///data/warehouse"));

var tableUri = StorageUri.Parse("file:///data/warehouse/sales_table");
var partitionSpec = PartitionSpec<SalesRecord>
    .By(x => x.EventDate)
    .ThenBy(x => x.Region);

var config = new ParquetConfiguration
{
    RowGroupSize = 100_000,
    Compression = Parquet.CompressionMethod.Snappy
};

// Write data
await using var writer = new DataLakeTableWriter<SalesRecord>(
    provider,
    tableUri,
    partitionSpec,
    config);

Console.WriteLine($"Snapshot ID: {writer.SnapshotId}");

var dataStream = new InMemoryDataStream<SalesRecord>(records, "SalesData");
await writer.AppendAsync(dataStream, CancellationToken.None);

Generated directory structure:

sales_table/
├── _manifest/
│   ├── manifest.ndjson
│   └── snapshots/
│       └── 20250215093045abcd1234.ndjson
├── event_date=2025-01-15-00-00-00/
│   ├── region=EU/
│   │   └── part-001.parquet
│   └── region=US/
│       └── part-001.parquet
└── event_date=2025-01-16-00-00-00/
    └── region=APAC/
        └── part-001.parquet

Reading Table Data

Use DataLakeTableSourceNode<T> to read all data in a table:

using NPipeline.Connectors.DataLake;

// Read latest snapshot
var sourceNode = new DataLakeTableSourceNode<SalesRecord>(
    provider,
    tableUri);

// Use in a pipeline
var builder = new PipelineBuilder();
var source = builder.AddSource(() => sourceNode, "table-source");

The source node:

1. Reads the manifest file to discover all data files
2. Deduplicates entries by path (keeps latest version)
3. Streams data from each file using ParquetSourceNode<T>

Time Travel Queries

Read table state as of a specific point in time:

using NPipeline.Connectors.DataLake;

// Read as of a specific timestamp
var asOfTimestamp = new DateTimeOffset(2025, 1, 15, 12, 0, 0, TimeSpan.Zero);
var timeTravelSource = new DataLakeTableSourceNode<SalesRecord>(
    provider,
    tableUri,
    asOfTimestamp);

// Read a specific snapshot by ID
var snapshotSource = new DataLakeTableSourceNode<SalesRecord>(
    provider,
    tableUri,
    snapshotId: "20250215093045abcd1234");

Why time travel: Time travel enables:

• Debugging: Reproduce issues by querying historical state
• Audit: Track data changes over time
• Rollback analysis: Compare current vs. previous snapshots
• Point-in-time reporting: Generate reports as of specific dates

Manifest Format

The manifest is stored as NDJSON at _manifest/manifest.ndjson:

{"path":"event_date=2025-01-15/region=EU/part-001.parquet","row_count":5000,"written_at":"2025-01-15T10:30:45Z","file_size_bytes":245678,"partition_values":{"event_date":"2025-01-15","region":"EU"},"snapshot_id":"20250215103045abcd1234","format_version":"v1","file_format":"parquet","compression":"snappy"}
{"path":"event_date=2025-01-15/region=US/part-001.parquet","row_count":3200,"written_at":"2025-01-15T10:30:46Z","file_size_bytes":198234,"partition_values":{"event_date":"2025-01-15","region":"US"},"snapshot_id":"20250215103045abcd1234","format_version":"v1","file_format":"parquet"}

Each ManifestEntry tracks:

Why NDJSON: Newline-delimited JSON allows:

• Append-only writes: New entries added without rewriting the entire file
• Streaming reads: Process entries line-by-line without loading entire manifest
• Human readability: Easy inspection with standard tools
• Per-snapshot files: Each snapshot gets its own manifest file for isolation

Inspecting the Manifest

using NPipeline.Connectors.DataLake.Manifest;

var manifestReader = new ManifestReader(provider, tableUri);

// Read all entries
var entries = await manifestReader.ReadAllAsync(cancellationToken);

foreach (var entry in entries)
{
    Console.WriteLine($"Path: {entry.Path}");
    Console.WriteLine($"  Rows: {entry.RowCount}, Size: {entry.FileSizeBytes:N0} bytes");
    Console.WriteLine($"  Snapshot: {entry.SnapshotId}");
    Console.WriteLine($"  Written: {entry.WrittenAt:yyyy-MM-dd HH:mm:ss}");

    if (entry.PartitionValues is not null)
    {
        var partitions = string.Join(", ",
            entry.PartitionValues.Select(kv => $"{kv.Key}={kv.Value}"));
        Console.WriteLine($"  Partitions: {partitions}");
    }
}

// Get available snapshot IDs
var snapshotIds = await manifestReader.GetSnapshotIdsAsync(cancellationToken);

Compaction

Use DataLakeCompactor to consolidate small files:

using NPipeline.Connectors.DataLake;
using NPipeline.Connectors.DataLake.FormatAdapters;

var compactor = new DataLakeCompactor(provider, tableUri, new ParquetConfiguration());

// Dry run to see what would be compacted
var dryRunRequest = new TableCompactRequest
{
    TableBasePath = tableUri,
    Provider = provider,
    SmallFileThresholdBytes = 32L * 1024 * 1024, // 32 MB
    MinFilesToCompact = 5,
    MaxFilesToCompact = 100,
    TargetFileSizeBytes = 256L * 1024 * 1024, // 256 MB
    DryRun = true,
    DeleteOriginalFiles = true
};

var dryRunResult = await compactor.CompactAsync(dryRunRequest, cancellationToken);
Console.WriteLine($"Would compact {dryRunResult.FilesCompacted} files into {dryRunResult.FilesCreated}");

// Perform actual compaction
var actualRequest = dryRunRequest with { DryRun = false };
var result = await compactor.CompactAsync(actualRequest, cancellationToken);

Console.WriteLine($"Compacted {result.FilesCompacted} files in {result.Duration.TotalSeconds:N1}s");
Console.WriteLine($"Bytes: {result.BytesBefore:N0} → {result.BytesAfter:N0}");

Why compaction: Small files degrade query performance because:

• Query engines must open and close many file handles
• Metadata overhead (footers, schemas) is repeated per file
• Network latency compounds with many S3/Blob Storage requests

Compaction merges small files into fewer, larger files while preserving data and updating the manifest.

Resilience

The manifest writer appends to _manifest/manifest.ndjson through an NResilience policy, DataLakeConnectorResilience.ManifestWrite: three attempts (two retries) with exponential backoff and full jitter from 100 ms, no attempt timeout, and no deadline. Only transient storage errors are retried (IOException, TimeoutException, socket errors). Access denied (UnauthorizedAccessException), missing paths, and invalid arguments fail at once. Pass a different policy to the ManifestWriter constructor to change this.

A retry recovers from a transient error, not from a conflict with a concurrent writer. Each attempt re-reads the manifest and skips the append when its entries are already present, so a retry after an attempt that committed doesn't duplicate entries.

The cloud storage providers' SDKs (Azure, AWS, Google) retry individual requests natively; this policy retries the whole append.

Concurrent writers

The main manifest is last-writer-wins. An append reads manifest.ndjson, adds its entries, and replaces the file (by atomic rename on providers that support it, such as ADLS Gen2 and the local file system, and by overwriting it on the others). There is no conditional write, so when two writers append at the same time, one writer's entries can be missing from the main manifest.

Readers still see every entry. Before appending, each flush writes the writer's per-snapshot manifest, _manifest/snapshots/{snapshotId}.ndjson, holding every entry that writer has flushed. Only that writer writes this file. ManifestReader (and so DataLakeTableSourceNode and time travel) merges all snapshot manifests into the main manifest, so entries lost from the main manifest come back. Two conditions apply:

• Give each writer its own snapshot ID. The built-in writers generate one with ManifestWriter.GenerateSnapshotId().
• Tools that read manifest.ndjson directly, without the snapshot files, can miss entries.

Reads use DataLakeConnectorResilience.ManifestRead (the same attempts, backoff, and classifier as writes). A missing manifest or snapshot directory reads as empty. Any other failure to list or read the manifest or a snapshot file is retried and then thrown, so a read never silently returns fewer entries than the table has. Pass a different policy to the ManifestReader constructor to change this.

Format Adapter Interface

Implement ITableFormatAdapter to support alternative table formats:

using NPipeline.Connectors.DataLake.FormatAdapters;

public class IcebergFormatAdapter : ITableFormatAdapter
{
    public string Name => "iceberg";

    public Task AppendAsync(TableAppendRequest request, CancellationToken cancellationToken = default)
    {
        // Implement Iceberg-specific commit protocol
        // Write metadata files, update snapshot log, etc.
    }

    public Task<TableSnapshot> GetSnapshotAsync(TableSnapshotRequest request, CancellationToken cancellationToken = default)
    {
        // Read Iceberg metadata to resolve snapshot
    }

    public Task<TableCompactResult> CompactAsync(TableCompactRequest request, CancellationToken cancellationToken = default)
    {
        // Implement compaction with Iceberg metadata updates
    }

    public Task<IReadOnlyList<SnapshotSummary>> ListSnapshotsAsync(StorageUri tableBasePath, CancellationToken cancellationToken = default)
    {
        // Read Iceberg snapshot log
    }

    public Task<bool> TableExistsAsync(StorageUri tableBasePath, CancellationToken cancellationToken = default)
    {
        // Check for Iceberg metadata files
    }

    public Task CreateTableAsync(StorageUri tableBasePath, CancellationToken cancellationToken = default)
    {
        // Initialize Iceberg table metadata
    }
}

Why format adapters: Different table formats (Iceberg, Delta Lake, Hudi) have different:

• Metadata file formats and locations
• Commit protocols and concurrency models
• Time travel and snapshot semantics

The adapter interface isolates format-specific logic while reusing the core partitioning and file I/O infrastructure.

Hive-Style Partition Paths

The connector generates standard Hive-style partition paths for compatibility with query engines:

event_date=2025-01-15/region=EU/

Path format rules:

Why Hive-style: Hive-style partitioning is supported by:

• Apache Spark / PySpark
• AWS Athena / Glue
• Azure Synapse / Data Lake Analytics
• Google BigQuery
• Trino / Presto
• DuckDB

This allows the same data files to be queried by multiple engines without ETL.

Production Considerations

File Sizing

Target file sizes between 256 MB and 1 GB for optimal query performance:

var config = new ParquetConfiguration
{
    RowGroupSize = 100_000,
    TargetFileSizeBytes = 512L * 1024 * 1024 // 512 MB
};

Memory Management

Control memory during high-cardinality partition writes:

var config = new ParquetConfiguration
{
    MaxBufferedRows = 500_000,  // Total rows across all partition buffers
    RowGroupSize = 50_000       // Rows per row group
};

Idempotent Writes

Use idempotency keys to prevent duplicate data after retries:

var request = new TableAppendRequest<SalesRecord>
{
    TableBasePath = tableUri,
    Provider = provider,
    PartitionSpec = partitionSpec,
    IdempotencyKey = $"batch-{batchId}"  // Deduplicates on retry
};

Manifest Backup

The manifest is the source of truth for table contents. Consider:

• Versioning the _manifest/ directory in object storage
• Periodic exports to a backup location
• Monitoring manifest file size (split if too large)

Compaction Strategy

Run compaction as a scheduled job:

// Compact files smaller than 32 MB into 256 MB files
var request = new TableCompactRequest
{
    TableBasePath = tableUri,
    Provider = provider,
    SmallFileThresholdBytes = 32L * 1024 * 1024,
    TargetFileSizeBytes = 256L * 1024 * 1024,
    MinFilesToCompact = 10,  // Only compact when enough small files exist
    DeleteOriginalFiles = true
};

Complete Pipeline Example

using NPipeline.Connectors.DataLake;
using NPipeline.Connectors.DataLake.Partitioning;
using NPipeline.Connectors.Parquet;
using NPipeline.Connectors.Parquet.Attributes;
using NPipeline.Pipeline;
using NPipeline.StorageProviders;
using NPipeline.StorageProviders.Models;

public class SalesRecord
{
    [ParquetColumn("sale_id")]
    public long Id { get; set; }

    [ParquetColumn("product_name")]
    public string ProductName { get; set; } = string.Empty;

    [ParquetDecimal(18, 2)]
    public decimal Amount { get; set; }

    public DateTime EventDate { get; set; }  // Partition column

    public string Region { get; set; } = string.Empty;  // Partition column
}

public class DataLakePipeline : IPipelineDefinition
{
    private readonly StorageUri _tableUri = StorageUri.Parse("s3://warehouse/sales_table/");

    public void Define(PipelineBuilder builder, PipelineContext context)
    {
        var resolver = StorageProviderFactory.CreateResolver();
        var provider = StorageProviderFactory.GetProviderOrThrow(resolver, _tableUri);

        var partitionSpec = PartitionSpec<SalesRecord>
            .By(x => x.EventDate)
            .ThenBy(x => x.Region);

        var config = new ParquetConfiguration
        {
            RowGroupSize = 100_000,
            Compression = Parquet.CompressionMethod.Snappy,
            TargetFileSizeBytes = 512L * 1024 * 1024
        };

        // Source: Read from Data Lake table with time travel
        var asOfDate = new DateTimeOffset(2025, 1, 15, 0, 0, 0, TimeSpan.Zero);
        var source = builder.AddSource(
            new DataLakeTableSourceNode<SalesRecord>(provider, _tableUri, asOfDate),
            "lake-source");

        // Transform: Process records
        var transform = builder.AddTransform<SalesTransform, SalesRecord, SalesRecord>("transform");

        // Sink: Write back to Data Lake with partitioning
        var sink = builder.AddSink(
            new DataLakePartitionedSinkNode<SalesRecord>(
                provider,
                _tableUri,
                partitionSpec,
                config),
            "lake-sink");

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

Related Packages

• NPipeline - Core pipeline framework
• NPipeline.Connectors.Parquet - Parquet file I/O (used internally)
• NPipeline.Connectors - Storage abstractions and base connectors
• NPipeline.Extensions.DependencyInjection - Dependency injection integration

License

This package is licensed under the Business Source License 1.1.

Free for non-production use. Production use is free for organizations with 4 or fewer developers and annual revenue of $5M AUD or less. Larger organizations require a commercial license. This license automatically converts to MIT two years after each release.