Etl.Net Documentation

ETL.NET is a set of .NET libraries designed to embed Business Intelligence (BI) ETL (Extract, Transform, Load) capabilities directly into any .NET application. It provides a developer-centric alternative to heavy BI tools like SSIS, allowing ETL processes to be part of the application source code, easily debuggable with a standard F5 workflow, and deployable alongside the application.

Key Capabilities:

• Operators: Implements SQL-inspired transformations such as Map, Join, Sort, Distinct, Lookup, Top, Pivot, Cross Apply, Union, Group By, and Aggregate.
• Data Sources/Destinations: Out-of-the-box support for Excel, CSV, SQL Server, XML, Entity Framework, File Systems, FTP/SFTP, Email, Dropbox, and Zip files.
• Observability: Detailed activity tracing with automatic filtering and saving capabilities.
• Extensibility: Designed to be easily extended with new operators or data sources using simple base classes.

For scenarios where you need to join stream data with database data, use EfCoreLookup. This is more efficient than running a query per event.

Default Behavior

By default, EfCoreLookup loads the entire target dataset into an in-memory dictionary. Warning: If your target dataset is too large, this will cause memory issues.

Handling Large Datasets

If the dataset is too large for memory, use one of these two strategies:

1. Subset Query: Use .Query() to limit the target dataset to a specific subset.
2. On-demand Cache: Use .NoCacheFullDataset() to switch to a behavior where the system checks the cache first, and if not found, queries the database and adds it to the cache. You can control the cache size with .CacheSize(int).

Creating missing records

In normalization workflows, you can use .CreateIfNotFound(func) to automatically create a target entity if the lookup fails to find a match.

// Example: Lookup with a subset query and on-demand caching
postStream
    .EfCoreLookup("get related authors", o => o
        .Query(o => o.Set<Author>().Where(a => a.TypeId == 6))
        .On(i => i.AuthorId, i => i.Id)
        .Select((l, r) => new { Post = l, Author = r })
        .NoCacheFullDataset()
        .CacheSize(500))
    .Do("show value on console", i => Console.WriteLine($"{i.Post.Title} ({i.Author.Name})"));

// Example: Lookup with automatic creation if not found
postStream
    .EfCoreLookup("get related authors", o => o
        .Set<Author>()
        .On(i => i.AuthorId, i => i.Id)
        .Select((l, r) => new { Post = l, Author = r })
        .CreateIfNotFound(p => new Author { Name = $"Name {p.AuthorId}" }))
    .Do("show value on console", i => Console.WriteLine($"{i.Post.Title} ({i.Author.Name})"));

The Paillave.Etl.XmlFile operator allows for declarative XML parsing using XPaths. It uses a forward-only, SAX-style parser, meaning it does not load the entire document into memory, making it efficient for large files.

To use it, you follow a three-step pattern:

1. CrossApplyXmlFile: Parses an IFileValue and produces a stream of XmlNodeParsed objects based on an XmlFileDefinition.
2. XmlFileDefinition: Defines the nodes to extract using AddNodeDefinition<T>(name, xPath, mapper).
3. XmlNodeOfType<T>: Filters the stream to emit strongly-typed instances of T based on the node name defined in step 2.

Important: XPaths must be absolute (e.g., starting with /root/...). The mapper resolves XPaths against the document root, not the current node.

root
    .CrossApply("source", _ => new[] { fileValue })
    .CrossApplyXmlFile("parse", d => d.AddNodeDefinition<Person>(
        "person",
        "/root/person",
        m => new Person
        {
            Id        = m.ToXPathQuery<int>("/root/person/@id"),
            FirstName = m.ToXPathQuery<string>("/root/person/@firstName"),
            LastName  = m.ToXPathQuery<string>("/root/person/@lastName"),
            Age       = m.ToXPathQuery<int>("/root/person/age"),
        }))
    .XmlNodeOfType<Person>("to person", "person")
    .Do("collect", p => Console.WriteLine(p.FirstName));

ETL.NET is built on a cascade of events where a set of operators (nodes) interact to form a stream.

An operator's lifecycle and behavior follow these rules:

• Listening: An operator listens to one or several sources of events.
• Emitting: Based on received events, an operator emits output events to its listeners.
• Reporting: Every operator reports its activity to the runtime. There are two types of reported events:
  • content event: Contains the payload/actual value resulting from the operation.
  • end of stream event: Notifies that no further content events will follow.
• Error Handling: If an error occurs, the operator emits it to the runtime. The runtime then requests all operators to stop, stops itself, and returns a failed execution status with the error.
• Completion: When the runtime receives an end of stream event from every operator in the process, it stops and returns a successful execution status.

The Paillave.EtlNet.SqlServer extension does not provide a dedicated lookup operator. Instead, use the core ETL.NET Lookup operator combined with a stream retrieved via CrossApplySqlServerQuery.

Warning: Memory Usage

Lookup waits for the entire target stream to complete and stores it in memory. For large datasets, use LeftJoin instead.

Optimized Join Pattern

To use LeftJoin efficiently with billions of rows, ensure both streams are sorted on the pivot key. You can use EnsureSorted to verify sorting without re-sorting, and EnsureKeyed to verify the target stream is sorted and contains no duplicates.

Note: If a stream is already sorted, do not call a sort operator; use EnsureSorted to validate it.

