Etl.Net

Go to full documentation

:rotating_light: I need some help!!! :ambulance:

I need some serious help for the following subject:

• :books: Documentation. The website of ETL.NET needs to be completed. It contains the essential to start and get in touch, but a dramatic amount of serious features is not documented in there.
• :heavy_check_mark: Unit tests. A proper test code coverage is the only real way to be taken seriously when it is about selecting the right open source library. There was a time when some units tests existed only for the reactive engine of the core (Paillave.EtlNet.Core). But it happened I had to decommission it in the even of the V2 release.

:envelope_with_arrow: Anybody who is keen to participate to the project in these regards is very welcome to contact me at admroyer@hotmail.com.

Presentation

Implementation of a mass processing engine to use in a similar way than Linq with every SSIS features and much more. The reactive approach for the implementation of this engine ensures parallelized multi streams, high performances and low memory foot print even with million rows to process.

ETL.NET is fully written in .NET for a multi platform usage and for a straight forward integration in any application.

Extend it takes 5mn... literally.

Package	nuget version	nuget downloads
Paillave.EtlNet.Core
Paillave.EtlNet.Autofac
Paillave.EtlNet.AzureStorageAccount
Paillave.EtlNet.Dropbox
Paillave.EtlNet.EntityFrameworkCore
Paillave.EtlNet.ExcelFile
Paillave.EtlNet.ExecutionToolkit
Paillave.EtlNet.FileSystem
Paillave.EtlNet.FromConfigurationConnectors
Paillave.EtlNet.Ftp
Paillave.EtlNet.Mail
Paillave.EtlNet.GraphApi
Paillave.EtlNet.Http
Paillave.EtlNet.JsonFile
Paillave.EtlNet.Sftp
Paillave.EtlNet.Pgp
Paillave.EtlNet.S3
Paillave.EtlNet.SqlServer
Paillave.EtlNet.TextFile
Paillave.EtlNet.Bloomberg
Paillave.EtlNet.XmlFile
Paillave.EtlNet.Zip
Paillave.EtlNet.Pdf
Paillave.EntityFrameworkCoreExtension

Examples

Extract it, parse it, upload it

Read all Zip files from a folder, unzip Csv files that are inside, parse them, exclude duplicates, and upsert them into database.

dotnet new console -o SimpleTutorial
cd SimpleTutorial
dotnet add package Paillave.EtlNet.Core
dotnet add package Paillave.EtlNet.FileSystem
dotnet add package Paillave.EtlNet.Zip
dotnet add package Paillave.EtlNet.TextFile
dotnet add package Paillave.EtlNet.SqlServer

using Microsoft.Data.SqlClient;
using Microsoft.Extensions.DependencyInjection;
using Paillave.Etl.Core;
using Paillave.Etl.FileSystem;
using Paillave.Etl.SqlServer;
using Paillave.Etl.TextFile;
using Paillave.Etl.Zip;
using System.Data;

//Register services
var serviceCollection = new ServiceCollection();
serviceCollection.AddTransient<IDbConnection>(_ =>
{
    var conn = new SqlConnection(@"Data Source=MyServer;Initial Catalog=MyDatabase;Integrated Security=True;");
    conn.Open();
    return conn;
});
var services = serviceCollection.BuildServiceProvider();


//Configure an ETL execution (defined below)
var processRunner = StreamProcessRunner.Create<string>(DefineProcess);
var executionOptions = new ExecutionOptions<string>
{
    Services = services,
};
var result = await processRunner.ExecuteAsync(@"C:\Path\To\My\Zip-Files", executionOptions);

Console.WriteLine(result.Failed
    ? $"{result.ErrorTraceEvent.NodeName} ({result.ErrorTraceEvent.NodeTypeName}): {result.ErrorTraceEvent.Content.Message}"
    : "Success!");


//Define the ETL process, in this case starting from a single path string
static void DefineProcess(ISingleStream<string> contextStream)
{
    contextStream
        .CrossApplyFolderFiles("List all Zip files", "*.zip", recursive: true)
        .CrossApplyZipFiles("Extract Csv files from Zips", "*.csv")
        .CrossApplyTextFile("Parse Csv files", FlatFileDefinition.Create(i => new Person
            {
                FirstName = i.ToColumn("first name"),
                LastName = i.ToColumn("last name"),
                DateOfBirth = i.ToDateColumn("date of birth", "yyyy-MM-dd"),
                Email = i.ToColumn("email"),
                Reputation = i.ToNumberColumn<int?>("reputation", ".")
            })
            .IsColumnSeparated(',')
        )
        .Distinct("Exclude duplicates based on the Email", p => p.Email)
        .SqlServerSave("Upsert using Email as key and ignore the Id", o => o
            .ToTable("dbo.Person")
            .SeekOn(p => p.Email)
            .DoNotSave(p => p.Id));
}


class Person
{
    public int Id { get; set; }
    public string FirstName { get; set; }
    public string LastName { get; set; }
    public DateTime DateOfBirth { get; set; }
    public string Email { get; set; }
    public int? Reputation { get; set; }
}

