Forem: Andreas Lennartz

High-Performance Bulk Operations for Dapper and ADO.NET

Andreas Lennartz — Mon, 26 Jan 2026 14:24:00 +0000

When working with ADO.NET and Dapper, developers often encounter a familiar limitation: efficient bulk write operations are not part of the core stack. Executing inserts, updates, or deletes row by row may be acceptable for small datasets, but performance degrades quickly as data volume grows. This is especially challenging when you need to merge data—combining inserts for new records and updates for existing ones in a single, efficient operation.

ETLBox DbExtensions addresses this gap by adding high-performance bulk operations directly to IDbConnection. It integrates cleanly with existing ADO.NET and Dapper workflows while keeping the API minimal and easy to adopt.

What Is DbExtensions?

ETLBox.DbExtensions is a lightweight extension library that enhances ADO.NET connections with bulk insert, update, delete, and merge capabilities.

Instead of introducing a new abstraction layer, DbExtensions works directly on IDbConnection, allowing developers to continue using their existing database access patterns while gaining efficient bulk write support.

Key Features

Bulk operations for ADO.NET
Works directly with IDbConnection
Designed for Dapper compatibility
Ideal companion for Dapper-based querying
Minimal setup
Install the package and start using it immediately

How DbExtensions Works

At its core, DbExtensions internally converts an ADO.NET connection into an ETLBox connection manager and uses ETLBox’s optimized bulk-operation infrastructure to execute these operations efficiently.

DbExtensions adds extension methods to IDbConnection. When a bulk operation is invoked:

The existing ADO.NET connection is reused
It is internally mapped to an ETLBox connection manager
ETLBox’s bulk operation logic is executed against the target database

This approach avoids duplicate connection management, keeps configuration minimal, and allows DbExtensions to support multiple database systems through ETLBox’s existing infrastructure.

Bulk Operations Overview

1. Bulk Insert

First, create the table using Dapper:

using Dapper;
using ETLBox;
using ETLBox.DbExtensions;
using Microsoft.Data.SqlClient;

using var connection = new SqlConnection(connectionString);

