NPipeline.Connectors.MySql 0.54.0

NPipeline.Connectors.MySQL

A fully-async MySQL and MariaDB connector for NPipeline, built on MySqlConnector (MIT).

Installation

dotnet add package NPipeline.Connectors.MySQL

Quick Start

using NPipeline.Connectors.MySql.Nodes;

// Source: read rows from MySQL
var source = new MySqlSourceNode<Product>(
    connectionString: "Server=localhost;Database=shop;User=root;Password=root;",
    query: "SELECT * FROM `products`");

// Sink: write rows to MySQL
var sink = new MySqlSinkNode<Product>(
    connectionString: "Server=localhost;Database=shop;User=root;Password=root;",
    tableName: "products");

Dependency Injection

using Microsoft.Extensions.DependencyInjection;
using NPipeline.Connectors.MySql.DependencyInjection;

services.AddMySqlConnector(options =>
{
    options.DefaultConnectionString =
        "Server=localhost;Database=shop;User=root;Password=root;";

    options.AddOrUpdateConnection("analytics",
        "Server=analytics-host;Database=analytics;User=etl;Password=secret;");

    options.DefaultConfiguration = new MySqlConfiguration
    {
        MinPoolSize = 2,
        MaxPoolSize = 20,
        MaxRetryAttempts = 3,
        RetryDelay = TimeSpan.FromSeconds(2),
    };
});

Attribute Mapping

using NPipeline.Connectors.MySql.Mapping;
using NPipeline.Connectors.Attributes;

[MySqlTable("products")]
public class Product
{
    [MySqlColumn("product_id", AutoIncrement = true)]
    public int Id { get; set; }

    [Column("name")]
    public string Name { get; set; } = string.Empty;

    [MySqlColumn("unit_price")]
    public decimal Price { get; set; }

    [IgnoreColumn]
    public bool InStock { get; set; }
}

Write Strategies

Configure via MySqlConfiguration.WriteStrategy:

var config = new MySqlConfiguration
{
    WriteStrategy = MySqlWriteStrategy.Batch,
    BatchSize = 500,
};
var sink = new MySqlSinkNode<Product>(connectionString, "products", config);

Upsert

var config = new MySqlConfiguration
{
    UseUpsert = true,
    UpsertKeyColumns = ["product_id"],
    OnDuplicateKeyAction = OnDuplicateKeyAction.Update,   // or .Ignore / .Replace
};
var sink = new MySqlSinkNode<Product>(connectionString, "products", config);

Generated SQL example:

INSERT INTO `products` (`product_id`, `name`, `unit_price`)
VALUES (@p0, @p1, @p2)
ON DUPLICATE KEY UPDATE `name` = VALUES(`name`), `unit_price` = VALUES(`unit_price`);

StorageUri

// mysql:// or mariadb:// schemes are both supported
var uri = StorageUri.Parse("mysql://root:root@localhost:3306/shop");
var source = new MySqlSourceNode<Product>(uri, "SELECT * FROM `products`");
var sink   = new MySqlSinkNode<Product>(uri, "products");

Configuration Reference

Transient Error Handling

The connector automatically retries on the following MySQL error codes:

Checkpointing

var config = new MySqlConfiguration
{
    CheckpointStrategy = CheckpointStrategy.KeyBased,
    CheckpointColumn  = "updated_at",
};

Supported strategies: None, InMemory, Offset, KeyBased, Cursor, CDC.

Custom Row Mapper

var source = new MySqlSourceNode<Product>(
    connectionString,
    "SELECT product_id, name FROM `products`",
    row => new Product
    {
        Id   = row.Get<int>("product_id"),
        Name = row.Get<string>("name") ?? string.Empty,
    });

MariaDB Support

Both mysql:// and mariadb:// StorageUri schemes resolve to MySqlDatabaseStorageProvider. The MySqlConnector driver is fully compatible with MariaDB 10.5+.

License

This package is licensed under the Business Source License 1.1.

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