Run it, debug it, track it, log it

Execute an ETL process, debug it by tracking debug events using the IDE debugger, catch execution events and log it into database.

using System;
using System.Threading.Tasks;
using Paillave.Etl.Core;
using Paillave.Etl.FileSystem;
using Paillave.Etl.Zip;
using Paillave.Etl.TextFile;
using Paillave.Etl.SqlServer;
using Microsoft.Data.SqlClient;

namespace SimpleTutorial
{
  class Program
  {
    static async Task Main(string[] args)
    {
      var processRunner = StreamProcessRunner.Create<string>(DefineProcess);
      processRunner.DebugNodeStream += (sender, e) => { /* PLACE A CONDITIONAL BREAKPOINT HERE FOR DEBUG */ };
      using (var cnx = new SqlConnection(args[1]))
      {
        cnx.Open();
        var executionOptions = new ExecutionOptions<string>
        {
          Resolver = new SimpleDependencyResolver().Register(cnx),
          TraceProcessDefinition = DefineTraceProcess,
          // UseDetailedTraces = true // activate only if per row traces are meant to be caught
        };
        var res = await processRunner.ExecuteAsync(args[0], executionOptions);
        Console.Write(res.Failed ? "Failed" : "Succeeded");
        if (res.Failed)
          Console.Write($"{res.ErrorTraceEvent.NodeName}({res.ErrorTraceEvent.NodeTypeName}):{res.ErrorTraceEvent.Content.Message}");
      }
    }
    private static void DefineProcess(ISingleStream<string> contextStream)
    {
      // TODO: define your ELT process here
    }
    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 ExecutionLog
          {
            DateTime = i.DateTime,
            ExecutionId = i.ExecutionId,
            EventType = i.Content switch
            {
              CounterSummaryStreamTraceContent => "EndOfNode",
              UnhandledExceptionStreamTraceContent => "Error",
              _ => "Unknown"
            },
            Message = i.Content switch
            {
              CounterSummaryStreamTraceContent counterSummary => $"{i.NodeName}: {counterSummary.Counter}",
              UnhandledExceptionStreamTraceContent unhandledException => $"{i.NodeName}({i.NodeTypeName}): [{unhandledException.Level.ToString()}] {unhandledException.Message}",
              _ => "Unknown"
            }
          })
        .SqlServerSave("save traces", o => o.ToTable("dbo.ExecutionTrace"));
    }
    private class ExecutionLog
    {
      public DateTime DateTime { get; set; }
      public Guid ExecutionId { get; set; }
      public string EventType { get; set; }
      public string Message { get; set; }
    }
  }
}

Normalize it

Dispatch rows from a flat file into several tables to normalize data thanks to the correlation mechanism.

private static void DefineProcess(ISingleStream<string> contextStream)
{
  var rowStream = contextStream
    .CrossApplyFolderFiles("list all required files", "*.csv", true)
    .CrossApplyTextFile("parse file", FlatFileDefinition.Create(i => new
    {
      Author = i.ToColumn("author"),
      Email = i.ToColumn("email"),
      TimeSpan = i.ToDateColumn("timestamp", "yyyyMMddHHmmss"),
      Category = i.ToColumn("category"),
      Link = i.ToColumn("link"),
      Post = i.ToColumn("post"),
      Title = i.ToColumn("title"),
    }).IsColumnSeparated(','))
    .SetForCorrelation("set correlation for row");

  var authorStream = rowStream
    .Distinct("remove author duplicates based on emails", i => i.Email)
    .Select("create author instance", i => new Author { Email = i.Email, Name = i.Author })
    .EfCoreSaveCorrelated("save or update authors", o => o
      .SeekOn(i => i.Email)
      .AlternativelySeekOn(i => i.Name));

  var categoryStream = rowStream
    .Distinct("remove category duplicates", i => i.Category)
    .Select("create category instance", i => new Category { Code = i.Category, Name = i.Category })
    .EfCoreSaveCorrelated("insert categories if doesn't exist, get it otherwise", o => o
      .SeekOn(i => i.Code)
      .DoNotUpdateIfExists());

  var postStream = rowStream
    .CorrelateToSingle("get related category", categoryStream, (l, r) => new { Row = l, Category = r })
    .CorrelateToSingle("get related author", authorStream, (l, r) => new { l.Row, l.Category, Author = r })
    .Select("create post instance", i => string.IsNullOrWhiteSpace(i.Row.Post)
      ? new LinkPost
      {
        AuthorId = i.Author.Id,
        CategoryId = i.Category.Id,
        DateTime = i.Row.TimeSpan,
        Title = i.Row.Title,
        Url = new Uri(i.Row.Link)
      } as Post
      : new TextPost
      {
        AuthorId = i.Author.Id,
        CategoryId = i.Category.Id,
        DateTime = i.Row.TimeSpan,
        Title = i.Row.Title,
        Text = i.Row.Post
      })
    .EfCoreSaveCorrelated("save or update posts", o => o
      .SeekOn(i => new { i.AuthorId, i.DateTime }));
}