// Create table with Dapper
connection.Execute(@"
    IF NOT EXISTS (SELECT * FROM sys.tables WHERE name = 'Customer')
    CREATE TABLE Customer (
        Id INT PRIMARY KEY,
        Name NVARCHAR(100),
        City NVARCHAR(100)
    )");

Insert large datasets efficiently with a single call:

var customers = Enumerable.Range(1, 5_000)
    .Select(i => new Customer {
        Id = i,
        Name = $"Name {i}",
        City = $"City {i % 50}"
    });

connection.BulkInsert(customers);

public class Customer {
    public int Id { get; set; }
    public string Name { get; set; }
    public string City { get; set; }
}

For more control, you can configure batch processing, table prefixes, progress tracking, and read generated values:

connection.BulkInsert(customers, options => {
    options.BatchSize = 500;
    options.TablePrefix = "dim";
    options.ReadGeneratedValues = true;
    options.OnProgress = progress => {
        if (progress % 1000 == 0)
            Console.WriteLine($"Inserted {progress} rows.");
    };
});

2. Bulk Update

Update many rows at once by identifying the primary key. The [IdColumn] attribute must be added to the property that identifies the primary key:

var customers = Enumerable.Range(1, 2_500)
    .Select(i => new Customer {
        Id = i,
        Name = $"Updated {i}",
        City = $"City {i % 25}"
    });

connection.BulkUpdate(customers);

You can control which columns to update and add batch processing callbacks:

connection.BulkUpdate(customers, options => {
    options.BatchSize = 500;
    options.UpdateColumns = new[] { new UpdateColumn() { UpdatePropertyName = "Name" } };
    options.BeforeBatchWrite = (batch) => {
        Console.WriteLine($"Updating batch with {batch.Length} rows.");
        return batch;
    };
});

public class Customer {
    [IdColumn]
    public int Id { get; set; }
    public string Name { get; set; }
    public string City { get; set; }
}

3. Bulk Delete

Delete multiple rows efficiently by identifying records to remove:

var customers = Enumerable.Range(1000, 1000)
    .Select(i => new Customer { Id = i });

connection.BulkDelete(customers);

For more control, you can specify custom ID columns or add batch processing callbacks:

connection.BulkDelete(customers, options => {
    options.BatchSize = 500;
    options.IdColumns = new[] { new IdColumn() { IdPropertyName = "Name" } };
    options.BeforeBatchWrite = (batch) => {
        Console.WriteLine($"Deleting batch with {batch.Length} rows.");
        return batch;
    };
});

public class Customer {
    [IdColumn]
    public int Id { get; set; }
    public string Name { get; set; }
    public string City { get; set; }
}

4. Bulk Merge

Combine inserts and updates (and optional deletes) in a single operation. Bulk merge automatically inserts new records and updates existing ones based on the primary key.

Important: The Customer class must either inherit from MergeableRow or implement IMergeableRow for bulk merge operations. This is required because merge operations need to track which action was performed (insert, update, or delete) for each row, and these interfaces provide the necessary properties like ChangeAction and ChangeDate to support this functionality.

var customers = Enumerable.Range(1, 1_500)
    .Select(i => new Customer { Id = i, Name = $"Update Customer {i}", City = $"City {i % 50}" })
    .Union(
        Enumerable.Range(3_000, 1500)
        .Select(i => new Customer { Id = i, Name = $"New Customer {i}", City = $"City {i % 50}" })
    );

connection.BulkMerge(customers);

For advanced scenarios, you can control merge behavior with options:

connection.BulkMerge(customers, options => {
    options.MergeMode = MergeMode.Delta;
    options.CompareColumns = new[] { new CompareColumn() { ComparePropertyName = "City" } };
    options.UpdateColumns = new[] { new UpdateColumn() { UpdatePropertyName = "City" } };
    options.ReadGeneratedValues = true;
});

public class Customer : IMergeableRow {
    [IdColumn]
    public int Id { get; set; }
    public string Name { get; set; }
    public string City { get; set; }

    public DateTime ChangeDate { get; set; }
    public ChangeAction? ChangeAction { get; set; }
}

Best Practices: Using DbExtensions with Dapper

Complementary Responsibilities

Use Dapper for querying and DbExtensions for bulk writes:

using var connection = new SqlConnection(connectionString);

// Query with Dapper
var activeCustomers = connection
    .Query<Customer>("SELECT * FROM Customer WHERE Active = 1")
    .ToList();

// Bulk update with DbExtensions
var updatedCustomers = activeCustomers
    .Select(c => new Customer {
        Id = c.Id,
        Name = c.Name + " Updated"
    })
    .ToList();

connection.BulkUpdate(updatedCustomers);

ETL-Style Workflows

DbExtensions fits naturally into extract-transform-load scenarios:

// Extract
var sourceData = sourceConnection
    .Query<Customer>("SELECT * FROM SourceCustomers")
    .ToList();

// Transform
var transformed = sourceData
    .Select(c => new Customer {
        Id = c.Id,
        Name = c.Name.ToUpper()
    })
    .ToList();

// Load
targetConnection.BulkInsert(transformed);

Performance Considerations

Compared to row-by-row operations, DbExtensions significantly reduces database round-trips and add the missing bulk extensions to IDbConnection and Dapper. This makes it well-suited for:

Large imports/exports
Synchronization jobs
Batch processing workloads

Actual performance gains depend on the database system, schema, and environment, but the architectural advantage over per-row execution is substantial.

Getting Started

Install the ETLBox DbExtensions package:

dotnet add package ETLBox.DbExtensions

Additionally, you need to install the ETLBox package for your specific database system:

# For SQL Server
dotnet add package ETLBox.SqlServer

# For PostgreSQL
dotnet add package ETLBox.Postgres

# For MySQL
dotnet add package ETLBox.MySql

After installation, bulk operations are immediately available on any supported IDbConnection—no additional configuration required.

Sample Code

The sample code for the bulk operations demonstrated in this article, along with additional examples, can be found on GitHub:

Example code on Github

Conclusion

ETLBox DbExtensions bridges an important gap in the ADO.NET ecosystem by providing efficient bulk write operations without sacrificing simplicity. When combined with Dapper for querying, it enables a clean and pragmatic data-access strategy: lightweight reads, high-performance writes.

Whether you are building ETL processes, importing large files, or synchronizing data between systems, DbExtensions offers a focused, developer-friendly solution for bulk database operations in .NET.

Why CRUD Doesn’t Work for ETL (And What to Do Instead)

Andreas Lennartz — Fri, 14 Mar 2025 10:32:37 +0000

Traditional CRUD operations don’t scale well in ETL workflows. Handling data row by row creates unnecessary overhead, slowing down performance. This article explores how to transition from CRUD-based processing to a more efficient dataflow approach using ETLBox.

From CRUD to Dataflow: Optimizing ETL Performance with ETLBox

When working with ETL (Extract, Transform, Load) processes, many developers instinctively apply traditional CRUD (Create, Read, Update, Delete) operations to manipulate data. However, this approach often leads to severe performance bottlenecks when dealing with large datasets.

The fundamental problem? CRUD operations are designed for transactional systems, not for ETL workflows. While CRUD works well for handling small sets of records in OLTP (Online Transaction Processing) systems, it quickly becomes inefficient when applied to ETL processes, where performance and scalability are critical.

Instead of treating ETL like a series of individual transactions, a dataflow approach enables processing data in bulk, streaming large datasets efficiently, and optimizing database interactions. This shift eliminates performance killers such as single-row reads, updates, and inserts, replacing them with batch operations, transformations, and streaming techniques.

In this article, we’ll explore:

Why CRUD operations are inefficient in ETL workflows
How to replace CRUD patterns with optimized dataflow-based transformations
How ETLBox enables efficient ETL processing through bulk inserts, lookups, and merging strategies

By the end, you’ll have a clear understanding of how to convert CRUD operations into a performant ETL dataflow and ensure your data pipelines scale effectively.

About ETLBox

ETLBox is a flexible and high-performance ETL framework for .NET, designed to simplify data integration and processing. It supports control flow tasks, data transformations, bulk operations, and streaming dataflows, making it an efficient alternative to traditional row-by-row CRUD operations. With built-in support for multiple databases, ETLBox enables scalable and maintainable ETL workflows.

You can learn more at ETLBox.net.

Handling "Create" Operations

Create operations are not part of a dataflow. They typically happen before or after the main ETL process.

Before loading data, tables must exist in the destination. Instead of manually defining schemas, ETLBox provides ControlFlow Tasks to automate this step. CreateTableTask is one example of how schema creation can be handled dynamically across different database systems.

Preprocessing with CreateTableTask

CreateTableTask generates the correct SQL syntax based on the chosen database provider. To use it, you need a connection manager for the target database. For SQL Server, reference ETLBox.SqlServer and use SqlConnectionManager:

var connMan = new SqlConnectionManager("Data Source=server;Initial Catalog=db;Integrated Security=True");
List<TableColumn> columns = new List<TableColumn>() {
    new TableColumn("Id", "INT",allowNulls:false, isPrimaryKey:true),
    new TableColumn("Value", "VARCHAR(100)", allowNulls:true)
};
CreateTableTask.Create(connMan, "mytable", columns);

For PostgreSQL, PostgresConnectionManager from ETLBox.Postgres is required, and for MySQL, MySqlConnectionManager from ETLBox.MySql. The generated SQL is always tailored to the specific database. ETLBox supports are wide range of databases, including Oracle, MariaDb, Db2, and many more.

CreateTableTask is just one example of how the C (Create) in CRUD is handled outside the dataflow. Other tasks, such as CreateViewTask, CreateDatabaseTask, or CreateSchemaTask, allow creating views, databases, and schemas in a similar way. These tasks help automate the setup process so that the ETL flow can focus solely on data movement and transformation.

Postprocessing: Index Rebuilding

For large inserts, indexes are often dropped beforehand and recreated afterward to improve performance. This should also be done before the actual data flow:

var connMan = new SqlConnectionManager("Data Source=server;Initial Catalog=db;Integrated Security=True");
DropIndexTask.DropIfExists(connMan, "indexname", "mytable");
// Execute the data flow here
CreateIndexTask.CreateOrRecreate(connMan, "indexname", "mytable",
    new List<string>() { "index_column_1", "index_column_1" });

This ensures fast inserts while maintaining query performance after loading.

Avoiding Single-Row "Read" Operations

Traditional CRUD-based reads often involve executing a separate SELECT statement for each row that needs additional data. This approach works in transactional systems but is highly inefficient in ETL processes, where performance is key.

The Problem with Row-by-Row Reads

When processing large datasets, querying additional data for each row can quickly become a performance bottleneck.

Example of an Inefficient Approach

foreach (var row in data)
{
    var lookupValue = conn.ExecuteScalar<string>($"SELECT Name FROM ReferenceTable WHERE Id = {row.ReferenceId}");
    row.Name = lookupValue;
}

This results in one query per row, causing significant performance degradation due to repeated database round trips.

LookupTransformation: A Better Approach

Instead of executing multiple small queries, the LookupTransformation allows enriching data within the dataflow while minimizing database interactions. It fetches required reference data in bulk and caches it efficiently.

Using LookupTransformation

var connMan = new SqlConnectionManager("Data Source=server;Initial Catalog=db;Integrated Security=True");
var lookup = new LookupTransformation<InputRow, LookupRow>();
lookup.CacheMode = CacheMode.Full;
lookup.Source = new DbSource<LookupRow>(conn, "LookupTable");
lookup.ApplyRetrievedCacheToInput = (input, cache) => {
    input.Value = cache.List.FirstOrDefault(l => l.LookupId == input.Id)?.LookupValue;
    return input;
};

This transformation:

Loads all required reference data once the data flow starts
Uses cached lookup data to prevent single-row reads
Supports different cache modes (full cache, partial cache) depending on memory constraints

Choosing the Right Caching Strategy

Full Cache: Loads all lookup data into memory at once, ensuring the fastest lookups but requiring sufficient memory.
Partial Cache: Caches only the most frequently used values, reducing memory usage while still improving performance.

There are different eviction policies available which allow you to control how the cache is managed.

By replacing row-by-row reads with LookupTransformation, ETL workflows avoid the performance pitfalls of single-row queries while maintaining flexibility and scalability.

Using Dataflow Sources

Instead of querying additional data row by row, a better approach is to integrate database sources directly into the ETL dataflow. This avoids excessive queries and leverages efficient streaming and batch processing.

DbSource: Streaming Data Efficiently

DbSource allows you to load data from a database into a dataflow in a single operation. Instead of executing a separate SELECT for each row, it opens a stream on the query result, continuously passing retrieved data to subsequent transformations as it becomes available.

Example: Using DbSource to Read from a Database

var connMan = new SqlConnectionManager("Data Source=server;Initial Catalog=db;Integrated Security=True");
var source = new DbSource<MyDataRow>()
{
    ConnectionManager = connMan,
    TableName = "SourceTable"
};

This streams the entire table into the ETL pipeline, avoiding repeated queries.

Using Custom SQL for More Control

Instead of reading entire tables, DbSource can execute custom SQL queries to filter, join, or aggregate data before loading.

var connMan = new SqlConnectionManager("Data Source=server;Initial Catalog=db;Integrated Security=True");
var source = new DbSource<MyDataRow>(conn)
{
    ConnectionManager = connMan,
    Sql = "SELECT Id, Value FROM MyTable WHERE Active = 1 ORDER BY ChangeDate"
};

Integrating Stored Procedure Results

For more complex logic, DbSource can also use the result of a stored procedure as input.

var connMan = new SqlConnectionManager("Data Source=server;Initial Catalog=db;Integrated Security=True");
var source = new DbSource<MyDataRow>(conn)
{
    ConnectionManager = connMan,
    Sql = "EXEC GetFilteredData @Param1, @Param2"
};

This allows the pre-processing of data within the database, reducing unnecessary transformations inside the ETL process.

Combining Dataflows with DbSource

When working with multiple sources, you can integrate them into the same dataflow:

var source1 = new DbSource<MyDataRow>(conn1) { TableName = "TableA" };
var source2 = new DbSource<MyDataRow>(conn2) { TableName = "TableB" };

// Merge, filter, or join data from both sources within the dataflow
var join = new MergeJoin<MyLeftRow, MyRightRow, MyOutputRow>(
    (leftRow, rightRow) => {
        return new MyOutputRow() {
            TotalValue = leftRow.Value1 + " " + rightRow.Value2
        };
    });

source1.LinkTo(join.LeftInput);
source2.LinkTo(join.RightInput);

Using DbSource instead of single-row lookups ensures that data is retrieved in the most efficient way possible, reducing database load and improving ETL performance.

Efficient "Write" Operations: Bulk Inserts

Writing data efficiently is crucial for ETL performance. Row-by-row inserts or updates are extremely slow, as each operation requires a separate database transaction. Instead, ETLBox provides optimized dataflow components for high-performance inserts and merges.

Bulk Inserts with DbDestination

For large data loads, bulk inserts are the fastest way to write data. DbDestination enables batch processing, reducing database overhead.

Example: Using DbDestination for Bulk Inserts

var connMan = new SqlConnectionManager("Data Source=server;Initial Catalog=db;Integrated Security=True");
var destination = new DbDestination<MyDataRow>()
{
    ConnectionManager = connMan,
    TableName = "TargetTable"
};
destination.BatchSize = 1000;

The dataflow processes all incoming records in batches (using the given batch size), significantly improving performance compared to row-by-row inserts.

Merging or Upserts with DbMerge

When dealing with delta data (inserts, updates, and deletes), DbMerge provides an efficient alternative to manual INSERT, UPDATE, and DELETE statements. It behaves similarly to MERGE INTO or UPSERT, ensuring that new records are inserted, existing ones are updated, and obsolete records can be removed.

Example: Using DbMerge to Sync Delta Data

var connMan = new SqlConnectionManager("Data Source=server;Initial Catalog=db;Integrated Security=True");
var merge = new DbMerge<MyMergeRow>(conn, "DestinationTable") {
    ConnectionManager = conn,
    TableName = "DestinationTable",
    MergeMode = MergeMode.Full
};
merge.IdColumns = new[] {
    new IdColumn() { IdPropertyName = "Id" }
};

If a record exists, it is updated.
If a record does not exist, it is inserted.
If deletions need to be handled, DbMerge supports two modes:
- Full Mode: Detects and removes missing records automatically.
- Delta Mode: Deletes records based on an explicit delete flag in the incoming data.

The DbMerge also supports the merge modes InsertsAndUpdatesOnly and UpdatesOnly.

Choosing Between Bulk Insert and Merge

Scenario	Recommended Approach
Loading large datasets	`DbDestination` (bulk insert)
Synchronizing changes (upsert)	`DbMerge`
Handling deletions	`DbMerge` (full or delta mode)
Staging before final processing	`DbDestination`, followed by processing

For more details on the Merge operation, see Merge Overview. Or refer to this example of loading data into Star Schema.

By using DbDestination and DbMerge, ETL workflows remain scalable, avoiding slow row-by-row operations while maximizing performance.

Rethinking CRUD in ETL

Traditional CRUD operations are designed for transactional systems, while ETL processes handle large datasets more efficiently using a dataflow-based approach.

Row-by-row operations create high overhead and slow down performance. Each database interaction adds latency, making single-record transactions inefficient.

ETL processes work best when processing data in bulk. Bulk inserts replace individual INSERT statements, and DbMerge synchronizes changes more efficiently than repeated UPDATE or DELETE operations.

Operation	CRUD Approach	Dataflow Approach
Create	`CREATE TABLE` per request	Pre-processing with `CreateTableTask`
Read	Single `SELECT` per row	LookupTransformation or DbSource
Update	Single `UPDATE` per row	DbMerge for bulk updates
Delete	Single `DELETE` per row	DbMerge (Full or Delta Mode)

ETLBox provides efficient alternatives:

Schema preparation with CreateTableTask
Optimized reads using DbSource and LookupTransformation
Fast writes with DbDestination for bulk inserts and DbMerge for upserts and deletes

Moving from CRUD to dataflow enables scalable, high-performance ETL processes.

Mastering Database Merging: Comparing Different Approaches

Andreas Lennartz — Fri, 12 Jul 2024 14:58:40 +0000

In today’s world, managing data efficiently is crucial for businesses. One key task in data management is merging new data into existing database tables. Whether you’re adding new information or updating existing records, knowing how to do this properly can make a big difference. However, this task can be complex, and there are various approaches and paradigms you should be aware of.

Basic Example Scenario

Let's start with an example. Imagine you have a table called customers with the following columns:

customer_id
name
email

And the following data:

customer_id	name	email
1	John Doe	john.doe@example.com
2	Jane Smith	xx
3	Bob Brown	bobbrown@mail.com

You receive a new batch of data with these changes:

customer_id	name	email
1	John Doe	john.doe@example.com
2	Jane Smith	jane.smith@example.com
4	Alice Johnson	alice.j@example.com

Your goal is to merge this new data with the existing customers table. As a result of the merge operation, you want the incoming data to be integrated into the target table. This will result in one insertion of a new record (Alice Johnson), one update (email of Jane Smith), and one deletion (Bob Brown). The record for John Doe should be left untouched, as nothing has changed for this record.

Note that this example uses only a small number of rows. In real-world scenarios, you will need to deal with a very high volume of changes (e.g., up to millions per day). In the following sections, we will focus on solutions that can handle large volumes of data.

Using a Staging Table

First, we create and insert the existing data into the customers table:

CREATE TABLE customers (
    customer_id INT NOT NULL PRIMARY KEY,
    name VARCHAR(100),
    email VARCHAR(100)
);

INSERT INTO customers (customer_id, name, email)
VALUES
(1, 'John Doe', 'john.doe@example.com'),
(2, 'Jane Smith', 'xx'),
(3, 'Bob Brown', 'bobbrown@mail.com');

To simplify this example, we currently do not take into account how we receive this data. For now, we will insert the new data into a staging table:

CREATE TABLE staging_customers (
    customer_id INT,
    name VARCHAR(100),
    email VARCHAR(100)
);

INSERT INTO staging_customers (customer_id, name, email) VALUES
(1, 'John Doe', 'john.doe@example.com'),
(2, 'Jane Smith', 'jane.smith@example.com'),
(4, 'Alice Johnson', 'alice.j@example.com');

SQL Example for Inserts and Updates

You can use the MERGE statement to perform an Insert or Update (a.k.a. UPSERT) operation. The MERGE statement is supported by many databases like SQL Server, PostgreSQL, and Oracle. Note that this requires the data from your source to be loaded into a staging table first.

MERGE INTO customers AS target
USING staging_customers AS source
ON target.customer_id = source.customer_id
WHEN MATCHED THEN
    UPDATE SET name = source.name, email = source.email
WHEN NOT MATCHED THEN
    INSERT (customer_id, name, email)
    VALUES (source.customer_id, source.name, source.email);

Running this MERGE statement will automatically perform the necessary insert and update operations. Note that this process does not handle deletions. Keep in mind that the first record will be updated even if the input data matches the existing data in the destination table.

UPSERT Example for MySQL/MariaDB

In databases like MySQL or MariaDB that don't support the MERGE statement, you can use an INSERT ... ON DUPLICATE KEY UPDATE statement for upserts:

INSERT INTO customers (customer_id, name, email)
SELECT customer_id, name, email FROM staging_customers
ON DUPLICATE KEY UPDATE
    name = VALUES(name),
    email = VALUES(email);

Please note that customer_id is the primary key of the customers table. This is a prerequisite for the SQL above to work, as the primary key is automatically determined as the key column for the ON DUPLICATE KEY operation.

Handling Deletes

To handle deletions, you need to identify rows that exist in the target table but not in the source. This can be achieved using a WHERE ... NOT IN statement.

DELETE FROM customers
WHERE customer_id NOT IN (
    SELECT customer_id FROM staging_customers
);

As a result, the customers table will now look like our input data:

customer_id	name	email
1	John Doe	john.doe@example.com
2	Jane Smith	jane.smith@example.com
4	Alice Johnson	alice.j@example.com

Depending on your database, there may be more elegant solutions to solve this. For some databases like SQL Server, the MERGE statement does support the WHEN NOT MATCHED BY SOURCE THEN DELETE statement, which would also include the delete operation. Or you could use a LEFT JOIN instead of a WHERE ... NOT IN to determine the records for deletions, which could result in a shorter execution time.

Importing Data

In the previous example, we assumed that our data magically appeared in our staging table. But in real-world scenarios, we often have to deal with different kinds of data sources from which we are expected to retrieve our data. Let's examine how we can import the data into our database.

Using `BULK INSERT` for SQL Server

Databases are often limited when importing large amounts of data. By default, most databases offer some form of CSV import. When using the database approach, you are limited to CSV files and to a specific location of the files, making it complicated if your data can't be copied onto the database server. For SQL Server, you can import CSV file data using the BULK INSERT statement.

Here is an example of loading a CSV file using BULK INSERT in SQL Server:

BULK INSERT customers
FROM '\data\customers.csv'
WITH (
    FIELDTERMINATOR = ',',
    ROWTERMINATOR = '\n'
);

Data Cleansing

Besides the limitation to have data available in a CSV format, the examples above also rely on a cleansed staging table, where all data is error-free and can be merged into the destination table. The MERGE (or UPSERT) statements are ACID operations, meaning they are all-or-nothing: either all data is merged, or none is. If there's a flawed record in your data, the entire operation will stop, and identifying the flawed row can be challenging, as error messages typically don't specify which row(s) caused the issue.

To make this work, it is necessary that all data is cleaned before being loaded into the target table (e.g., all duplicates removed, etc.). Additionally, importing various data types (e.g., XML from a web service) can complicate the process. Therefore, it makes sense to clean and enrich the data before loading it into the staging table.

ETL Code with C

To load data from any source into the target table, you can use ETL data flows. ETL (Extract, Transform, Load) jobs are commonly used to clean and enrich data before loading it into the database. This approach allows you to define your own pipelines, through which your data is processed before it is written into your target table. Using an ETL approach, you can use any source and are not limited to CSV files.

A simple way to create an ETL pipeline with C# is to use the library ETLBox. Here is an example of ETLBox loading CSV data into the staging database table using CsvSource and DbDestination. After the pipeline has run, we can execute our MERGE statement.

ETL into Staging Table

When creating ETL with C#, you can load the cleansed data into the staging table and then run the MERGE statements, or you can directly use the DbMerge. Here is an example with the MERGE statement:

public class CustomerRow
{
    [ColumnMap("customer_id")]
    [IdColumn]
    public int CustomerId { get; set; }

    [ColumnMap("name")]
    public string Name { get; set; }

    [ColumnMap("email")]
    public string Email { get; set; }
}

// Data flow definition
var dbConnection = new SqlConnectionManager("..connection string here...");
var source = new CsvSource<CustomerRow>("data/customers.csv");
var cleanData = new RowTransformation<CustomerRow, CustomerRow>(row => {
    row.Email = row.Email.Trim().ToLower();// Custom data cleaning logic here
    return row;
});

var staging = new DbDestination<CustomerRow>(dbConnection, "staging_customers");

// Linking & executing the data flow
source.LinkTo(cleanData);
cleanData.LinkTo(staging);

Network.Execute(source);

// Running the MERGE operation
SqlTask.Execute(dbConnection, @"
MERGE INTO customers AS target
USING staging_customers AS source
ON target.customer_id = source.customer_id
WHEN MATCHED THEN
    UPDATE SET name = source.name,
               email = source.email
WHEN NOT MATCHED THEN
    INSERT (customer_id, name, email)
    VALUES (source.customer_id, source.name, source.email);
");

ETL with DbMerge

ETLBox already has a built-in DbMerge feature. Here is an example of loading data from a CsvSource, cleaning it using custom code in a RowTransformation, and then merging the cleaned data using DbMerge:

public class CustomerRow : MergeableRow
{
    [ColumnMap("customer_id")]
    [IdColumn]
    public int CustomerId { get; set; }

    [ColumnMap("name")]
    public string Name { get; set; }

    [ColumnMap("email")]
    public string Email { get; set; }
}

// Defining the data flow
var dbConnection = new SqlConnectionManager("..connection string here...");
var source = new CsvSource<CustomerRow>("data/customers.csv");
var cleanData = new RowTransformation<CustomerRow, CustomerRow>(row => {
    row.Email = row.Email.Trim().ToLower();// Custom data cleaning logic here
    return row;
});
var dbTypeCheck = new DbTypeCheck<CustomerRow>(dbConnection, "customers");
var merge = new DbMerge<CustomerRow>(dbConnection, "customers");
merge.BatchSize = 2000;
var typeErrors = new JsonDestination<CustomerRow>("type_errors.json");
var errorDest = new JsonDestination<ETLBoxError>("other_errors.json");

// Linking
source.LinkTo(cleanData);
cleanData.LinkTo(dbTypeCheck);
dbTypeCheck.LinkTo(merge);

// Error handling
dbTypeCheck.LinkFlawedTo(typeErrors);
merge.LinkErrorTo(errorDest);

// Executing the network
Network.Execute(source);

Please note that we no longer use the MERGE statement. The DbTypeCheck will verify all incoming data for compatibility with the target table before forwarding the rows to the target. Flawed rows will be written into the type_errors.json file. Even though we have cleaned and type-checked our data, there could still be issues when we (bulk) insert/update/delete our data in our destination. For performance reasons, this will happen in batches. Due to the ACID nature of the bulk operation, a batch can still fail - these failed batches are forwarded into other_errors.json. But instead of rejecting the whole batch (like in the previous MERGE statement), we now only reject a small batch of 2000 rows (defined in the BatchSize property).

Database Transactions

Of course, you could still spin up a database transaction for the whole operation when creating the connection:

var dbConnection = new SqlConnectionManager("..connection string here...");
dbConnection.BeginTransaction();

If an error occured, you could simply call dbConnection.Rollback() (e.g. in a try/catch block) and revert your changes.

Performance Improvements

Handling large amounts of data can make importing and merging complex. While the example provided involves only a few records, real-world scenarios often require dealing with vast amounts of data. Using a simple loop to detect each change and generate individual insert, update, or delete statements would be inefficient and result in poor performance. We need methods optimized for batch-processing large amounts of data.

Sophisticated methods such as batch processing, bulk operations, or specialized tools like ETL (Extract, Transform, Load) frameworks are necessary. These approaches handle large volumes of data efficiently by minimizing the number of database hits, grouping multiple operations into fewer transactions, and greatly enhancing performance and scalability.

Using Smaller Batches

The merge operation on the database is powerful, but if you have, for example, 1 million rows that need to be changed in an even larger table, the merge may become too slow. In such cases, it is better to process the incoming data in smaller batches. Using an ETL approach allows us to reduce the number of rows per batch (e.g., to 10,000 or 1,000 rows per operation), which reduces the load on the database and improves performance.

Full Load vs. Delta Load

Another important step to improve performance is to reduce the amount of data that actually needs to be loaded. When integrating data, you can either perform a full load or a delta load.

Full Load: Replaces all the data in the target table with the new data. This method is simple to implement and ensures data consistency, but it can be slow and resource-intensive, especially for large datasets.
Delta Load: Applies only the changes (inserts, updates, deletes) since the last load. This method is faster and less resource-intensive but more complex to implement because it requires tracking changes.

Switching from a full load to a delta load can significantly reduce the load time.

But how can a delta load be established with a data source? The following section will show examples of how this can be introduced to another database table as a data source, but similar strategies can be applied to other types of data sources as well.

Change Data Capture (CDC)

CDC is a technique used to identify changes in a database. It captures insert, update, and delete operations on tables and can be very useful for delta loads. When your source is a database, CDC can help by automatically tracking changes. Many databases support CDC, which can be set up to log changes in a separate table.

If you have CDC enabled on your source_customers table, you might have a change table like this:

customer_id	operation	name	email	change_time
1	U	John Doe	john.doe@example.com	2024-07-08 10:00:00
2	I	Jane Smith	jane.smith@example.com	2024-07-08 10:05:00
3	D	Alice Johnson	alice.j@example.com	2024-07-08 10:10:00

with I for Insert, U for Update and D for Delete.

You can use this change table to apply only the changes to your target table.

For example, we can create an ETL database source like this:

var lastLoadDate = DateTime.Now().AddDays(-1);
var source = new DbSource<CustomerRow>() {
    ConnectionManager = dbConnection,
    TableName = "source_customers",
    Sql = "SELECT customer_id, email, name FROM source_customers WHERE change_time > @lastLoadPar",
    SqlParameter = new[] {
        new QueryParameter("lastLoadPar","DATETIME",lastLoadDate)
    }
};

Handling Non-CDC Sources

If your source does not support CDC, you need other methods to track changes. One common approach is using a Last Modified Timestamp. You can compare this with the last load time to identify changes.

For example, our data might look like this when including a last_modified column:

customer_id	name	email	last_modified
1	John Doe	john.doe@example.com	2024-07-08 09:00:00
2	Jane Smith	jane.smith@example.com	2024-07-08 09:15:00
3	Alice Johnson	alice.j@example.com	2024-07-08 09:20:00

You can now select rows where last_modified is greater than the last load time to get the changes.

Truncate Target Table

In scenarios where you want to do a full load but have a large number of changes, it may be faster to truncate the target table first and then replace it with all data from the source. This approach can significantly reduce the time and resources needed to update the table.

Here is an example using ETLBox to truncate a table and reload data:

// Define the connection manager
var dbConnection = new SqlConnectionManager("..connection string here...");

// Truncate the target table
TruncateTableTask.Truncate(dbConnection, "customers");

// Define the data flow
var source = new CsvSource<CustomerRow>("data/customers.csv");
var cleanData = new RowTransformation<CustomerRow, CustomerRow>(row => {
    // Custom data cleaning logic here
    return row;
});
var destination = new DbDestination<CustomerRow>(dbConnection, "customers");

// Linking & executing the data flow
source.LinkTo(cleanData);
cleanData.LinkTo(destination);

Network.Execute(source);

Removing and Adding Indexes and Constraints

Additionally, it could be time-saving to remove any indexes / statistics / unique key constraints from the table before the insert operation, and then re-adding them after all data has been copied.

Enhancing the DbMerge

The built-in DbMerge of ETLBox supports various performance scenarios out of the box.

Using Delta with DbMerge

Both full and delta loads are supported by the DbMerge component of ETLBox. The following MergeModes are supported:

Delta: Does inserts and updates, deletions only with a flag
Full: Does inserts, updates, and deletions (deletions if the record is missing)
InsertsAndUpdatesOnly: Does inserts and updates only
UpdatesOnly: Only updates, no inserts or deletions

In the ETL code, you can adjust your data transfer class using different attributes to describe your id, update, and compare columns. For delta loads, you can also specify an attribute to indicate when a column should be deleted.

Here is an example setup of a data transfer class, that also uses the DeleteColumn attribute:

public class CustomerRow
{
    [ColumnMap("customer_id")]
    [IdColumn]
    public int CustomerId { get; set; }

    [ColumnMap("name")]
    [CompareColumn]
    [UpdateColumn]
    public string Name { get; set; }

    [UpdateColumn]
    [ColumnMap("email")]
    public string Email { get; set; }

    [DeleteColumn(true)]
    public bool IsDeletion { get;set; }
}

Cache Mode

By default, DbMerge will always load all data from the destination into memory first. If you want to avoid this, for example, because your target table is quite large, consider setting the CacheMode to CacheMode.Partial.

Setting the cache mode to partial will only load the data into memory for the currently processed batch. All other data will be removed from the cache. You can choose from different eviction policies that control how data is removed from the cache when it reaches its maximum size:

Least Recently Used (LRU): Evicts the least recently accessed items from the cache.
Least Frequently Used (LFU): Evicts the least frequently accessed items from the cache.
First In First Out (FIFO): Evicts the oldest items in the cache first.
Last In First Out (LIFO): Evicts the most recently added items first.

DbMerge<MyMergeRow> merge = new DbMerge<MyMergeRow>(connection, "DestinationTable");
merge.CacheMode = CacheMode.Partial;
merge.EvictionPolicy = CacheEvictionPolicy.LeastRecentlyUsed;

Truncating Target

Additionally, DbMerge has a setting UseTruncateMethod = true, which allows it to load the data into memory first, then truncate the table, and then insert the data while still detecting the changes. This method combines the benefits of truncating the table with the ability to track changes.

var merge = new DbMerge<CustomerRow>(dbConnection, "customers") {
    BatchSize = 2000,
    UseTruncateMethod = true
};

In this example, DbMerge collects merge information and sends it to another table (merge_info_table). The UseTruncateMethod setting ensures that data is loaded into memory, the target table is truncated, and then the data is inserted, improving performance while still detecting changes.

Delta Table

The DbMerge has a property DeltaTable which is a list containing information about what records were updated, inserted, or deleted. The operation and change date are stored in the corresponding ChangeDate and ChangeAction properties.

For example, the merge_info_table could have the following structure:

CREATE TABLE merge_info_table (
    customer_id INT,
    operation CHAR(1),
    change_time DATETIME
);

This table will store the customer_id of the affected rows, the type of operation (Insert, Update, or Delete), and the timestamp of the change.

This delta table can be accessed if the DbMerge is not treated as a source but as a transformation. If the DbMerge is linked to other components, it will write the delta records into its output.

In this example, you can send the delta information further down in the data flow, for example, to another database table.

// Define the connection manager
var dbConnection = new SqlConnectionManager("..connection string here...");
// Define the data flow
var source = new CsvSource<CustomerRow>("data/customers.csv");
var dbTypeCheck = new DbTypeCheck<CustomerRow>(dbConnection, "customers");
var merge = new DbMerge<CustomerRow>(dbConnection, "customers") {
    BatchSize = 2000,
    CacheMode = CacheMode.Partial,
    EvictionPolicy = CacheEvictionPolicy.LeastRecentlyUsed,
    MergeMode = MergeMode.Full
};
var mergeInfo = new DbDestination<MergeInfoRow>("merge_info_table", dbConnection);

// Linking
source.LinkTo(dbTypeCheck);
dbTypeCheck.LinkTo(merge);

// Collecting merge information
merge.LinkTo(mergeInfo);

// Executing the network
Network.Execute(source);

Going Further: Building Dynamic Data Flows

The above examples for the DbMerge make use of strongly typed objects (POCOs). However, the DbMerge also supports the use of the dynamic ExpandoObject. When using the ExpandoObject, you don't have to create a strongly typed object beforehand. The corresponding properties are created on-the-fly whenever a value is assigned.

This feature is very powerful. Imagine you want to write integration code for hundreds of tables, perhaps even from different sources. Using the dynamic object approach lets you easily write a data flow that can handle different tables and data types without needing to define a separate POCO for each table. The target table could also be created dynamically if needed.

Example Use Case

Consider a scenario where you need to integrate data from two different CSV files, each representing a different entity (customers and orders), but having an overlapping property (eMail) that should be cleaned the same way. Instead of defining a POCO for each entity, you can use ExpandoObject to dynamically handle the data.

using System.Dynamic;
using ETLBox.DataFlow;
using ETLBox.Connection;

// Define the connection manager
void RunMerge(string entity) {
    var dbConnection = new SqlConnectionManager("..connection string here...");
    var source = new CsvSource<ExpandoObject>($"data/{entity}.csv");

    // Data transformation
    var cleanData = new RowTransformation<ExpandoObject, ExpandoObject>(row => {
        // Data cleaning logic for both entities
        dynamic dynamicRow = row;
        dynamicRow.eMail = dynamicRow.eMail?.Trim().ToLower();
        return dynamicRow;
    });

    // Define DbMerge dynamically
    var merge = new DbMerge<ExpandoObject>() {
        ConnectionManager = dbConnection,
        TableName = entity + "s",
        BatchSize = 2000,
        MergeMode = MergeMode.Full
    };

    // Linking and executing the data flow
    source.LinkTo(cleanData);
    cleanData.LinkTo(merge);
    Network.Execute(source);
}

You could further enhance the dynamic nature of your data flow by creating target tables on the fly based on the schema of the incoming data. This can be particularly useful in scenarios where the schema is not known beforehand. An introduction to creating dynamic data flows like this will be part of another article.

Conclusion

Merging data is a key skill that helps keep your business information accurate and useful. Whether you're working with a few records or millions of them, using the right methods can save time, reduce mistakes, and make your processes smoother. Tools like ETLBox can make this even easier by helping you clean, transform, and load your data quickly and efficiently. By using these techniques and tools, you can ensure your data is always up-to-date and your business stays ahead in the data-driven world.

Unlock the Power of C# in Polyglot Notebooks

Andreas Lennartz — Tue, 02 Jul 2024 15:46:11 +0000

This article shows how to use C# skills in polyglot notebooks for better data management and analysis.

C# and Polyglot Notebooks

C# is a powerful programming language widely used for building a variety of applications, from web apps to complex data processing systems. Polyglot notebooks, like Jupyter notebooks, allow you to write and execute code in multiple programming languages within the same document. This makes it easier to share and collaborate on projects, especially those involving data analysis, visualization, or machine learning.

Traditionally, C# hasn't been as common in the realm of polyglot notebooks, which have been more associated with languages like Python and R. However, Microsoft has made significant improvements in recent years, making it much easier to use C# within these environments. With these advancements, we can now leverage the full power of C# for tasks that require high performance and robust tooling.

The Use-case of ETL in a Polyglot Notebook

ETL stands for Extract, Transform, Load, and it's a process commonly used in data management to move data from one place to another, transform it into a usable format, and then load it into a database or another storage system. In polyglot notebooks, basic tasks like loading data from a CSV file are supported out-of-the-box. However, when dealing with more complex data sources such as databases or web APIs, and then transforming this data, ETL pipelines become indispensable.

We are using ETLBox, a library designed for ETL processes in .NET environments, because it allows us to write ETL code in C#. By using a polyglot notebook, we can combine the flexibility of different programming languages with the power of C# for data manipulation.

How to Create Your Own Polyglot Notebook

To create your own polyglot notebook in VS Code, follow these steps:

Prerequisites: Make sure you have Visual Studio Code (VS Code) installed on your computer. You will also need to install the Polyglot Notebooks extension to enable multi-language support within your notebook.
Press CTRL + SHIFT + P to open the command dialogue.
Select Polyglot Notebook: Create Default Notebook.
Choose the .ipynb extension and select the C# language.
Add the following code blocks to start coding in C#:



   // This is where your C# code will go
   var s = "Hello, Polyglot Notebook!";
   s

(Note the missing ; at the end of the last line)

Run the code: This should print Hello, Polyglot Notebook! as output.

These steps will set up a fresh new notebook where you can start exploring using Polyglot Notebooks with C#.

Setting Up a Database Table for ETL Using C# in a Polyglot Notebook

Let's see how to set up a simple database table using C# within a polyglot notebook.

Install ETLBox.SqlServer Package

Before we begin, ensure you have the ETLBox.SqlServer package installed. This package provides the necessary tools to interact with SQL Server databases using ETLBox.



   #r "nuget:ETLBox.SqlServer, 3.4.0"

Set Up Connection Credentials



   using ETLBox;
   using ETLBox.SqlServer;
   using ETLBox.ControlFlow;

   IConnectionManager connMan = new SqlConnectionManager("Data Source=localhost;User Id=sa;Password=YourStrong@Passw0rd;Initial Catalog=demo;TrustServerCertificate=true;");

Replace the connection string in the code below with your SQL Server credentials. This example assumes you are connecting to a local SQL Server instance (localhost), using the sa user with the password YourStrong@Passw0rd, and targeting a database named demo.

Define Table Structure

We'll define a simple table named Test with three columns: Id (INT, identity column), XValue (DATETIME), and YValue (INT). This table will store sample data for demonstration purposes.



   var def = new TableDefinition("Test", new List<TableColumn>() {
       new TableColumn("Id", "INT", allowNulls: false, isIdentity:true, isPrimaryKey:true),
       new TableColumn("XValue", "DATETIME", allowNulls: false),
       new TableColumn("YValue", "INT", allowNulls: false)
   });

Create the Table

Before creating the table, ensure any existing table with the same name is dropped to start fresh. This step is handled by the DropTableTask.DropIfExists method.



   DropTableTask.DropIfExists(connMan, "Test");
   CreateTableTask.CreateIfNotExists(connMan, def);

Insert Sample Data

Populate the Test table with sample data using SQL INSERT statements.



   SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Test VALUES('2022-01-01',100)");
   SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Test VALUES('2022-01-02',350)");
   SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Test VALUES('2022-01-03',470)");
   SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Test VALUES('2022-01-04',134)");
   SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Test VALUES('2022-01-05',42)");

Execute the code

After executing the code, check your SQL Server database (demo in this example) to verify that the Test table has been created and populated with the sample data.

Loading Data using an ETL Pipeline

Let's see how to load data from a SQL Server database into memory and analyze it using DataFrames in a polyglot notebook environment.

Setup Environment

Ensure you have the necessary libraries installed and configured. We are using ETLBox for ETL operations, including data extraction from SQL Server, and Microsoft's DataFrame for data analysis.



   #r "nuget:ETLBox, 3.4.0"
   #r "nuget:ETLBox.SqlServer, 3.4.0"
   #r "nuget:ETLBox.Analysis, 3.4.0"

   using ETLBox;
   using ETLBox.ControlFlow;
   using ETLBox.SqlServer;
   using ETLBox.DataFlow;
   using ETLBox.Analysis;

Loading Data from Database

Initialize the connection manager and define a DbSource to extract data from the Test table in our SQL Server database.



   //The connMan was already defined in a previous step!
   //IConnectionManager connMan = new SqlConnectionManager("Data Source=localhost;User Id=sa;Password=YourStrong@Passw0rd;Initial Catalog=demo;TrustServerCertificate=true;");
   var source = new DbSource(connMan, "Test");

Transforming Data

Use a RowTransformation to manipulate the data in memory. In this example, we multiply the YValue column by 1000.



   var row = new RowTransformation(row => {
       dynamic r = row as dynamic;
       r.YValue = r.YValue * 1000;
       return row;
   });

Multicasting Data

Utilize a Multicast component to split the data flow into two branches:

One branch feeds data into a MemoryDestination, storing data in a C# List (memDest.Data).
The other branch stores data into a DataFrameDestination, creating a DataFrame (dfDest.DataFrame) for further analysis.



   var multicast = new Multicast();
   var memDest = new MemoryDestination();
   var dfDest = new DataFrameDestination();
   multicast.OnProgress = pc => Console.WriteLine($"Records loaded from database: {pc}");

Running the Pipeline

Until now, we've defined our pipeline components. Now, let's link them together and trigger the data flow.



   source.LinkTo(row);
   row.LinkTo(multicast);
   multicast.LinkTo(memDest);
   multicast.LinkTo(dfDest);

   Network.Execute(source);

Accessing Loaded Data

We stored the data loaded into the MemoryDestination (memDest.Data) and the DataFrame from DataFrameDestination (dfDest.DataFrame) for further analysis and manipulation in variables in our notebook.



   var data = memDest.Data; // This is a List<object[]>
   var df = dfDest.DataFrame; // This is a Microsoft.Analysis.DataFrame object

Displaying and Analyzing Data

Let's see how we can display and analyze the data loaded into tjhe polyglot notebook.

Displaying the DataFrame

To begin with, let's display the contents of the DataFrame (df). This DataFrame contains the transformed data from our ETL pipeline.



// Displaying the DataFrame
df

We will get an output like this:

index	Id	XValue	YValue
0	1	2022-01-01 00:00:00Z	100000
1	2	2022-01-02 00:00:00Z	350000
2	3	2022-01-03 00:00:00Z	470000
3	4	2022-01-04 00:00:00Z	134000
4	5	2022-01-05 00:00:00Z	42000

Transforming the DataFrame

We can perform additional transformations directly on the DataFrame. In this example, we'll create a new column x by doubling the values in the YValue column.



// Transforming the DataFrame
df["x"] = df["YValue"] * 2;
df

Now our output becomes this:

index	Id	XValue	YValue
0	1	2022-01-01 00:00:00Z	100000
1	2	2022-01-02 00:00:00Z	350000
2	3	2022-01-03 00:00:00Z	470000
3	4	2022-01-04 00:00:00Z	134000
4	5	2022-01-05 00:00:00Z	42000

The Microsoft.Analysis.DataFrame is highly effective as an in-memory structure for data analysis and querying.

You can learn more about it here in the Microsft Documentation: Getting started with DataFrames

Visualizing Data with ScottPlot

Next, we'll visualize the loaded data using ScottPlot, a plotting library for .NET.



#r "nuget:ScottPlot, 4.1.69"

using Microsoft.DotNet.Interactive.Formatting;

// Setup a custom formatter to display plots as images
Formatter.Register(typeof(ScottPlot.Plot), (plot, writer) =>
    writer.Write(((ScottPlot.Plot)plot).GetImageHtml()), HtmlFormatter.MimeType);

var plt = new ScottPlot.Plot(600, 400);

// Extract data from memDest.Data for plotting
var dataX = memDest.Data.Select(row => (DateTime)(row as dynamic).XValue)
                        .Select(dt => dt.ToOADate()).ToArray();
var dataY = memDest.Data.Select(row => (double)(row as dynamic).YValue).ToArray();

// Add scatter plot to ScottPlot
plt.XAxis.DateTimeFormat(true); // Enable DateTime formatting on X-axis if applicable
plt.AddScatter(dataX, dataY);

plt

This will print out our data as a plot:

Final Thoughts

Using C# in polyglot notebooks allows data professionals to work without needing to learn a new programming language. Polyglot notebooks enable users to use their existing knowledge about C# and the .NET framework in one environment, enhanced by the power of Jupyter notebooks.

By default, loading data using existing data frameworks only support loading basic files (like CSV). This makes the inclusion of ETL processes in notebooks essential for managing more complex data tasks. ETLBox, tailored for C#, stands out as the best framework designed specifically for this purpose.

Together, C#, Polyglot notebooks and ETLBox simplify data processing, supporting data professional to work more efficiently and make better decisions based on their data.

Code on Github

The notebook code for this example is available on GitHub for further exploration and contribution.

Building a Data Warehouse with ETLBox: A .NET Developer's Guide

Andreas Lennartz — Fri, 10 Nov 2023 13:46:37 +0000

Introduction

In the world of data, transformation is key. But not just any transformation—it's about transforming raw data into actionable insights. That's where ETL (Extract, Transform, Load) processes shine, and ETLBox is a .NET star in this domain. ETLBox offers a set of tools and components to build scalable and efficient ETL processes with minimal hassle. Whether you're a seasoned data engineer or a developer venturing into data warehousing, ETLBox could be your toolkit of choice.

Data Warehousing Concepts

Before we start to dive into a practical example, let's learn more about the basic building blocks of a data warehouse (DWH).

Advantages of Using a DWH & Star Schema

A data warehouse serves as a pivotal element in the modern data ecosystem, acting as a centralized data repository where information from various sources is consolidated. This consolidation enables improved decision-making by providing access to historical data critical for thorough analysis and forecasting. One of the key features of a data warehouse is the enhancement of data quality and consistency, achieved through meticulous data cleansing and transformation processes. Data warehouses are particularly optimized for high-performance querying, handling complex queries efficiently and thus allowing for significant volumes of data to be processed. Moreover, they ensure the operational systems' performance is unaffected by offloading analytical queries.

The star schema plays a vital role within the data warehouse architecture. Its design is simple, centering around a fact table which is directly linked to a set of di mension tables. This simplicity translates to enhanced query performance due to fewer joins and predictable query patterns, which also simplifies the ETL process and facilitates easier data management.

Understanding Surrogate IDs

In data warehousing, a Surrogate ID (SID) is a unique identifier for each row in a dimension table, which is unrelated to any natural key or business identifier. This concept is essential in ensuring that dimension records are uniquely identifiable, simplifying the handling of slowly changing dimensions (SCDs), and maintaining historical accuracy in your data warehouse.

In our example, we employ the use of the IDENTITY keyword in SQL Server to automatically generate these surrogate IDs. This feature is equivalent to AUTO_INCREMENT in MySQL or SERIAL in PostgreSQL. When a new record is inserted into a dimension table, the database engine auto-generates a sequential numeric value for the SID. This automated process alleviates the need for manual ID management and ensures that each new record can be tracked with a unique identifier throughout the lifecycle of the data warehouse.

Here’s a snippet demonstrating the use of IDENTITY within the context of our dimension tables:

TableDefinition CustomerDimTableDef = new TableDefinition("DimCustomer",
    new List<TableColumn>() {
        new TableColumn("DimId", "INT", allowNulls: false, isPrimaryKey:true, isIdentity: true),
        // Other columns...
});

In the above table definition, DimId is the surrogate ID column. Its value is automatically incremented for each new record, starting with 1 and increasing by 1 for each new insert, thereby ensuring that each customer dimension record has a unique identifier.

Dimensions and Facts

SCD Type 1

Slowly Changing Dimension (SCD) Type 1 strategy addresses the need to manage and reflect changes in dimension attributes by overwriting the existing records. This approach is best suited for situations where maintaining historical changes is not necessary, such as error corrections or updates that do not require tracking the attribute history. SCD Type 1 is relatively straightforward to implement and maintains the dimension table's size by avoiding redundancy.

SCD Type 2

In contrast to Type 1, SCD Type 2 is designed to preserve historical changes. Whenever a change occurs in a dimension attribute, a new record is added to the dimension table, with the original record remaining intact. This approach allows the history of dimension changes to be tracked over time, with each record typically including 'Valid From' and 'Valid To' dates to define the period of validity. SCD Type 2 is essential for businesses that require an audit trail or historical reporting.

Fact Tables

The fact table is the cornerstone of the star schema within a data warehouse. It links all dimension tables together and stores quantitative metrics that are the focus of analysis. Fact tables can contain various types of facts, such as additive (can be summed up), semi-additive (can be aggregated in some dimensions), and non-additive (cannot be aggregated). The granularity of the fact table is a crucial consideration; it defines the level of detail stored and determines the depth of analysis possible.

Special: Date Dimension

A unique aspect of data warehousing is the use of a date dimension, which is crucial for any time-based analysis. This dimension allows for a consistent time-based reference across all fact tables, supporting the analysis across various time hierarchies like days, weeks, and months. The date dimension often includes attributes beyond the standard calendar dates, such as fiscal periods, holidays, and special event flags, which greatly enhance the data warehouse's analytical capabilities.

Implementing a DWH with ETLBox

By mastering these data warehousing concepts and understanding the structure and function of elements like the star schema, surrogate IDs, SCDs, and fact tables, organizations can extract substantial value from their data, paving the way for informed decision-making and effective strategic planning. As this knowledge forms the foundation, the next step is to bring these concepts to life within the data warehousing environment. That's where ETLBox comes into play—a powerful .NET library designed to simplify the extract, transform, and load processes. In subsequent sections, we will delve into how ETLBox can be used to streamline these operations, ensuring your data warehousing efforts are as efficient and effective as possible.

Preparing the Database

The first step is to set up your OLTP (Online Transaction Processing) databases and the DWH database. ETLBox can assist with this by enabling you to programmatically create databases, tables, and other database objects. It simplifies the process of data preparation by executing SQL tasks that can create tables and insert demo data for initial testing.

In our example, we will be using SQL Server. To ensure a smooth and isolated development environment, we recommend setting up a local Docker container running SQL Server. This approach offers several benefits, including easy setup, consistency across environments, and the ability to replicate or destroy environments without affecting your local system. For detailed instructions on setting up SQL Server in a Docker container, refer to Microsoft's comprehensive guide.

To kickstart the database setup, we create a helper method that efficiently recreates the databases whenever needed. This method is especially useful in development and testing environments where frequent resetting of the database state is common. Here’s how it’s implemented:

public static void RecreateDatabase(string dbName, SqlConnectionString connectionString)
{
    var masterConnection = new SqlConnectionManager(connectionString.CloneWithMasterDbName());
    DropDatabaseTask.DropIfExists(masterConnection, dbName);
    CreateDatabaseTask.Create(masterConnection, dbName);
}

Prepare Connection Managers

Before we proceed to execution, we need to create the ETLBox connection managers that wrap the underlying ADO.NET connection.

static string OLTP_ConnectionString = @"Data Source=localhost;Initial Catalog=OLTP_DB;Integrated Security=false;User=sa;password=YourStrong@Passw0rd;TrustServerCertificate=true";
static string OLAP_ConnectionString = @"Data Source=localhost;Initial Catalog=OLAP_DB;Integrated Security=false;User=sa;password=YourStrong@Passw0rd;TrustServerCertificate=true";
static SqlConnectionManager OLTP_Connection => new SqlConnectionManager(OLTP_ConnectionString);
static SqlConnectionManager OLAP_Connection => new SqlConnectionManager(OLAP_ConnectionString);

Security Note on Connection Strings:

When implementing these examples in a production environment, it's crucial to secure your database credentials. Avoid hardcoding passwords in your codebase. Instead, use environment variables, Azure Key Vault, or other secure methods to manage your credentials.

Following this, we proceed to create the databases that will act as our OLTP system and our OLAP data warehouse:

Console.WriteLine("(Re)Creating demo databases 'OLTP_DB' & 'OLAP_DB'");
PrepareDb.RecreateDatabase("OLTP_DB", OLTP_ConnectionString);
PrepareDb.RecreateDatabase("OLAP_DB", OLTP_ConnectionString);

OLTP Demo Tables

Next, we establish the OLTP database tables. The code snippet below lays out the table definitions and populates them with sample data.

public static void CreateOLTPTables(SqlConnectionString connectionString)
{
    var connMan = new SqlConnectionManager(connectionString);

    TableDefinition OrderDataTableDef = new TableDefinition("Orders",
        new List<TableColumn>() {
            new TableColumn("OrderId", "INT",allowNulls: false, isPrimaryKey:true),
            new TableColumn("OrderDate","DATETIME", allowNulls: false),
            new TableColumn("ProductNumber","CHAR(7)", allowNulls: false),
            new TableColumn("CustomerNumber","CHAR(6)", allowNulls: false),
            new TableColumn("ActualPrice","MONEY", allowNulls: false),
    });

    TableDefinition CustomerTableDef = new TableDefinition("Customer",
        new List<TableColumn>() {
            new TableColumn("CustomerNumber", "CHAR(6)",allowNulls: false, isPrimaryKey:true),
            new TableColumn("Name","NVARCHAR(200)", allowNulls: false),
    });

    TableDefinition ProductTableDef = new TableDefinition("Product",
    new List<TableColumn>() {
            new TableColumn("ProductNumber", "CHAR(7)",allowNulls: false, isPrimaryKey:true),
            new TableColumn("Name", "NVARCHAR(50)",allowNulls: false),
            new TableColumn("Description", "NVARCHAR(500)",allowNulls: true),
            new TableColumn("RecommendedPrice","MONEY", allowNulls: false),
            new TableColumn("LastUpdated","DATETIME", allowNulls: false)

    });
    //Create demo tables & fill with demo data
    OrderDataTableDef.CreateTable(connMan) ;
    CustomerTableDef.CreateTable(connMan);
    ProductTableDef.CreateTable(connMan);            
    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Customer VALUES('C-1000', 'Kevin Doe')");
    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Customer VALUES('C-1001','Nick Newman')");
    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Customer VALUES('C-1002','Zoe Trunk')");

    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Product VALUES('P-00010','Smartphone', 'The newest phone',399, '2023-01-01')");
    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Product VALUES('P-00011','uPhone', 'Same as a smartphone, but double the price',799,'2023-01-01')");
    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Product VALUES('P-00012','Computer', 'A brand new desktop',899,'2023-01-01')");
    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Product VALUES('P-00013','Notebook', NULL,1599,'2023-01-01')");

    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Orders VALUES(10003,'2023-01-01', 'P-00010', 'C-1000', 379)");
    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Orders VALUES(10007,'2023-01-02', 'P-00011', 'C-1000', 699)");
    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Orders VALUES(10012,'2023-01-03', 'P-00012', 'C-1000', 849)");
    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Orders VALUES(10016,'2023-01-01', 'P-00012', 'C-1001', 949)");
    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Orders VALUES(10020,'2023-01-02', 'P-00011', 'C-1001', 849)");
    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Orders VALUES(10033,'2023-01-03', 'P-00013', 'C-1001', 1699)");
    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Orders VALUES(10053,'2023-01-01', 'P-00010', 'C-1002', 299)");
    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Orders VALUES(10193,'2023-01-01', 'P-00011', 'C-1002', 699)");
    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Orders VALUES(10253,'2023-01-03', 'P-00012', 'C-1002', 799)");
    SqlTask.ExecuteNonQuery(connMan, "INSERT INTO Orders VALUES(10323,'2023-01-03', 'P-00013', 'C-1002', 1299)");

}

The tables are schematically outlined with sample data as follows:

CustomerNumber	Name
C-1000	Kevin Doe
C-1001	Nick Newman
C-1002	Zoe Trunk

ProductNumber	Name	Description	RecommendedPrice	LastUpdated
P-00010	Smartphone	New phone	399.0000	2023-01-01
P-00011	uPhone	Another phone	799.0000	2023-01-01
P-00012	Computer	A computer	899.0000	2023-01-01
P-00013	Notebook		1599.0000	2023-01-01

OrderId	OrderDate	ProductNumber	CustomerNumber	ActualPrice
10003	2023-01-01	P-00010	C-1000	379.00
10007	2023-01-02	P-00011	C-1000	699.00
10012	2023-01-03	P-00012	C-1000	849.00
10016	2023-01-01	P-00012	C-1001	949.00
10020	2023-01-02	P-00011	C-1001	849.00
10033	2023-01-03	P-00013	C-1001	1699.00
10053	2023-01-01	P-00010	C-1002	299.00
10193	2023-01-01	P-00011	C-1002	699.00
10253	2023-01-03	P-00012	C-1002	799.00
10323	2023-01-03	P-00013	C-1002	1299.00

We can now run the OLTP database preparation script:

Console.WriteLine("Create OLTP tables");
PrepareDb.CreateOLTPTables(OLTP_ConnectionString);

OLAP Demo Tables

To define our star schema for OLAP (Online Analytical Processing), we create the following tables:

public static void CreateStarSchema(SqlConnectionString connectionString) {
    var connMan = new SqlConnectionManager(connectionString);

    TableDefinition OrderFactTableDef = new TableDefinition("FactOrders",
        new List<TableColumn>() {
            new TableColumn("FactId", "INT",allowNulls: false, isPrimaryKey:true, isIdentity:true),
            new TableColumn("SourceOrderId", "INT",allowNulls: false),
            new TableColumn("DateDim","INT", allowNulls: false),
            new TableColumn("ProductDim","INT", allowNulls: false),
            new TableColumn("CustomerDim","INT", allowNulls: false),                    
            new TableColumn("ActualPriceFact","MONEY", allowNulls: false),
    });

    TableDefinition CustomerDimTableDef = new TableDefinition("DimCustomer",
        new List<TableColumn>() {
            new TableColumn("DimId", "INT",allowNulls: false, isPrimaryKey:true, isIdentity: true),
            new TableColumn("CustomerNumber", "NVARCHAR(10)",allowNulls: false),
            new TableColumn("Name","NVARCHAR(200)", allowNulls: true),
    });

    TableDefinition ProductDimTableDef = new TableDefinition("DimProduct",
       new List<TableColumn>() {
           new TableColumn("DimId", "INT",allowNulls: false, isPrimaryKey:true, isIdentity: true),
            new TableColumn("ProductNumber", "NVARCHAR(10)",allowNulls: false),
            new TableColumn("Name", "NVARCHAR(50)",allowNulls: true),
            new TableColumn("Description", "NVARCHAR(500)",allowNulls: true),
            new TableColumn("RecommendedPrice","MONEY", allowNulls: false),
            new TableColumn("ValidFrom","DATETIME", allowNulls: false, isPrimaryKey:true),
            new TableColumn("ValidTo","DATETIME", allowNulls: false),
    });

    //Create demo tables & fill with demo data
    OrderFactTableDef.CreateTable(connMan);
    CustomerDimTableDef.CreateTable(connMan);
    ProductDimTableDef.CreateTable(connMan);          
}

This is the schematic outline of the tables:

DimId	CustomerNumber	Name
1	C-XXXX	Name

DimId	ProductNumber	Name	Description	RecommendedPrice	ValidFrom	ValidTo
1	P-XXXXX	Name	Description	99.00	1900-01-01	9999-12-31

FactId	SourceOrderId	DateDim	ProductDim	CustomerDim	ActualPriceFact
1	10001	20230101	1	1	79.00

Let's also create the tables for our OLAP database:

Console.WriteLine("Create OLAP tables");
PrepareDb.CreateStarSchema(OLAP_ConnectionString);

Loading SCD Type 1 Dimensions

When dealing with SCD (Slowly Changing Dimension) Type 1 in data warehousing scenarios, ETLBox streamlines the extract, transform, and load (ETL) process. The SCD Type 1 method involves updating existing records with new data as it comes in, without maintaining historical changes. This means that any updates in the dimension data will overwrite the previous entries, thus always reflecting the most current state of data.

ETLBox helps with these operations with its dataflow tasks. Generating surrogate keys is simplified using SQL Server's Identity column feature, which is a common practice to uniquely identify records in a DWH environment.

In the context of ETLBox, to decide whether a source row should be inserted as new or used to update an existing record, the DbMerge task is utilized. By setting the MergeMode to InsertsAndUpdates, ETLBox handles the decision-making process of the merge operation, which is particularly useful since deletions are generally not performed in a DWH in order to maintain historical data consistency.

Here's how we implement an SCD Type 1 load using ETLBox:

We define a strongly-typed object Customer, which helps map source data to the destination schema, including any necessary identifiers.
DbSource is used to load all data from the Customer table in the OLTP source system.
DbMerge is then configured to sync this incoming data with the destination table in the DWH by either inserting new records or updating existing ones based on the primary key.

Below is the C# code for loading SCD Type 1 dimensions:

public class Customer : MergeableRow
{            
    public string Name { get; set; }
    [IdColumn]
    [DbColumnMap("CustomerNumber")]
    public string Number { get; set; }
    public int? DimId { get; set; }
}

private static void LoadCustomer_SCD1() {
    var source = new DbSource<Customer>(OLTP_Connection, "Customer");

    var dest = new DbMerge<Customer>(OLAP_Connection, "DimCustomer");
    dest.MergeMode = MergeMode.InsertsAndUpdates;

    source.LinkTo(dest);
    Network.Execute(source);
}

private static void ChangeCustomer() {
    SqlTask.ExecuteNonQuery(OLTP_Connection, "INSERT INTO Customer VALUES('C-1003','Kevin Justin')");
    SqlTask.ExecuteNonQuery(OLTP_Connection, "UPDATE Customer SET Name='Jack Apples' WHERE CustomerNumber = 'C-1002'");
}

To demonstrate the SCD Type 1 process in action:

Console.WriteLine("Load customer as Slowly Changing Dimension Type 1");
LoadCustomer_SCD1();
Console.WriteLine("Changing customer data & reloading");
ChangeCustomer();
LoadCustomer_SCD1();
Console.WriteLine("Customer dimension successfully loaded!");

This script is designed to be run in sequence: first to load initial data into the DWH, then to simulate changes in source data, and finally to reload the data, illustrating how the SCD Type 1 dimension is kept up-to-date.

Loading SCD Type 2 Dimensions

For SCD Type 2, ETLBox must manage the complexity of maintaining historical changes. This typically involves:

Identifying new and changed records in the source data.
Inserting new records into the dimension table for new data.
Ending the validity of the current records and inserting new records for changed data.

ETLBox help with these tasks by providing control flow tasks and data flow components that manage the data transformations and loads.

Here is the source code to load a SCD Type 2 with ETLBox:

public class Product
{
    [RetrieveColumn("DimId")]
    public int? DimId { get; set; }
    public string Name { get; set; }
    [DbColumnMap("ProductNumber")]
    [MatchColumn("Number")]
    public string Number { get; set; }
    public string Description { get; set; }
    public decimal RecommendedPrice { get; set; }
    public DateTime LastUpdated { get; set; }
    [RetrieveColumn("ValidFrom")]
    public DateTime? ValidFrom { get; set; } = new DateTime(1900, 1, 1);
    public DateTime? ValidTo { get; set; } = new DateTime(9999, 12, 31);
}

private static void LoadProducts_SCD2(DateTime lastUpdateDate) {
    var source = new DbSource<Product>(OLTP_Connection);
    source.Sql = "SELECT ProductNumber, Name, Description, RecommendedPrice, LastUpdated FROM Product WHERE LastUpdated >= @par1";
    source.SqlParameter = new[] {
        new QueryParameter() {
            Name="par1", SqlType = "DATETIME", Value = lastUpdateDate
        }
    };

    var lookup = new LookupTransformation<Product, Product>();
    var lookupSource = new DbSource<Product>(OLAP_Connection, "DimProduct");
    lookupSource.Sql = @"
SELECT DISTINCT ProductNumber,
    LAST_VALUE(ValidFrom) OVER(
        PARTITION BY ProductNumber ORDER BY ValidFrom
        RANGE BETWEEN
        UNBOUNDED PRECEDING AND
        UNBOUNDED FOLLOWING
) ValidFrom,
LAST_VALUE(DimId) OVER(
        PARTITION BY ProductNumber ORDER BY DimId
        RANGE BETWEEN
        UNBOUNDED PRECEDING AND
        UNBOUNDED FOLLOWING
) DimId
FROM DimProduct";
    lookup.Source = lookupSource;

    var destNewRows = new DbDestination<Product>(OLAP_Connection, "DimProduct");
    var destChangedRows = new DbDestination<Product>(OLAP_Connection, "DimProduct");
    var destUpdateRows = new DbDestination<Product>(OLAP_Connection, "DimProduct");

    destUpdateRows.BulkOperation = BulkOperation.Update;
    destUpdateRows.IdColumns = new[] {
        new IdColumn() { IdPropertyName = "Number"},
        new IdColumn() { IdPropertyName = "ValidFrom"},
    };
    destUpdateRows.UpdateColumns = new[] {
        new UpdateColumn() { UpdatePropertyName ="ValidTo"}
    };

    var multicast = new Multicast<Product>();

    var updateValidFromTransformation = new RowTransformation<Product>(
        row => {
            row.ValidFrom = row.LastUpdated;
            return row;
        });

    var updateValidToTransformation = new RowTransformation<Product>(
        row => {
            row.ValidTo = row.LastUpdated.AddSeconds(-1);
            return row;
        });

    source.LinkTo(lookup);
    lookup.LinkTo(destNewRows, row => row.DimId == null);
    lookup.LinkTo(multicast, row => row.DimId != null);
    multicast.LinkTo(updateValidFromTransformation).LinkTo(destChangedRows);
    multicast.LinkTo(updateValidToTransformation).LinkTo(destUpdateRows);
    Network.Execute(source);

}

Our ETL process begins by defining a Product class to mirror the structure of our dimension table. Key to this process is the use of LookupTransformation, which differentiates between new records and those requiring updates due to changes. Here’s how the transformation is applied:

New Records: The ETL flow inserts new entries with the ValidFrom and ValidTo dates set to the predefined start and end dates, respectively. This marks the beginning of their validity period within the dimension.
Changed Records: For updated records, the ETL process involves two key steps:
- Ending the current record's validity by updating the ValidTo field to a timestamp just before the new LastUpdated value.
- Inserting a new record with updated information and a ValidFrom value equal to the LastUpdated timestamp of the incoming data.

The distinction and handling of new and updated records are efficiently managed by ETLBox's control and data flows, including the LookupTransformation. The process is then followed by the insertion of new records and the adjustment of historical records to reflect changes accurately.

Furthermore, we enhance the integrity of our historical data by employing an SQL window function post-load to adjust the ValidFrom column. This is necessary to ensure the seamless continuity of record validity, aligning ValidFrom dates with existing ValidTo values. The window function leverages the LAST_VALUE() function within a partitioned data set to provide the most recent ValidFrom date for each unique product number, ensuring that our validity periods are accurate and up to date.

To test this part of the ETL process, you can execute the following sequence to load initial data, apply changes, and then reload to see the updates in the SCD Type 2 dimension:

Console.WriteLine("Load products as Slowy Changing Dimension Type 2");
LoadProducts_SCD2(new DateTime(2023, 1, 1));
Console.WriteLine("Changing product data & reloading");
ChangeProduct();
LoadProducts_SCD2(new DateTime(2023, 1, 4));
Console.WriteLine("Changing product data again & reloading");
ChangeProduct2();
LoadProducts_SCD2(new DateTime(2023, 1, 5));
Console.WriteLine("Product dimension successfully loaded!");

This script simulates the loading of data, applying changes, and reloading, showcasing how ETLBox can be leveraged to manage the complex requirements of SCD Type 2 in a data warehousing environment.

Creating a Date Dimension

Creating a date dimension is a fundamental step that enhances your time-based analysis capabilities. ETLBox can help generate this dimension with the necessary attributes for days, weeks, months, fiscal periods, and so on. This process might include generating a range of dates and the corresponding attributes and then loading them into your date dimension table in the DWH.

Here's the code provided to generate and load the date dimension:

public class DateDimension
{
    [DbColumnMap(IgnoreColumn =true)]
    public DateTime Date { get; set; }
    public int DateID => int.Parse(this.Date.ToString("yyyyMMdd"));
    public string DateString => this.Date.ToString("MM/dd/yyyy"); 
    public int Month => this.Date.Month; 
    public int Day => this.Date.Day; 
    public int Year => this.Date.Year; 
    public int DayofWeek => (int)this.Date.DayOfWeek; 
    public string DayofWeekName => this.Date.DayOfWeek.ToString(); 
    public int DayofYear => this.Date.DayOfYear; 
    public int WeekOfYear => this.GetWeekOfYear();
    public string MonthName => this.Date.ToString("MMMM");
    public int QuarterOfYear => this.GetQuarterOfYear();

    private int GetQuarterOfYear() {
        return (int)Math.Floor(((decimal)this.Date.Month + 2) / 3);
    }

    private int GetWeekOfYear() {
        System.Globalization.DateTimeFormatInfo dfi = System.Globalization.DateTimeFormatInfo.CurrentInfo;
        System.Globalization.Calendar cal = dfi.Calendar;
        return cal.GetWeekOfYear(this.Date, dfi.CalendarWeekRule, dfi.FirstDayOfWeek);
    }

    public static IEnumerable<DateDimension> Generate(DateTime StartDate, DateTime EndDate) {

        int TotalDays = (int)(EndDate.AddDays(1) - StartDate).TotalDays;
        return Enumerable.Range(0, TotalDays).Select(e =>
            new DateDimension() { Date = StartDate.AddDays(e) }
            ).OrderBy(e => e.DateID).ToList();
    }
}


private static void CreateDateDimension() {
    DropTableTask.DropIfExists(OLAP_Connection, "DimDate");
    CreateTableTask.CreateIfNotExists(OLAP_Connection, "DimDate",
        typeof(DateDimension).GetProperties()
        .Where(prop => prop.Name != nameof(DateDimension.Date))
        .Select(prop => 
            new TableColumn(
            name: prop.Name,
            dataType: prop.PropertyType == typeof(int) ? "INT" : "VARCHAR(30)",
            allowNulls: false,
            isPrimaryKey: prop.Name == nameof(DateDimension.DateID)
            )).ToList());

    var source = new MemorySource<DateDimension>();
    source.Data = DateDimension.Generate(new DateTime(2023, 1, 1), new DateTime(2023, 1, 5));
    var dest = new DbDestination<DateDimension>(OLAP_Connection, "DimDate");
    source.LinkTo(dest);
    Network.Execute(source);
}

The ETL process involves the following steps:

The DateDimension class is constructed to mirror the date dimension table in the DWH, encompassing all the necessary properties like Month, Day, Year, etc., and helper methods to compute WeekOfYear and QuarterOfYear.
The Generate method within the DateDimension class facilitates the creation of the date range needed. It calculates the total number of days between the start and end dates and generates a corresponding list of DateDimension objects.
To prepare for data loading, existing tables are dropped and new tables are created within the database using ETLBox's DropTableTask and CreateTableTask methods.
The MemorySource and DbDestination classes are used to create a data flow where the generated date dimension data is written from memory to the database dimension table.

To load the date dimension into the DWH, you would run:

Console.WriteLine("Creating a generic date dimension");
CreateDateDimension();

Loading Fact Data

Loading data into a fact table is crucial when establishing a Data Warehouse. This operation typically involves the extraction of data from source systems, its transformation to fit the DWH schema, and finally, its insertion into the fact table.

Incremental Loads vs. Full Loads

ETLBox facilitates both incremental and full data loads, providing flexibility depending on your needs:

Incremental Loads are more efficient for production environments because they only add new or updated data.
Full Loads are more common during the initial setup of a DWH or when a complete data refresh is required.

Deciding between incremental loads and full loads depends on the specific needs of your organization and the nature of your data.

Code example

The provided code snippet defines an ETL process for loading the orders fact data into our DWH:

public class Order
{
    [DbColumnMap("SourceOrderId")]
    public int OrderId { get; set; }
    public DateTime OrderDate { get; set; }
    public int? DateDim => int.Parse(OrderDate.ToString("yyyyMMdd"));
    public string ProductNumber { get; set; }
    public int? ProductDim { get; set; }
    public string CustomerNumber { get; set; }
    public int? CustomerDim { get; set; }
    [DbColumnMap("ActualPriceFact")]
    public decimal ActualPrice { get; set; }
}

private static DateTime GetMaxOrderDate() =>
    SqlTask.ExecuteScalar<DateTime>(OLTP_Connection, "SELECT MAX(OrderDate) FROM Orders");


private static void AddOrders() {
    SqlTask.ExecuteNonQuery(OLTP_Connection, "INSERT INTO Orders VALUES(20001,'2023-01-04', 'P-00013', 'C-1001', 1555)");
    SqlTask.ExecuteNonQuery(OLTP_Connection, "INSERT INTO Orders VALUES(20002,'2023-01-04', 'P-00010', 'C-1002', 288)");
    SqlTask.ExecuteNonQuery(OLTP_Connection, "INSERT INTO Orders VALUES(20003,'2023-01-04', 'P-00011', 'C-1003', 689)");
}

private static void AddOrderWithFlawedRecord() {
    SqlTask.ExecuteNonQuery(OLTP_Connection, "INSERT INTO Orders VALUES(30002,'2023-01-06', 'P-XXXXX', 'C-1002', 1999)");
}



private static void LoadOrders(DateTime lastLoadDate) {
    var source = new DbSource<Order>(OLTP_Connection);
    source.SqlParameter = new[] {
        new QueryParameter() { Name = "date", Value = lastLoadDate }
    };
    source.Sql = @"SELECT OrderId, OrderDate, ProductNumber, CustomerNumber, ActualPrice
                       FROM Orders
                       WHERE OrderDate > @date";

    var dest = new DbDestination<Order>(OLAP_Connection, "FactOrders");

    var customerLookup = new LookupTransformation<Order, Customer>();
    var customerLookupSource = new DbSource<Customer>(OLAP_Connection, "DimCustomer");
    customerLookup.MatchColumns = new[] {
        new MatchColumn() {
            InputPropertyName = nameof(Order.CustomerNumber),
            LookupSourcePropertyName = nameof(Customer.Number)
        }
    };
    customerLookup.RetrieveColumns= new[] {
        new RetrieveColumn() {
            InputPropertyName = nameof(Order.CustomerDim),
            LookupSourcePropertyName = nameof(Customer.DimId)
        }
    };
    customerLookup.Source = customerLookupSource;

    var productLookup = new LookupTransformation<Order, Product>();
    var productLookupSource = new DbSource<Product>(OLAP_Connection, "DimProduct");
    productLookup.MatchColumns = new[] {
        new MatchColumn() {
            InputPropertyName = nameof(Order.ProductNumber),
            LookupSourcePropertyName = nameof(Product.Number)
        }
    };
    productLookup.RetrieveColumns = new[] {
        new RetrieveColumn() {
            InputPropertyName = nameof(Order.ProductDim),
            LookupSourcePropertyName = nameof(Product.DimId)
        }
    };
    productLookup.Source = productLookupSource;

    var errorLog = new JsonDestination<ETLBoxError>("errros.log");

    source.LinkTo(customerLookup);
    customerLookup.LinkTo(productLookup);
    productLookup.LinkTo(dest);

    dest.LinkErrorTo(errorLog);

    Network.Execute(source);
}

This ETL process includes:

Data Extraction: The DbSource class is used to pull data from the source database that has been added since the last load date.
Data Transformation: The LookupTransformation class enriches the data by retrieving dimension keys from the DimCustomer and DimProduct tables based on the incoming order's customer and product numbers.
Error Handling: An error.log file is set up to capture any records that fail to load properly, outputting them in a JSON format for troubleshooting.
Data Loading: The DbDestination class handles the insertion of transformed data into the FactOrders table, using bulk insert sql statements for efficiency.
Performance Tuning: The BatchSize property of DbDestination could be adjusted to optimize performance. Larger batch sizes can increase throughput, though they may also increase resource consumption. The default batch size is 1000 rows per bulk insert.

If the insert of fact data fails for any reason, the entire flawed batch is redirected into an error file ("errors.log"). It's important to note that this file will contain all records from the batch, not just the flawed ones. For more granular error checking, it's advisable to incorporate data validation or cleansing transformations into the data flow beforehand. This would help in filtering out erroneous data before the insertion into the database occurs.

To run the initial and subsequent incremental data loads, the following commands are executed:

Console.WriteLine("Adding fact data with one flawed record and loading again - flawed entry is redirected into 'error.json'");
lastLoadDate = GetMaxOrderDate();
AddOrderWithFlawedRecord();
LoadOrders(lastLoadDate);
Console.WriteLine("Fact table successfully loaded!");

This command sequence does the following:

Gets the most recent OrderDate from the fact table to determine the starting point for the incremental load.
Adds an additional order with a flawed record to simulate an error condition.
Executes the LoadOrders method to perform the incremental load.

Wrap-up

In this guide, we've taken a thorough look at how ETLBox, a versatile .NET tool, can be used for building and running Data Warehouses. We've gone over a range of key tasks, including managing different types of Slowly Changing Dimensions (SCD), working with date dimensions, and handling fact tables. Along the way, we also gave you a glimpse into how ETLBox handles errors, showcasing its reliability in managing ETL processes.

Code on GitHub

For a deeper dive and hands-on experience, the complete code for this example is available on GitHub. You can access and explore it to better understand the functionalities and possibilities of ETLBox:

ETLBox Example Code on GitHub

Conclusion

ETLBox presents a code-centric, flexible approach for building a DWH, fully utilizing the .NET framework. This makes it an ideal solution for teams familiar with .NET, allowing them to leverage their existing skills to create a powerful and efficient DWH.

However, ETLBox isn't limited to statically typed objects; it also supports dynamic data types through the use of the ExpandoObject. This dynamic object, especially when paired with the dynamic keyword in C#, empowers developers to create adaptable data flows without the need for predefined data structures.

Leveraging dynamic objects alongside a configuration-driven approach not only increases the reusability of your ETL code across various dimensions and fact tables but also introduces an impressive degree of flexibility to the data loading process.

Furthermore, the ability to craft your ETL code in C# opens the door to applying standard testing practices, such as Unit or Integration tests. This means your ETL processes can be fully integrated with existing test frameworks like xUnit or NUnit, ensuring reliability and stability. Additionally, ETLBox naturally fits into your existing Continuous Integration/Continuous Deployment (CI/CD) pipelines and version control systems, aligning your data processing workflows with modern development practices.

Ultimately, the strength and adaptability of ETLBox make it a standout ETL tool for any organization seeking to develop or expand their data warehousing solutions.

Redefining ETL: Data Flows Powered by C# (Part III)

Andreas Lennartz — Wed, 11 Oct 2023 18:59:43 +0000

This is Part 3 in a series of posts. If you'd like to read Part 2, click here. Or start with the first post.

Creating Dynamic Dataflows

In Part 1, we delved into ETLBox, a robust .NET tool for data management. We explored its features, such as its straightforward API, versatility in handling transformations, and efficiency with large data volumes. Using ETLBox, we set up a data flow to process a CSV file of supplier data, emphasizing data validation and error management.

In Part 2, we enhanced the data flow, focusing on adding custom error messages. We also addressed the challenge of handling repeated or different supplier data loads. To capture these changes effectively, we implemented a Slowly Changing Dimension Type 2 (SCD Type 2).

Looking ahead to Part 3, we are going to convert the same data flow away from static types into a dynamic and configurable one

From static to dynamic types

Traditionally, C# has been anchored to the realm of static typing. While this provides a robust structure and type safety, it can sometimes be a constraint, especially when dealing with flexible and ever-evolving data structures in the world of ETL.

Enter the realm of ExpandoObject and the dynamic keyword. These tools break free from the rigidity of static types, offering a more fluid data handling experience. Just because C# is statically typed doesn't mean we can't explore the versatility of dynamic data.

Thankfully, ETLBox is not just confined to static types. It fully embraces and supports the dynamic nature of ExpandoObject. In this part, we'll pivot our example, shedding the static and ushering in the dynamic, showing you how to achieve greater adaptability in your dataflow processes.

Diving into Dynamic Types and ExpandoObject

In C#, the dynamic keyword introduces a paradigm shift from the traditionally statically-typed nature of the language. The dynamic type bypasses compile-time type checking. Instead, types are resolved at runtime. This opens the door to greater flexibility, especially when the exact structure of data isn't known until runtime.

The Dynamic Keyword

Using dynamic, you can declare variables without a clearly defined type:

dynamic value = "Hello, World!";
Console.WriteLine(value);  // Outputs: Hello, World!

value = 42;  
Console.WriteLine(value);  // Outputs: 42

Here, the type of value is not fixed and can change as required.

ExpandoObject

ExpandoObject is an in-built .NET class that lets you add properties, methods, and events to an object dynamically at runtime. It's especially useful when working with data structures that might be fluid and not set in stone.

dynamic person = new ExpandoObject();
person.Name = "John";
person.Age = 30;

Console.WriteLine(person.Name);  // Outputs: John
Console.WriteLine(person.Age);   // Outputs: 30

But what makes ExpandoObject even more powerful is its ability to be treated as a dictionary where properties can be accessed with keys.

IDictionary with ExpandoObject

Every ExpandoObject can be cast to IDictionary<string, object>, allowing properties to be accessed, added, or removed using dictionary-like syntax.

dynamic person = new ExpandoObject();
var personDict = (IDictionary<string, object>)person;
personDict.Add("Country", "USA");

Console.WriteLine(personDict["Country"]);  // Outputs: USA
Console.WriteLine(person.Country);  // Outputs: USA

This approach allows for extremely versatile and flexible data handling, especially when dynamically constructing or deconstructing objects based on changing requirements.

In the world of ETL, where data sources and structures can change frequently, the power of dynamic and ExpandoObject provides an invaluable toolset for adaptable and resilient data transformations.

Adding Configuration Table

To facilitate more dynamic behaviour in our application, we'll establish a configuration table called Meta. This table will host essential metadata such as business keys, obligatory fields, and potential field length constraints, as well as the associated filename in the source and its corresponding database column. This configuration is aligned with our static table structure from the previous segment.

//Prepare the "meta" table that contains our table definition and rules
DropTableTask.DropIfExists(connectionManager, "Meta");
CreateTableTask.CreateIfNotExists(connectionManager, "Meta",
    new List<TableColumn>() {
        new TableColumn() { Name = "ColumnName", DataType = "VARCHAR(1024)", AllowNulls = false  },
        new TableColumn() { Name = "DataType", DataType = "VARCHAR(20)", AllowNulls = false  },
        new TableColumn() { Name = "IsMandatory", DataType = "INT", AllowNulls = false },
        new TableColumn() { Name = "IsBusinessKey", DataType = "INT", AllowNulls = false, DefaultValue = "0"  },
        new TableColumn() { Name = "FileColumnName", DataType = "VARCHAR(1024)", AllowNulls = false  },
        new TableColumn() { Name = "MinFieldLength", DataType = "INT", AllowNulls = true },
        new TableColumn() { Name = "MaxFieldLength", DataType = "INT", AllowNulls = true },
        new TableColumn() { Name = "Trim", DataType = "INT", AllowNulls = false, DefaultValue = "0" },
        new TableColumn() { Name = "Uppercase", DataType = "INT", AllowNulls = false, DefaultValue = "0" },
    });

SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO Meta (ColumnName, DataType, IsMandatory, IsBusinessKey, FileColumnName, MinFieldLength, MaxFieldLength, Trim, Uppercase)
        VALUES('VendorName','VARCHAR(100)', 1, 0, 'Name', 5, 50, 1, 1)");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO Meta (ColumnName, DataType, IsMandatory, IsBusinessKey, FileColumnName, MinFieldLength, MaxFieldLength, Trim, Uppercase)
        VALUES('Code','CHAR(5)', 1, 1, 'Code', 5, 5, 0, 0)");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO Meta (ColumnName, DataType, IsMandatory, IsBusinessKey, FileColumnName, MinFieldLength, MaxFieldLength, Trim, Uppercase)
        VALUES('Custom','VARCHAR(10)', 1, 1, 'Custom', NULL, NULL, 0, 0)");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO Meta (ColumnName, DataType, IsMandatory, IsBusinessKey, FileColumnName, MinFieldLength, MaxFieldLength, Trim, Uppercase)
        VALUES('Country','CHAR(2)', 0, 0, 'Country', NULL, NULL, 0, 0)");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO Meta (ColumnName, DataType, IsMandatory, IsBusinessKey, FileColumnName, MinFieldLength, MaxFieldLength, Trim, Uppercase)
        VALUES('Contact','VARCHAR(50)', 0, 0, 'Contact', NULL, NULL, 0, 0)");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO Meta (ColumnName, DataType, IsMandatory, IsBusinessKey, FileColumnName, MinFieldLength, MaxFieldLength, Trim, Uppercase)
        VALUES('Info','VARCHAR(50)', 0, 0, 'TraceInfo', NULL, NULL, 0, 0)");

Loading configuration into Memory

While there are robust ORM mappers like Dapper or EntityFramework, for our purpose, we’ll employ a concise ETLBox data flow to load our configurations into an in-memory object.


//Load the configuration into memory - we are using ETLBox for this
var configSource = new DbSource<Meta>(connectionManager, "Meta");
var configDest = new MemoryDestination<Meta>();
configSource.LinkTo(configDest);
Network.Execute(configSource);
var config = configDest.Data;

public class Meta
{
    public string ColumnName { get; set; }
    public string DataType { get; set; }
    public bool IsMandatory { get; set; }
    public bool IsBusinessKey { get; set; }
    public string FileColumnName { get; set; }
    public int MinFieldLength { get; set; }
    public int MaxFieldLength { get; set; }
    public bool Trim { get; set; }
    public bool Uppercase { get; set; }
}

Constructing Demo Table Using Configuration

Instead of a fixed setup, we'll now leverage our configuration data to create our demo table dynamically.

//Prepare the target database based on configuration
DropTableTask.DropIfExists(connectionManager, "VendorMaster");
var columns = new List<TableColumn>();
columns.Add(new TableColumn() { Name = "Id", DataType = "INT", AllowNulls = false, IsPrimaryKey = true, IsIdentity = true });
columns.AddRange(config.Select(c =>
    new TableColumn() { Name = c.ColumnName, DataType = c.DataType, AllowNulls = !c.IsMandatory })
);
columns.Add(new TableColumn() { Name = "ValidFrom", DataType = "DATETIME", AllowNulls = false });
columns.Add(new TableColumn() { Name = "ValidTo", DataType = "DATETIME", AllowNulls = false });
CreateTableTask.CreateIfNotExists(connectionManager, "VendorMaster", columns);

SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom, Country, Contact, Info, ValidFrom,ValidTo)
        VALUES('BIG HOLDING', 'H1234', 'HD', 'US', 'Hans', 'T0', '1900-1-1','9999-12-31')");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom,Country, ValidFrom,ValidTo)
        VALUES('UNICORN', 'UNI10', 'U', 'NO', '1900-1-1','9999-12-31')");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom,Country, ValidFrom,ValidTo)
        VALUES('UNICORN TWO', 'UNI20', 'U', 'SE', '1900-1-1','9999-12-31')");

It's important to note that we're still incorporating static columns, namely the id column, ValidFrom, and ValidTo. These columns aren't found in our dynamic configuration; therefore, our dynamic data flow will work under the assumption that they're present.

Dynamically Crafting ETLBox Components

ETLBox components are flexibly designed with both generic and non-generic class definitions. Often, in the context of ETL processes, developers define strongly-typed objects, or POCOs (Plain Old CLR Objects), to represent the data flow, as demonstrated in earlier sections. However, for more dynamic operations, using the ExpandoObject can be advantageous.

For instance, rather than using RowTransformation<VendorMaster>, we can employ RowTransformation<ExpandoObject>, which is equivalent to the simple RowTransformation.

Let's look at the dynamic component creation and decode the details.

//Create the data flow components
var source = new CsvSource("sourceData.csv");
var normalize = new RowTransformation(row => Normalize(row));
var errorTarget = new JsonDestination("errors.json");
var dbTarget = new DbDestination(connectionManager, "VendorMaster");
dbTarget.ColumnMapping = config
    .Where(c => c.ColumnName != c.FileColumnName)
    .Select(c => new DbColumnMap() { DbColumnName = c.ColumnName, PropertyName = c.FileColumnName })
    .ToList();
var lookupExisting = new LookupTransformation();
var lookupSource = new DbSource(connectionManager, "VendorMaster");
lookupExisting.Source = lookupSource;
lookupExisting.MatchColumns = config
    .Where(c => c.IsBusinessKey)
    .Select(c => new MatchColumn() { InputPropertyName = c.FileColumnName, LookupSourcePropertyName = c.ColumnName })
    .ToList();
lookupExisting.RetrieveColumns = new[] {
    new RetrieveColumn() { InputPropertyName = "DbId", LookupSourcePropertyName ="Id"}
};
var duplicateCheck = new Distinct();
duplicateCheck.DistinctColumns = config
    .Where(c => c.IsBusinessKey)
    .Select(c => new DistinctColumn() { DistinctPropertyName = c.FileColumnName })
    .ToList();

//Adjust ValidFrom to current data if record exists
var adjustValidFrom = new RowTransformation(row => {
    dynamic r = row as dynamic;
    if ((row as IDictionary<string,object>).ContainsKey("DbId") && r.DbId > 0)
        r.ValidFrom = DateTime.Now;
    else 
        r.ValidFrom = new DateTime(1900, 1, 1);
    r.ValidTo = new DateTime(9999, 12, 31);
    return row;
});

//Add custom error message to duplicates
var addNotDistinctErrorMessage = new RowTransformation(
    row => {
        (row as dynamic).ErrorMessage = "Duplicate in source data detected!";
        return row;
    });

//Modify Merge to work with SCD-2
dbTarget.AfterBatchWrite = batch => {
    SqlTask.ExecuteNonQuery(connectionManager, @"
UPDATE VendorMaster
SET VendorMaster.ValidTo = calc.validto
FROM (SELECT Id,
             Code,
             Custom, 
             ValidFrom,
             LEAD(ValidFrom)
                  OVER (
                      PARTITION BY Code,Custom
                      ORDER BY ValidFrom
                      ) validto
      FROM VendorMaster) calc
WHERE VendorMaster.Id = calc.Id
  AND calc.validto IS NOT NULL
");
};


dynamic Normalize(IDictionary<string,object> row) {
    foreach (var configEntry in config) {
        if (configEntry.Uppercase == true)
            row[configEntry.FileColumnName] = (row[configEntry.FileColumnName] as string).ToUpper();
        if (configEntry.Trim == true)
            row[configEntry.FileColumnName] = (row[configEntry.FileColumnName] as string).Trim();
    }
    return row;
}

We've transitioned the Normalize method, previously part of the VendorMaster class, into a standalone function. This function has been revamped to utilize the configuration data retrieved from our Meta table.

Furthermore, previously defined attributes like DbColumnMap, used for property mappings, are now set dynamically via the ColumnMapping attribute on the DbDestination component. This method of dynamic assignment also extends to other attributes, such as those for LookupTransformation and the Distinct transformation. Importantly, these dynamically assigned properties will replace any existing attributes that may already exist on a class definition, offering a valuable tool for minor adjustments in data flow behaviour.

Other transformations, including RowTransformation, have been updated to support dynamic objects.

Connecting Components Dynamically

Crafting a dynamic network flow now becomes straightforward:

/* Linking the components */
source.LinkTo(normalize);

normalize.LinkTo(lookupExisting, row => IsValid(row));
normalize.LinkTo(errorTarget, row => true); //Valid rows are already send to the lookup!

//var debug = new MemoryDestination()
lookupExisting.LinkTo(adjustValidFrom);
adjustValidFrom.LinkTo(duplicateCheck);

duplicateCheck.LinkTo(dbTarget);
duplicateCheck.LinkDuplicatesTo(addNotDistinctErrorMessage);
addNotDistinctErrorMessage.LinkTo(errorTarget);

//Execute the network
await Network.ExecuteAsync(source);

bool IsValid(IDictionary<string, object> row) {
    foreach (var configEntry in config) {
        if (configEntry.IsMandatory == true &&
            string.IsNullOrEmpty(row[configEntry.FileColumnName] as string)) {
            row.Add("ErrorMessage", "Empty required column detected!");
            return false;
        }
        if (configEntry.MinFieldLength > 0 &&
            (row[configEntry.FileColumnName] as string).Length < configEntry.MinFieldLength) {
            row.Add("ErrorMessage", "Unsupported length!");
            return false;
        }
        if (configEntry.MaxFieldLength > 0 &&
            (row[configEntry.FileColumnName] as string).Length > configEntry.MaxFieldLength) {
            row.Add("ErrorMessage", "Unsupported length!");
            return false;
        }
    }
    return true;
}

We've refined the IsValid method to cater to the ExpandoObject. Moreover, this method also leverages the loaded configuration data for row validation.

Complete code overview

Here is an overview of the whole code the example. It is also available on GitHub.

//Adjust connection string to your local database
string SqlConnectionString = @"Data Source=localhost;User Id=sa;Password=YourStrong@Passw0rd;Initial Catalog=demo;TrustServerCertificate=true";
SqlConnectionManager connectionManager = new SqlConnectionManager(SqlConnectionString);

//Prepare the "meta" table that contains our table definition and rules
DropTableTask.DropIfExists(connectionManager, "Meta");
CreateTableTask.CreateIfNotExists(connectionManager, "Meta",
    new List<TableColumn>() {
        new TableColumn() { Name = "ColumnName", DataType = "VARCHAR(1024)", AllowNulls = false  },
        new TableColumn() { Name = "DataType", DataType = "VARCHAR(20)", AllowNulls = false  },
        new TableColumn() { Name = "IsMandatory", DataType = "INT", AllowNulls = false },
        new TableColumn() { Name = "IsBusinessKey", DataType = "INT", AllowNulls = false, DefaultValue = "0"  },
        new TableColumn() { Name = "FileColumnName", DataType = "VARCHAR(1024)", AllowNulls = false  },
        new TableColumn() { Name = "MinFieldLength", DataType = "INT", AllowNulls = true },
        new TableColumn() { Name = "MaxFieldLength", DataType = "INT", AllowNulls = true },
        new TableColumn() { Name = "Trim", DataType = "INT", AllowNulls = false, DefaultValue = "0" },
        new TableColumn() { Name = "Uppercase", DataType = "INT", AllowNulls = false, DefaultValue = "0" },
    });

SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO Meta (ColumnName, DataType, IsMandatory, IsBusinessKey, FileColumnName, MinFieldLength, MaxFieldLength, Trim, Uppercase)
        VALUES('VendorName','VARCHAR(100)', 1, 0, 'Name', 5, 50, 1, 1)");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO Meta (ColumnName, DataType, IsMandatory, IsBusinessKey, FileColumnName, MinFieldLength, MaxFieldLength, Trim, Uppercase)
        VALUES('Code','CHAR(5)', 1, 1, 'Code', 5, 5, 0, 0)");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO Meta (ColumnName, DataType, IsMandatory, IsBusinessKey, FileColumnName, MinFieldLength, MaxFieldLength, Trim, Uppercase)
        VALUES('Custom','VARCHAR(10)', 1, 1, 'Custom', NULL, NULL, 0, 0)");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO Meta (ColumnName, DataType, IsMandatory, IsBusinessKey, FileColumnName, MinFieldLength, MaxFieldLength, Trim, Uppercase)
        VALUES('Country','CHAR(2)', 0, 0, 'Country', NULL, NULL, 0, 0)");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO Meta (ColumnName, DataType, IsMandatory, IsBusinessKey, FileColumnName, MinFieldLength, MaxFieldLength, Trim, Uppercase)
        VALUES('Contact','VARCHAR(50)', 0, 0, 'Contact', NULL, NULL, 0, 0)");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO Meta (ColumnName, DataType, IsMandatory, IsBusinessKey, FileColumnName, MinFieldLength, MaxFieldLength, Trim, Uppercase)
        VALUES('Info','VARCHAR(50)', 0, 0, 'TraceInfo', NULL, NULL, 0, 0)");

//Load the configuration into memory - we are using ETLBox for this
var configSource = new DbSource<Meta>(connectionManager, "Meta");
var configDest = new MemoryDestination<Meta>();
configSource.LinkTo(configDest);
Network.Execute(configSource);
var config = configDest.Data;

//Prepare the target database based on configuration
DropTableTask.DropIfExists(connectionManager, "VendorMaster");
var columns = new List<TableColumn>();
columns.Add(new TableColumn() { Name = "Id", DataType = "INT", AllowNulls = false, IsPrimaryKey = true, IsIdentity = true });
columns.AddRange(config.Select(c =>
    new TableColumn() { Name = c.ColumnName, DataType = c.DataType, AllowNulls = !c.IsMandatory })
);
columns.Add(new TableColumn() { Name = "ValidFrom", DataType = "DATETIME", AllowNulls = false });
columns.Add(new TableColumn() { Name = "ValidTo", DataType = "DATETIME", AllowNulls = false });
CreateTableTask.CreateIfNotExists(connectionManager, "VendorMaster", columns);

SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom, Country, Contact, Info, ValidFrom,ValidTo)
        VALUES('BIG HOLDING', 'H1234', 'HD', 'US', 'Hans', 'T0', '1900-1-1','9999-12-31')");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom,Country, ValidFrom,ValidTo)
        VALUES('UNICORN', 'UNI10', 'U', 'NO', '1900-1-1','9999-12-31')");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom,Country, ValidFrom,ValidTo)
        VALUES('UNICORN TWO', 'UNI20', 'U', 'SE', '1900-1-1','9999-12-31')");


//Create the data flow components
var source = new CsvSource("sourceData.csv");
var normalize = new RowTransformation(row => Normalize(row));
var errorTarget = new JsonDestination("errors.json");
var dbTarget = new DbDestination(connectionManager, "VendorMaster");
dbTarget.ColumnMapping = config
    .Where(c => c.ColumnName != c.FileColumnName)
    .Select(c => new DbColumnMap() { DbColumnName = c.ColumnName, PropertyName = c.FileColumnName })
    .ToList();
var lookupExisting = new LookupTransformation();
var lookupSource = new DbSource(connectionManager, "VendorMaster");
lookupExisting.Source = lookupSource;
lookupExisting.MatchColumns = config
    .Where(c => c.IsBusinessKey)
    .Select(c => new MatchColumn() { InputPropertyName = c.FileColumnName, LookupSourcePropertyName = c.ColumnName })
    .ToList();
lookupExisting.RetrieveColumns = new[] {
    new RetrieveColumn() { InputPropertyName = "DbId", LookupSourcePropertyName ="Id"}
};
var duplicateCheck = new Distinct();
duplicateCheck.DistinctColumns = config
    .Where(c => c.IsBusinessKey)
    .Select(c => new DistinctColumn() { DistinctPropertyName = c.FileColumnName })
    .ToList();

//Adjust ValidFrom to current data if record exists
var adjustValidFrom = new RowTransformation(row => {
    dynamic r = row as dynamic;
    if ((row as IDictionary<string,object>).ContainsKey("DbId") && r.DbId > 0)
        r.ValidFrom = DateTime.Now;
    else 
        r.ValidFrom = new DateTime(1900, 1, 1);
    r.ValidTo = new DateTime(9999, 12, 31);
    return row;
});

//Add custom error message to duplicates
var addNotDistinctErrorMessage = new RowTransformation(
    row => {
        (row as dynamic).ErrorMessage = "Duplicate in source data detected!";
        return row;
    });

//Modify Merge to work with SCD-2
dbTarget.AfterBatchWrite = batch => {
    SqlTask.ExecuteNonQuery(connectionManager, @"
UPDATE VendorMaster
SET VendorMaster.ValidTo = calc.validto
FROM (SELECT Id,
             Code,
             Custom, 
             ValidFrom,
             LEAD(ValidFrom)
                  OVER (
                      PARTITION BY Code,Custom
                      ORDER BY ValidFrom
                      ) validto
      FROM VendorMaster) calc
WHERE VendorMaster.Id = calc.Id
  AND calc.validto IS NOT NULL
");
};


/* Linking the components */
source.LinkTo(normalize);

normalize.LinkTo(lookupExisting, row => IsValid(row));
normalize.LinkTo(errorTarget, row => true); //Valid rows are already send to the lookup!

//var debug = new MemoryDestination()
lookupExisting.LinkTo(adjustValidFrom);
adjustValidFrom.LinkTo(duplicateCheck);

duplicateCheck.LinkTo(dbTarget);
duplicateCheck.LinkDuplicatesTo(addNotDistinctErrorMessage);
addNotDistinctErrorMessage.LinkTo(errorTarget);

//Execute the network
await Network.ExecuteAsync(source);

Console.WriteLine("Done!");

dynamic Normalize(IDictionary<string,object> row) {
    foreach (var configEntry in config) {
        if (configEntry.Uppercase == true)
            row[configEntry.FileColumnName] = (row[configEntry.FileColumnName] as string).ToUpper();
        if (configEntry.Trim == true)
            row[configEntry.FileColumnName] = (row[configEntry.FileColumnName] as string).Trim();
    }
    return row;
}
bool IsValid(IDictionary<string, object> row) {
    foreach (var configEntry in config) {
        if (configEntry.IsMandatory == true &&
            string.IsNullOrEmpty(row[configEntry.FileColumnName] as string)) {
            row.Add("ErrorMessage", "Empty required column detected!");
            return false;
        }
        if (configEntry.MinFieldLength > 0 &&
            (row[configEntry.FileColumnName] as string).Length < configEntry.MinFieldLength) {
            row.Add("ErrorMessage", "Unsupported length!");
            return false;
        }
        if (configEntry.MaxFieldLength > 0 &&
            (row[configEntry.FileColumnName] as string).Length > configEntry.MaxFieldLength) {
            row.Add("ErrorMessage", "Unsupported length!");
            return false;
        }
    }
    return true;
}

public class Meta
{
    public string ColumnName { get; set; }
    public string DataType { get; set; }
    public bool IsMandatory { get; set; }
    public bool IsBusinessKey { get; set; }
    public string FileColumnName { get; set; }
    public int MinFieldLength { get; set; }
    public int MaxFieldLength { get; set; }
    public bool Trim { get; set; }
    public bool Uppercase { get; set; }
}

By introducing these nuanced changes to the data flow, we've successfully transitioned from a static type-based approach to a dynamic, configuration-driven data loading process. This approach significantly boosts flexibility, making it more adaptable to evolving data requirements.

Conclusion

Over the course of three articles, we've delved deep into ETLBox, a potent .NET library dedicated to data management. In the first installment, we discovered the essence of ETLBox: a tool primarily designed for ETL processes without the dependence on a GUI. Key takeaways about ETLBox include its user-friendly API and the capacity to support various transformations and data sources. A hands-on example showed us how to craft an ETL data flow with ETLBox, processing a CSV supplier dataset and emphasizing its capabilities in data validation, normalization, and error management.

The second article brought enhancements to the table. We enhanced error reporting and incorporated Slowly Changing Dimension (SCD) Type 2, using ValidFrom and ValidTo date columns, enabling us to chronicle data alterations over time.

In the third segment, we shifted gears, handling data flow using dynamic data objects. We ventured into the realms of the dynamic keyword and ExpandoObject within C#, illustrating the power of moving beyond fixed data types. A new configuration table emerged, housing pivotal metadata and fostering the app's dynamic nature.

To sum it all up, the series spotlighted ETLBox as a formidable ETL tool for the .NET community. Its range from foundational data flow setups to intricate enhancements like SCD Type 2 and dynamic data operations underscores its versatility, dependability, and adaptability. As ETLBox introduces dynamic data flows, it sets a new standard by addressing the ever-evolving challenges of data sourcing and structuring, offering an unparalleled level of flexibility. Although there's undeniable value in static typing, the embrace of a dynamic approach positions ETLBox prominently within the .NET ETL ecosystem.

Redefining ETL: Data Flows Powered by C# (Part II)

Andreas Lennartz — Wed, 11 Oct 2023 18:56:45 +0000

This is Part 2 in a series of posts. If you'd like to read Part 1, click here.

Enhancing the Data Flow

In Part 1, we explored ETLBox, a .NET library designed for efficient data management. Without the need for a graphical interface, ETLBox simplifies data processes by providing an intuitive API, supporting various transformations and data sources, and ensuring optimal performance. It's specifically tailored for the .NET framework and emphasizes stable ETL processes. We also discussed how to use ETLBox packages from NuGet and demonstrated how to process a CSV file of supplier data, emphasizing steps like validation and error management.

In this part, we'll enhance this data flow. We'll not only boost error logging but also tackle the challenge of managing repeated or varied supplier data loads. Our goal is to create a Slowly Changing Dimension Type 2 (SCD Type 2) to capture changes in the CSV data.

Enhanced Error Logging

Errors are inevitable in any data processing task. Our solution in this example is to add a custom error message to each faulty record and store it in our pre-existing error file. If you are interested in a more sophisticated way of redirecting error, you can also have a look at the LinkErrorTo methods provided on each data flow component. To keep this example simple, we will just focus on adding a simple error message to the data itself.

Implementing SCD Type 2

SCD Type 2 (Slowly Changing Dimension Type 2) is a methodology used to track historical changes in data. We also want to improve our data flow to now work with a SCD Type 2. With Type 2, we get historical data tracking. This ensures that the data's history is preserved, and any changes can be tracked over time. To achieve this, each historical record in an SCD Type 2 gets a validity range, using a ValidFrom and ValidTo date column. When a new record is inserted, it gets a ValidFrom date which is in the past, before any of our transactional data starts. The ValidTo is a date in the far future. When a new record is created due to a change, the old record is given an "end date", which is also the start date of the new record.

Adding custom error messages

In order to append an error message when a record becomes invalid, we integrate the ErrorMessage property into our VendorMaster class. Simultaneously, the IsValid method is modified to craft appropriate error messages for such records.

public class VendorMaster {
    ...
    public string ErrorMessage { get; set; }
    ...
    public bool IsValid() {
        if (string.IsNullOrEmpty(Name) || string.IsNullOrEmpty(Code) || string.IsNullOrEmpty(Custom)) {
            ErrorMessage = "Empty required column detected!";
            return false;
        }
        if (Name.Length < 5 || Name.Length > 50) {
            ErrorMessage = "Name has unsupported length!";
            return false;
        }
        if (Code.Length != 5) {
            ErrorMessage = "Code has unsupported length!";
            return false;
        }
        return true;
    }
}

Changing predicates in links

Subsequently, we need to refine the links to our error destination.
Previously, when segregating valid from invalid records towards the error destination, our approach was:

normalize.LinkTo(lookupExisting, row => row.IsValid());
normalize.LinkTo(errorTarget, row => !row.IsValid());

Nevertheless, invoking the IsValid method twice could boost readability but might also cause unintended side effects. Importantly, when creating links between components using predicates, the sequence in which these links are established becomes crucial. Predicates undergo tests in the order they're appended. Once a predicate evaluates as true, subsequent ones aren't examined. Hence, all valid rows dispatched to the lookup using the first link implies we can enhance our code for identical behaviour as shown:

normalize.LinkTo(lookupExisting, row => row.IsValid());
normalize.LinkTo(errorTarget, row => true); //Valid rows are already send to the lookup!

Similarly, while linking the Lookup to the Distinct transformation, we want to embed an error message for records already present in our database. Here, for demonstration purposes, we will include the error message within our predicate method.

lookupExisting.LinkTo(duplicateCheck, row => !row.IsInDb);
lookupExisting.LinkTo(errorTarget, row => {
    row.ErrorMessage = "Record already exists in target database!";
    return true;
});

Lastly, our objective is to associate an error message with rows identified as redundant by the Distinct transformation. We employ an extra RowTransformation to achieve this:

var addNotDistinctErrorMessage = new RowTransformation<VendorMaster>(
    row => {
        row.ErrorMessage = "Duplicate in source data detected!";
        return row;
    });

duplicateCheck.LinkTo(dbTarget);
duplicateCheck.LinkDuplicatesTo(addNotDistinctErrorMessage);
addNotDistinctErrorMessage.LinkTo(errorTarget);

Checking error file

Upon executing our refined code, the resultant error log file, errors.json, displays the following:

[
  {
    "Name": "RAINBOW SOFTWARE",
    "Code": "",
    "Custom": "",
    "Country": "",
    "Contact": "",
    "Info": "",
    "IsInDb": false,
    "ErrorMessage": "Empty required column detected!"
  },
  {
    "Name": "DATA SOFT UNLIMTED",
    "Code": "D1000",
    "Custom": "DS",
    "Country": "US",
    "Contact": "Rucy",
    "Info": "B",
    "IsInDb": false,
    "ErrorMessage": "Duplicate in source data detected!"
  },
  {
    "Name": "DATA HARD",
    "Code": "D1000",
    "Custom": "DS",
    "Country": "IE",
    "Contact": "Lucy",
    "Info": "C",
    "IsInDb": false,
    "ErrorMessage": "Duplicate in source data detected!"
  },
  {
    "Name": "UNICORN 1",
    "Code": "UNI10",
    "Custom": "U",
    "Country": "IE",
    "Contact": "Trippi",
    "Info": "AS",
    "DbId": 2,
    "IsInDb": true,
    "ErrorMessage": "Record already exists in target database!"
  },
  {
    "Name": "UNICORN 2",
    "Code": "UNI20",
    "Custom": "U",
    "Country": "",
    "Contact": "X",
    "Info": "X",
    "DbId": 3,
    "IsInDb": true,
    "ErrorMessage": "Record already exists in target database!"
  }
]

Incorporating SCD Type 2

Introducing Validity Columns

To accommodate SCD-2, it's essential to introduce ValidFrom and ValidTo date attributes in both our class definition and our SQL table.

In the class:

public class VendorMaster {
    ...
    public DateTime ValidFrom { get; set; } = new DateTime(1900,1,1);
    public DateTime ValidTo { get; set; } = new DateTime(9999,12,31);
}

In the database:

CreateTableTask.CreateIfNotExists(connectionManager, "VendorMaster",
    new List<TableColumn>() {        
        ...
        new TableColumn() { Name = "ValidFrom", DataType = "DATETIME", AllowNulls = false },
        new TableColumn() { Name = "ValidTo", DataType = "DATETIME", AllowNulls = false },
});

It's worth noting that we have set default values corresponding to the minimum and maximum ranges for ValidTo and ValidFrom, respectively.

Furthermore, it's crucial to include these columns in our pre-existing test data:

SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom, Country, Contact, Info, ValidFrom,ValidTo)
        VALUES('BIG HOLDING', 'H1234', 'HD', 'US', 'Hans', 'T0', '1900-1-1','9999-12-31')");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom,Country, ValidFrom,ValidTo)
        VALUES('UNICORN', 'UNI10', 'U', 'NO', '1900-1-1','9999-12-31')");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom,Country, ValidFrom,ValidTo)
        VALUES('UNICORN TWO', 'UNI20', 'U', 'SE', '1900-1-1','9999-12-31')");

Managing New Entries

Previously, our aim was to exclude records that were already present in the database. However, with SCD Type 2, instead of filtering these records, we should introduce them with a fresh ValidFrom/ValidTo range and adjust the existing record's validity period to conclude.

When we identify a pre-existing record, we will designate the ValidFrom attribute to the present timestamp. For this task, we'll utilize a row transformation:

//Adjust ValidFrom to current data if record exists
var adjustValidFrom = new RowTransformation<VendorMaster>(row => {
    if (row.IsInDb)
        row.ValidFrom = DateTime.Now;
    return row;
});

Let's incorporate this transformation into our data flow and eliminate the prior filtering mechanism for extant records:

lookupExisting.LinkTo(adjustValidFrom);
adjustValidFrom.LinkTo(duplicateCheck);

Executing our script now would result in the ValidFrom of already present records being set to the current timestamp. Nonetheless, we're yet to modify the prevailing database record such that its ValidTo corresponds to the new record's ValidFrom. The database destination permits the inclusion of custom logic post writing a data batch. Consequently, we can leverage SQL to align the ValidFrom/ValidTo columns, deploying window functions.

//Modify Merge to work with SCD-2
dbTarget.AfterBatchWrite = batch => {
    SqlTask.ExecuteNonQuery(connectionManager, @"
UPDATE VendorMaster
SET VendorMaster.ValidTo = DATEADD(ss, -1, calc.validto)
FROM (SELECT Id,
             Code,
             Custom,
             ValidFrom,
             LEAD(ValidFrom)
                  OVER (
                      PARTITION BY Code,Custom
                      ORDER BY ValidFrom
                      ) validto
      FROM VendorMaster) calc
WHERE VendorMaster.Id = calc.Id
  AND calc.validto IS NOT NULL
");
};

Batch-Specific Updates

It's important to acknowledge that the above SQL code updates all entries across the entire table. For brevity, the SQL script isn't restricted solely to the batch-specific entries. If you desire to limit the updates to the current batch's scope, the batch variable lets you interact with the objects unique to that batch, encompassing all incoming data. Thus, if such a specification is of interest, you might consider integrating the Id column to refine your SQL. Currently, the DbId property stores the Id values, but only for entries that match the lookup. For a comprehensive retrieval of all id values, consider reading back the auto-generated ids from your database:

[DbColumnMap("Id")]
[ValueGenerationColumn]
public int? DbId { get; set; }

With this setup, the batch variable grants access to all auto-generated Id values.

Complete code overview

Here is an overview of the whole code of part 2 of the example. It is also available on GitHub.

//Adjust connection string to your local database
string SqlConnectionString = @"Data Source=localhost;User Id=sa;Password=YourStrong@Passw0rd;Initial Catalog=demo;TrustServerCertificate=true";

//Prepare the target database
SqlConnectionManager connectionManager = new SqlConnectionManager(SqlConnectionString);
DropTableTask.DropIfExists(connectionManager, "VendorMaster");
CreateTableTask.CreateIfNotExists(connectionManager, "VendorMaster",
    new List<TableColumn>() {
        new TableColumn() { Name = "Id", DataType = "INT", AllowNulls = false, IsPrimaryKey = true, IsIdentity = true },
        new TableColumn() { Name = "VendorName", DataType = "VARCHAR(100)", AllowNulls = false },
        new TableColumn() { Name = "Code", DataType = "CHAR(5)", AllowNulls = false },
        new TableColumn() { Name = "Custom", DataType = "VARCHAR(10)", AllowNulls = false },
        new TableColumn() { Name = "Country", DataType = "CHAR(2)", AllowNulls = true },
        new TableColumn() { Name = "Contact", DataType = "VARCHAR(50)", AllowNulls = true },
        new TableColumn() { Name = "Info", DataType = "VARCHAR(50)", AllowNulls = true },
        new TableColumn() { Name = "ValidFrom", DataType = "DATETIME", AllowNulls = false },
        new TableColumn() { Name = "ValidTo", DataType = "DATETIME", AllowNulls = false },
    });
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom, Country, Contact, Info, ValidFrom,ValidTo)
        VALUES('BIG HOLDING', 'H1234', 'HD', 'US', 'Hans', 'T0', '1900-1-1','9999-12-31')");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom,Country, ValidFrom,ValidTo)
        VALUES('UNICORN', 'UNI10', 'U', 'NO', '1900-1-1','9999-12-31')");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom,Country, ValidFrom,ValidTo)
        VALUES('UNICORN TWO', 'UNI20', 'U', 'SE', '1900-1-1','9999-12-31')");


//Create the data flow components
var source = new CsvSource<VendorMaster>("sourceData.csv");
var normalize = new RowTransformation<VendorMaster>(row => row.Normalize());
var errorTarget = new JsonDestination<VendorMaster>("errors.json");
var dbTarget = new DbDestination<VendorMaster>(connectionManager, "VendorMaster");
var duplicateCheck = new Distinct<VendorMaster>();
var lookupExisting = new LookupTransformation<VendorMaster, VendorMasterDbEntry>();
var lookupSource = new DbSource<VendorMasterDbEntry>(connectionManager, "VendorMaster");
lookupExisting.Source = lookupSource;

//Adjust ValidFrom to current data if record exists
var adjustValidFrom = new RowTransformation<VendorMaster>(row => {
    if (row.IsInDb)
        row.ValidFrom = DateTime.Now;
    return row;
});

//Add custom error message to duplicates
var addNotDistinctErrorMessage = new RowTransformation<VendorMaster>(
    row => {
        row.ErrorMessage = "Duplicate in source data detected!";
        return row;
    });

//Modify Merge to work with SCD-2
dbTarget.AfterBatchWrite = batch => {
    SqlTask.ExecuteNonQuery(connectionManager, @"
UPDATE VendorMaster
SET VendorMaster.ValidTo = calc.validto
FROM (SELECT Id,
             Code,
             Custom,
             ValidFrom,
             LEAD(ValidFrom)
                  OVER (
                      PARTITION BY Code,Custom
                      ORDER BY ValidFrom
                      ) validto
      FROM VendorMaster) calc
WHERE VendorMaster.Id = calc.Id
  AND calc.validto IS NOT NULL
");
};


/* Linking the components */
source.LinkTo(normalize);

normalize.LinkTo(lookupExisting, row => row.IsValid());
normalize.LinkTo(errorTarget, row => true); //Valid rows are already send to the lookup!

lookupExisting.LinkTo(adjustValidFrom);
adjustValidFrom.LinkTo(duplicateCheck);

duplicateCheck.LinkTo(dbTarget);
duplicateCheck.LinkDuplicatesTo(addNotDistinctErrorMessage);
addNotDistinctErrorMessage.LinkTo(errorTarget);

//Execute the network
await Network.ExecuteAsync(source);

Console.WriteLine("Done!");

public class VendorMaster
{

    [DbColumnMap("VendorName")]
    public string Name { get; set; }

    [DistinctColumn]        
    public string Code { get; set; }

    [DistinctColumn]        
    public string Custom { get; set; }

    public string Country { get; set; }

    public string Contact { get; set; }

    [CsvHelper.Configuration.Attributes.Name("TraceInfo")]
    public string Info { get; set; }

    [DbColumnMap("Id")]
    [ValueGenerationColumn]
    public int? DbId { get; set; }
    public bool IsInDb => DbId.HasValue && DbId > 0;

    public VendorMaster Normalize() {
        this.Name = Name.Trim().ToUpper();
        this.Contact = Contact.Trim();
        return this;
    }

    public bool IsValid() {

        if (string.IsNullOrEmpty(Name) || string.IsNullOrEmpty(Code) || string.IsNullOrEmpty(Custom)) {
            ErrorMessage = "Empty required column detected!";
            return false;
        }
        if (Name.Length < 5 || Name.Length > 50) {
            ErrorMessage = "Name has unsupported length!";
            return false;
        }
        if (Code.Length != 5) {
            ErrorMessage = "Code has unsupported length!";
            return false;
        }
        return true;
    }

    public string ErrorMessage { get; set; }

    public DateTime ValidFrom { get; set; } = new DateTime(1900,1,1);
    public DateTime ValidTo { get; set; } = new DateTime(9999,12,31);
}

public class VendorMasterDbEntry
{
    [RetrieveColumn(nameof(VendorMaster.DbId))]
    public int Id { get; set; }
    [MatchColumn(nameof(VendorMaster.Code))]
    public string Code { get; set; }
    [MatchColumn(nameof(VendorMaster.Custom))]
    public string Custom { get; set; }
}

Conclusion

We've made big steps to improve our data process. First, we added error messages. This means if something goes wrong with the data, we'll know about it right away and can fix it. Then, we started using SCD Type 2. With this, we can see the history of our data, which is great. We can track any changes and understand why they happened. These changes don’t just make our data look good; they also make our whole process more reliable and easy to follow. We're now set up for success!

Up Next: Configurable Data Flows

So far, we've enhanced error logging and integrated the SCD Type 2 methodology. In Part 3, we're taking it up a notch. Let's challenge the constraints of static types and aim for a dataflow that's entirely configurable. This means more flexibility, scalability, and adaptability without the rewrites. Stay tuned for a dynamic turn in our data management journey!

Redefining ETL: Data Flows Powered by C# (Part I)

Andreas Lennartz — Wed, 11 Oct 2023 18:38:17 +0000

Overview

In the world of data management, ETL is critical. It extracts data from a plethora of sources, transforms it into a desired format, and loads it into a target system like a data warehouse or database. This is where ETLBox shines. This nimble yet robust .NET library doesn’t come with a User Interface, setting it apart from other ETL tools like SQLServer Integration Services (SSIS) or Azure Data Factory. But no worries! This emphasizes the advantage of crafting data flows programmatically – a genuine game-changer for numerous ETL and data integration challenges.

The Nitty-Gritty

ETLBox simplifies the extraction, transformation, and loading (ETL) of data, supporting a wide range of data sources and targets and provides a set of straightforward, yet powerful components for automating these tasks. Here’s what ETLBox brings to the table:

Simplicity: With its user-friendly API, navigating the ETL processes becomes a breeze.
Flexibility: Adapt various data transformations including lookups, column splitting, and data aggregation with ETLBox’s versatile features.
Performance: Stand out in performance with ETLBox’s guarantee of high-speed and efficient handling of substantial data volumes.
.NET Support: Fully integrated with the .NET framework and written in C#, ETLBox is a natural fit for .NET developers.
Test-driven: ETLBox is the only ETL framework that supports Test-driven development right from the start, underscoring its commitment to robust and reliable ETL processes.

In a nutshell, ETLBox is not just a tool. It’s a comprehensive solution for your specific ETL needs. It allows developers to use .NET to code their ETL job, ensuring seamless data management from extraction to loading.

Understanding ETL

ETL is the foundation of modern Business Intelligence applications. It separates the analysis part from everything that comes before it. ETL, standing for extract, transform, and load, embodies a three-step process:

Extracting: Gathering data from a specific source in its rawest form, ensuring all relevant data is gathered.
Transforming: Cleaning and harmonizing the data, making it easy for the end user to read, and ensuring high data quality by removing duplicates and sorting data into relevant categories.
Loading: The ETL tool then harmonizes your data and loads it into a target location, making it ready for analysis.

What is a Data Flow?

ETLBox empowers you to construct an efficient ETL pipeline with its comprehensive set of Data Flow components. In constructing a Data Flow pipeline with ETLBox, define, link, and execute the components. This process allows for asynchronous data processing, optimizing flow and memory usage. Direct connections between sources and destinations are possible, providing flexibility and efficiency. Components process data as it’s read, promoting continuous data movement and processing, and enabling optimized throughput for various operations.

How to Get ETLBox

ETLBox, developed in C#, targets .NET Standard 2.0 and higher, ensuring compatibility with various .NET versions, including .NET Framework 4.X and above, .NET Core 2.X and above, and .NET 5.0 and 6.0. All ETLBox packages are hosted on NuGet. Installing the core package is essential for working with ETLBox as it contains default connectors and transformations. Integrate it into your project using the NuGet package manager with the command: dotnet add package ETLBox. Choose the appropriate connector package based on your needs, for connecting to SQLServer databases or loading data from a Json file or REST endpoint, for example. Visit the project home page for more information.

Data Flow by Example

In this example, let's assume we receive supplier data in a file daily. As a first step, we want to copy these data to a database table.

Preparation

First, we need an example file with supplier data. In our example, we use the CSV format, but essentially, ETLBox supports every other common format, including proprietary ones. Should the data source change later on, like the supplier data coming from a REST service instead of a file, this can be adjusted relatively quickly and easily without affecting the rest of the data flows.

CSV File

Below is the content of the CSV file:

Name,Code,Custom,Country,Contact,TraceInfo
"HoneyPot Inc ",CO110,HP,UK,"Peter",XC
"AppleBees Corp",CO200,AB,DE,"Paul",
" Peach United",CO301,PU,CH,"Mary",ED
"Rainbow Software",,,,
"Data Soft Ltd.",D1000,DS,XX,"Ruby",A
"Data Soft Unlimted",D1000,DS,US,"Rucy",B
"Data Hard",D1000,DS,IE,"Lucy",C
"Unicorn 1",UNI10,U,IE,"Trippi",AS
"Unicorn 2",UNI20,U,,X,X

Database Table

We also need a target table, let’s call it VendorMaster. Let’s create three entries in this table:

VendorName	Code	Custom	Country	Contact	Info
Big Holding	H1234	HD	US	Hans	T0
Unicorn	UNI10	U	NO
Another Unicorn	UNI20	U	SE

Here is the code to create this table:

//Adjust connection string to your local database
string SqlConnectionString = @"Data Source=localhost;User Id=sa;Password=YourStrong@Passw0rd;Initial Catalog=demo;TrustServerCertificate=true";

//Prepare the target database
SqlConnectionManager connectionManager = new SqlConnectionManager(SqlConnectionString);
DropTableTask.DropIfExists(connectionManager, "VendorMaster");
CreateTableTask.CreateIfNotExists(connectionManager, "VendorMaster",
    new List<TableColumn>() {
        new TableColumn() { Name = "Id", DataType = "INT", AllowNulls = false, IsPrimaryKey = true, IsIdentity = true },
        new TableColumn() { Name = "VendorName", DataType = "VARCHAR(100)", AllowNulls = false },
        new TableColumn() { Name = "Code", DataType = "CHAR(5)", AllowNulls = false },
        new TableColumn() { Name = "Custom", DataType = "VARCHAR(10)", AllowNulls = false },
        new TableColumn() { Name = "Country", DataType = "CHAR(3)", AllowNulls = true },
        new TableColumn() { Name = "Contact", DataType = "VARCHAR(50)", AllowNulls = true },
        new TableColumn() { Name = "Info", DataType = "VARCHAR(50)", AllowNulls = true },
    });
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom, Country, Contact, Info)
        VALUES('Big Holding', 'H1234', 'HD', 'US', 'Hans', 'T0')");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom) 
        VALUES('Unicorn', 'UNI10', 'U', 'NO')");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom) 
        VALUES('Another Unicorn', 'UNI20', 'U', 'SE')");

If someone does not have a database installed, it can be started in a container using Docker Desktop with just a few commands. See also: https://hub.docker.com/_/microsoft-mssql-server

Requirements for the Data Flow

In this section, we will build an ETL data flow with ETLBox. One of the ETL steps in loading data is cleansing and checking for duplicates.

In our data flow, we first want to check the data for validity. The fields Name, Code, and Custom must not be empty. The Name must be between 5 and 50 characters, and the Code should be exactly 5 characters long.

We also want to clean the data. We want to trim any additional spaces at the beginning and end of the Name and Code fields. All vendor names should be stored in uppercase.

Next, we want to check the data for duplicates. We want to ensure that there are no entries in the target database with the same combination of the Code/Custom columns. There should also be no entries within the file with the same Code & Custom combination.

We want to write all faulty records to a log file. In this example, we choose JSON as the target format.

Creating the Data Model

For the first step, we'll set up a class that represents our data model.

public class VendorMaster
{

    [DbColumnMap("VendorName")]
    public string Name { get; set; }

    [DistinctColumn]
    public string Code { get; set; }

    [DistinctColumn]
    public string Custom { get; set; }

    public string Country { get; set; }

    public string Contact { get; set; }

    [CsvHelper.Configuration.Attributes.Name("TraceInfo")]
    public string Info { get; set; }

    public int? DbId { get; set; }
    public bool IsInDb => DbId.HasValue && DbId > 0;

    public VendorMaster Normalize() {
        this.Name = Name.Trim().ToUpper();
        this.Contact = Contact.Trim();
        return this;
    }

    public bool IsValid() {

        if (string.IsNullOrEmpty(Name) || string.IsNullOrEmpty(Code) || string.IsNullOrEmpty(Custom))
            return false;
        if (Name.Length < 5 || Name.Length > 50)
            return false;
        if (Code.Length != 5)
            return false;
        return true;
    }
}

This class is already provided with additional information:

We use DbColumnMap("VendorName") to indicate that the database name is "VendorName", but we receive Name in the CSV file.
The DistinctColumn attribute specifies the two fields that must not be duplicated.
For the TraceInfo field, we approach it differently than with Name - we already specify the field as Info, and with the help of the attribute CsvHelper.Configuration.Attributes.Name, it gets assigned to the TraceInfo column in the CSV file.
Additionally, we've implemented two methods: one to normalize the data and another to validate its correctness.

Defining the Data Flow Components

To meet these requirements with a data flow, we need various components. We can read the data from the CSV file with a CsvSource.

Here is an overall view:

var source = new CsvSource<VendorMaster>("sourceData.csv");
var normalize = new RowTransformation<VendorMaster>(row => row.Normalize());
var errorTarget = new JsonDestination<VendorMaster>("errors.json");
var dbTarget = new DbDestination<VendorMaster>(connectionManager, "VendorMaster");
var duplicateCheck = new Distinct<VendorMaster>();
var lookupExisting = new LookupTransformation<VendorMaster, VendorMasterDbEntry>();
var lookupSource = new DbSource<VendorMasterDbEntry>(connectionManager, "VendorMaster");
lookupExisting.Source = lookupSource;

With the CsvSource, we determine the source of our data flow. The RowTransformation then takes each incoming row and applies the Normalize function. The errorTarget defines the file "errors.json" as a Json destination. The dbTarget describes our target table in the database. The Distinct transformation checks for duplicates based on the DistinctColumn attribute. The LookupTransformation loads the data from the dbSource and determines an already existing database ID.

Linking the Data Flow Components

Now, we need to connect these components. This is accomplished by using the LinkTo method, which routes data from one component to the next when the data flow is started. If there’s no filtering condition (predicate), data moves unfiltered.

source.LinkTo(normalize);

normalize.LinkTo(lookupExisting, row => row.IsValid());
normalize.LinkTo(errorTarget, row => !row.IsValid());

lookupExisting.LinkTo(duplicateCheck, row => !row.IsInDb);
lookupExisting.LinkTo(errorTarget, row => row.IsInDb);

duplicateCheck.LinkTo(dbTarget);
duplicateCheck.LinkDuplicatesTo(errorTarget);

Predicates

When linking components in a data flow, you have the option to add a filter expression to the link, known as a predicate. This filter expression is assessed for every row that passes through the link. If the expression evaluates to true, the data will continue to the linked component. If it evaluates to false, the data flow will attempt the next link to transmit its data through.

This example code will split up the data from the source and send it into three different destinations, depending on their Value:

source.LinkTo(dest1, row => row.Value == "A");
source.LinkTo(dest2, row => row.Value == "B");
source.LinkTo(dest3, row => row.Value != "A" && row.Value != "B");

This mechanism allows for the partitioning of data from the source, sending it into various destinations based on their values. The data will be sent only into one of the connected links. If there is more than one link, the first link that either has no predicate or whose predicate returns true is used. If you wish to duplicate your data, consider adding a Multicast into your flow.

Ensure that all records are sent to any type of destination. If not all records arrive at a destination, your data flow may never complete as it will wait for all records to be processed.

Running the Network

Finally, the entire network is initiated with just a simple command:

await Network.ExecuteAsync(source);

Note: Regardless of whether a network is started synchronously with Execute or asynchronously with ExecuteAsync, the individual components are always run asynchronously. This means that data from the CSV source is read concurrently while some of the data is already being processed or written into the target table. In our example, this might not play a significant role due to the small amount of data. However, with a larger dataset (e.g., 1 million suppliers), this concurrent execution becomes noticeable.

Result

After executing, we obtain the following data in our target table:

Id	VendorName	Code	Custom	Country	Contact	Info
1	BIG HOLDING	H1234	HD	US	Hans	T0
2	UNICORN	UNI10	U	NO
3	UNICORN TWO	UNI20	U	SE
4	HONEYPOT INC	CO110	HP	UK	Peter	XC
5	APPLEBEES CORP	CO200	AB	DE	Paul
6	PEACH UNITED	CO301	PU	CH	Mary	ED
7	DATA SOFT LTD.	D1000	DS	XX	Ruby	A

The Errors.Json file contains the following content:

[
  {
    "Name": "RAINBOW SOFTWARE",
    "Code": "",
    "Custom": "",
    "Country": "",
    "Contact": "",
    "Info": "",
    "IsInDb": false
  },
  {
    "Name": "DATA SOFT UNLIMTED",
    "Code": "D1000",
    "Custom": "DS",
    "Country": "US",
    "Contact": "Rucy",
    "Info": "B",
    "IsInDb": false
  },
  {
    "Name": "DATA HARD",
    "Code": "D1000",
    "Custom": "DS",
    "Country": "IE",
    "Contact": "Lucy",
    "Info": "C",
    "IsInDb": false
  },
  {
    "Name": "UNICORN 1",
    "Code": "UNI10",
    "Custom": "U",
    "Country": "IE",
    "Contact": "Trippi",
    "Info": "AS",
    "DbId": 2,
    "IsInDb": true
  },
  {
    "Name": "UNICORN 2",
    "Code": "UNI20",
    "Custom": "U",
    "Country": "",
    "Contact": "X",
    "Info": "X",
    "DbId": 3,
    "IsInDb": true
  }
]

Our data flow has successfully read the CSV files, normalized, and cleansed them. Duplicates within the file and the database were identified. The error file contains the filtered data.

Complete code overview

Here is an overview of the whole code the example. It is also available on GitHub.

//Adjust connection string to your local database
string SqlConnectionString = @"Data Source=localhost;User Id=sa;Password=YourStrong@Passw0rd;Initial Catalog=demo;TrustServerCertificate=true";

//Prepare the target database
SqlConnectionManager connectionManager = new SqlConnectionManager(SqlConnectionString);
DropTableTask.DropIfExists(connectionManager, "VendorMaster");
CreateTableTask.CreateIfNotExists(connectionManager, "VendorMaster",
    new List<TableColumn>() {
        new TableColumn() { Name = "Id", DataType = "INT", AllowNulls = false, IsPrimaryKey = true, IsIdentity = true },
        new TableColumn() { Name = "VendorName", DataType = "VARCHAR(100)", AllowNulls = false },
        new TableColumn() { Name = "Code", DataType = "CHAR(5)", AllowNulls = false },
        new TableColumn() { Name = "Custom", DataType = "VARCHAR(10)", AllowNulls = false },
        new TableColumn() { Name = "Country", DataType = "CHAR(2)", AllowNulls = true },
        new TableColumn() { Name = "Contact", DataType = "VARCHAR(50)", AllowNulls = true },
        new TableColumn() { Name = "Info", DataType = "VARCHAR(50)", AllowNulls = true },
    });
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom, Country, Contact, Info)
        VALUES('BIG HOLDING', 'H1234', 'HD', 'US', 'Hans', 'T0')");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom,Country) 
        VALUES('UNICORN', 'UNI10', 'U', 'NO')");