// Optimized pattern using LeftJoin and sorting validation
var authorStream = contextStream
    .CrossApplySqlServerQuery("get authors", o => o
        .FromQuery("select a.* from dbo.Author as a order by a.Id")
        .WithMapping(i => new
        {
            Id = i.ToNumberColumn<int>("Id"),
            Name = i.ToColumn("Name"),
            Reputation = i.ToNumberColumn<int>("Reputation")
        }))
    .EnsureKeyed("ensure authors are sorted by Id with no duplicate", i => i.Id);

postStream
    .EnsureSorted("ensure posts are sorted by AuthorId", i => i.AuthorId)
    .LeftJoin("get related author", authorStream,
        l => l.AuthorId,
        r => r.Id,
        (l, r) => new { Post = l, Author = r })
    .Do("show value on console", i => Console.WriteLine($"{i.Post.Title} ({i.Author.Name})"));

You can define a specialized trace process using TraceProcessDefinition in ExecutionOptions. This process receives two streams: an IStream<TraceEvent> (the telemetry) and an ISingleStream<string> (the original content stream).

Common TraceEvent.Content types for monitoring:

• CounterSummaryStreamTraceContent: Emitted when a node finishes processing, containing a Counter of processed items.
• UnhandledExceptionStreamTraceContent: Emitted when a node encounters an error, containing the exception Message and Level.

By filtering these types, you can create a dedicated log file (e.g., a CSV) that records only the lifecycle events and errors of your main ETL pipeline.

private static void DefineTraceProcess(IStream<TraceEvent> traceStream, ISingleStream<string> contentStream)
{
    traceStream
        .Where("keep only summary of node and errors", i => 
            i.Content is CounterSummaryStreamTraceContent || 
            i.Content is UnhandledExceptionStreamTraceContent)
        .Select("create log entry", i => new
        {
            DateTime = DateTime.Now,
            Type = i.Content switch
            {
                CounterSummaryStreamTraceContent => "EndOfNode",
                UnhandledExceptionStreamContent => "Error",
                _ => "Unknown"
            },
            Message = i.Content switch
            {
                CounterSummaryStreamTraceContent counterSummary => $"{i.NodeName}: {counterSummary.Counter}",
                UnhandledExceptionStreamTraceContent unhandledException => $"{i.NodeName}: {unhandledException.Message}",
                _ => "Unknown"
            }
        })
        .ToTextFileValue("write log file", "log.csv", ...);
}

In ETL.NET, normalizing flat structures (dispatching one input row into multiple related database tables) is achieved using correlated streams.

By default, streams are not correlated to optimize performance and memory. A correlated stream is a stream where each row carries a list of unique identifiers (UIDs) that link it back to its original source rows. This allows subsequent operators to maintain relationships between data even after transformations like Distinct or grouping.

The Normalization Workflow

1. Correlate the stream: Use SetForCorrelation to assign unique identifiers to each row.
2. Transform/Group: Perform operations like Distinct. The resulting rows will aggregate the UIDs of all source rows that formed that distinct entry.
3. Save Parent Data: Save the distinct/grouped rows to a database (e.g., using Paillave.EtlNet.EntityFrameworkCore or Paillave.EtlNet.SqlServer). The database returns the generated primary keys.
4. Re-correlate to Children: Use the CorrelateToSingle operator to map the original individual rows back to the newly saved parent records by matching their UIDs. This allows you to save child rows with the correct foreign keys.

// Conceptual workflow for normalization
// 1. Set correlation
// 2. Distinct/Transform
// 3. Save to DB (gets Parent ID)
// 4. CorrelateToSingle (links original row to Parent ID)
// 5. Save child row with Parent ID as Foreign Key

Connectors provide an abstract way to define input and output points (FTP, SFTP, Email, File System, etc.) before the process starts. They are identified by a unique code.

1. Register Connectors: Add FileValueProvider (for inputs) or FileValueProcessor (for outputs) to a FileValueConnectors instance.
2. Configure Execution: Pass the FileValueConnectors to the Connectors property of ExecutionOptions.
3. Use in Stream: Use .FromConnector("description", "CODE") to start a stream from an input, and .ToConnector("description", "CODE") to send data to an output.

// 1. Setup Connectors
var executionOptions = new ExecutionOptions<string>
{
    Connectors = new FileValueConnectors()
        .Register(new FileSystemFileValueProvider("PTF", "Portfolios", Path.Combine(Environment.CurrentDirectory, "InputFiles"), "*.Portfolios.csv"))
        .Register(new FileSystemFileValueProvider("POS", "Positions", Path.Combine(Environment.CurrentDirectory, "InputFiles"), "*.Positions.csv"))
        .Register(new FileSystemFileValueProcessor("OUT", "Result", Path.Combine(Environment.CurrentDirectory, "OutputFiles"))),
};

// 2. Use in Process
contextStream
    .FromConnector("Get portfolio files", "PTF")
    .CrossApplyTextFile("Parse portfolio file", FlatFileDefinition.Create(i => new
    {
        SicavCode = i.ToColumn("SicavCode"),
        // ... other columns
    }).IsColumnSeparated(','))
    .Do("print portfolio names to console", i => Console.WriteLine(i.PortfolioName));

contextStream
    .FromConnector("Get position files", "POS")
    .Do("print position file names to console", i => Console.WriteLine(i.Name))
    .ToConnector("Save copy of position file", "OUT");