SqlTask.ExecuteNonQuery(connectionManager,
    @"INSERT INTO VendorMaster (VendorName, Code, Custom,Country) 
        VALUES('UNICORN TWO', 'UNI20', 'U', 'SE')");


//Create the data flow components
var source = new CsvSource<VendorMaster>("sourceData.csv");
var normalize = new RowTransformation<VendorMaster>(row => row.Normalize());
var errorTarget = new JsonDestination<VendorMaster>("errors.json");
var dbTarget = new DbDestination<VendorMaster>(connectionManager, "VendorMaster");
var duplicateCheck = new Distinct<VendorMaster>();
var lookupExisting = new LookupTransformation<VendorMaster, VendorMasterDbEntry>();
var lookupSource = new DbSource<VendorMasterDbEntry>(connectionManager, "VendorMaster");
lookupExisting.Source = lookupSource;

/* Linking the components */
source.LinkTo(normalize);

normalize.LinkTo(lookupExisting, row => row.IsValid());
normalize.LinkTo(errorTarget, row => !row.IsValid());

lookupExisting.LinkTo(duplicateCheck, row => !row.IsInDb);
lookupExisting.LinkTo(errorTarget, row => row.IsInDb);

duplicateCheck.LinkTo(dbTarget);
duplicateCheck.LinkDuplicatesTo(errorTarget);

//Execute the network
await Network.ExecuteAsync(source);

Console.WriteLine("Done!");

public class VendorMaster
{

    [DbColumnMap("VendorName")]
    public string Name { get; set; }

    [DistinctColumn]
    public string Code { get; set; }

    [DistinctColumn]
    public string Custom { get; set; }

    public string Country { get; set; }

    public string Contact { get; set; }

    [CsvHelper.Configuration.Attributes.Name("TraceInfo")]
    public string Info { get; set; }

    public int? DbId { get; set; }
    public bool IsInDb => DbId.HasValue && DbId > 0;

    public VendorMaster Normalize() {
        this.Name = Name.Trim().ToUpper();
        this.Contact = Contact.Trim();
        return this;
    }

    public bool IsValid() {

        if (string.IsNullOrEmpty(Name) || string.IsNullOrEmpty(Code) || string.IsNullOrEmpty(Custom))
            return false;
        if (Name.Length < 5 || Name.Length > 50)
            return false;
        if (Code.Length != 5)
            return false;
        return true;
    }

}

public class VendorMasterDbEntry
{
    [RetrieveColumn(nameof(VendorMaster.DbId))]
    public int Id { get; set; }
    [MatchColumn(nameof(VendorMaster.Code))]
    public string Code { get; set; }
    [MatchColumn(nameof(VendorMaster.Custom))]
    public string Custom { get; set; }
}

Conclusion

The ETLBox library provides a comprehensive set of tools for building efficient and scalable ETL solutions within the .NET ecosystem. It allows developers to define complex data flow pipelines with ease, providing robust error handling, data transformation, and validation features out-of-the-box. It offers:

Flexibility and Adaptability: It can connect to various data sources and targets, facilitating seamless data integration across diverse platforms and technologies.
Programmability: It lets developers code their ETL process using .NET, providing more control and customization over the ETL processes.
Performance: It handles substantial data volumes efficiently, ensuring high-speed data processing and loading.
Testability: Being a test-driven framework, it emphasizes the importance of robust and reliable ETL processes, allowing developers to ensure the correctness and efficiency of their ETL workflows.

In our example, we built a small ETL pipeline to process a CSV file, perform data validation and normalization, and load the cleaned data into a database, handling errors and duplicates gracefully. The process involved defining a data model, setting up various data flow components, linking them together, and executing the network.

Next article: Adding SCD Type 2

In the next step, we will turn to improvements of this data flow. In addition to enhanced error logging, we will address the issue of how to reconcile data when repeatedly loading different or the same supplier data. We aim to build a Slowly Changing Dimension Type 2 (SCD Type 2) that stores the corresponding changes in the CSV data.

Forem: Andreas Lennartz

High-Performance Bulk Operations for Dapper and ADO.NET

What Is DbExtensions?

Key Features

How DbExtensions Works

Bulk Operations Overview

1. Bulk Insert

2. Bulk Update

3. Bulk Delete

4. Bulk Merge

Best Practices: Using DbExtensions with Dapper

Complementary Responsibilities

ETL-Style Workflows

Performance Considerations

Getting Started

Sample Code

Conclusion

Why CRUD Doesn’t Work for ETL (And What to Do Instead)

From CRUD to Dataflow: Optimizing ETL Performance with ETLBox

About ETLBox

Handling "Create" Operations

Preprocessing with CreateTableTask

Postprocessing: Index Rebuilding

Avoiding Single-Row "Read" Operations

The Problem with Row-by-Row Reads

Example of an Inefficient Approach

LookupTransformation: A Better Approach

Using LookupTransformation

Choosing the Right Caching Strategy

Using Dataflow Sources

DbSource: Streaming Data Efficiently

Example: Using DbSource to Read from a Database

Using Custom SQL for More Control

Integrating Stored Procedure Results

Combining Dataflows with DbSource

Efficient "Write" Operations: Bulk Inserts

Bulk Inserts with DbDestination

Example: Using DbDestination for Bulk Inserts

Merging or Upserts with DbMerge

Example: Using DbMerge to Sync Delta Data

Choosing Between Bulk Insert and Merge

Rethinking CRUD in ETL

Mastering Database Merging: Comparing Different Approaches

Basic Example Scenario

Using a Staging Table

SQL Example for Inserts and Updates

UPSERT Example for MySQL/MariaDB

Handling Deletes

Importing Data

Using BULK INSERT for SQL Server

Data Cleansing

ETL Code with C

ETL into Staging Table

ETL with DbMerge

Database Transactions

Performance Improvements

Using Smaller Batches

Full Load vs. Delta Load

Change Data Capture (CDC)

Handling Non-CDC Sources

Truncate Target Table

Removing and Adding Indexes and Constraints

Enhancing the DbMerge

Using Delta with DbMerge

Cache Mode

Truncating Target

Delta Table

Going Further: Building Dynamic Data Flows

Example Use Case

Conclusion

Unlock the Power of C# in Polyglot Notebooks

C# and Polyglot Notebooks

The Use-case of ETL in a Polyglot Notebook

How to Create Your Own Polyglot Notebook

Setting Up a Database Table for ETL Using C# in a Polyglot Notebook

Loading Data using an ETL Pipeline

Displaying and Analyzing Data

Displaying the DataFrame

Transforming the DataFrame

Visualizing Data with ScottPlot

Using `BULK INSERT` for SQL Server