Custom Data Source

It is possible to fetch data from a custom source such as an API, other database, file system, etc.

There are multiple ways:

Implement GenericDataSource(Async) - Simplified source which should return C# objects. This is the recomended way to start implementing a source.
Implement ReadBaseOperator - This allows the creation of a low-level read operator, where serialization, state storage, watermarks and checkpointing must be handled.

Generic Data Source

The generic data source allows easy implementation against custom sources that returns C# objects.

It allows:

Full batch reloads, where all the data is imported again and delta is computed.
Delta loads, where delta should be returned.
Custom watermark provided by the source.
Scheduling of full batch and delta reloads.

There are two classes that can be implemented for the generic data source:

GenericDataSourceAsync - Data is returned by an IAsyncEnumerable, this should be used with remote sources.
GenericDataSource - Data is returned by an IEnumerable, which should be used in cases where data is already in memory.

When implementing a generic data source, it is important to think about memory usage, for instance, do not fetch all rows and store them in memory and then return them, this can cause huge memory spikes or out of memory. Instead yield return values and fetch the data in batches. The operator stores the data in B+ trees that will be temporarily stored on disk if the memory usage is too high.

Implementation example

public class ExampleDataSource : GenericDataSourceAsync<User>
{
    private readonly IUserRepository _userRepository;

    public ExampleDataSource(IUserRepository userRepository) 
    {
        _userRepository = userRepository;
    }

    // Fetch delta every 1 second
    public override TimeSpan? DeltaLoadInterval => TimeSpan.FromSeconds(1);

    // Reload all data every 1 hours, this is not required, but can be useful.
    // If for instance deletes cant be found in deltas from the source,
    // a full reload would find all deleted rows.
    public override TimeSpan? FullLoadInterval => TimeSpan.FromHours(1);

    protected override IEnumerable<FlowtideGenericObject<User>> DeltaLoadAsync(long lastWatermark)
    {
        var changes = _userRepository.GetChangesFromWatermarkAsync(lastWatermark);

        await foreach(var change in changes) {
            yield return new FlowtideGenericObject<User>(change.Id, change, change.Timestamp);
        }
    }

    protected override IEnumerable<FlowtideGenericObject<User>> FullLoadAsync()
    {
        var data = _userRepository.GetAllDataAsync(lastWatermark);
        
        await foreach(var row in data) {
            yield return new FlowtideGenericObject<User>(row.Id, row, row.Timestamp);
        }
    }
}

To use your data source, add the following to the ConnectorManager:

connectorManager.AddCustomSource(
    "{the table name}", 
    (readRelation) => new ExampleDataSource(userRepository));

All rows must return an identifier that is used when calculating changes of the data. This key can be accessed when querying with the special name of __key.

Push-based delta loading

Instead of polling with DeltaLoadAsync, it is possible to use a push-based model where the source controls when data is submitted. This can be useful for sources such as message queues, webhooks or event streams.

To use push-based loading, override IsPushBased to return true and implement RunDeltaLoad:

public class EventDrivenSource : GenericDataSourceAsync<User>
{
    private readonly Channel<User> _channel = Channel.CreateUnbounded<User>();

    public override bool IsPushBased => true;

    // Set to null to disable interval-based triggering
    public override TimeSpan? DeltaLoadInterval => null;

    public override async Task RunDeltaLoad(
        IDeltaLoadContext<User> context,
        CancellationToken cancellationToken)
    {
        while (await _channel.Reader.WaitToReadAsync(cancellationToken))
        {
            // Begin a transaction, this acquires the checkpoint lock
            // ensuring all items in this batch are included in the same checkpoint.
            await using var tx = await context.BeginTransactionAsync(cancellationToken);

            while (_channel.Reader.TryRead(out var user))
            {
                var obj = new FlowtideGenericObject<User>(
                    user.Id, user, DateTimeOffset.UtcNow.ToUnixTimeMilliseconds(), false);
                await tx.SubmitAsync(obj, cancellationToken);
            }
            // Disposing the transaction releases the checkpoint lock
        }
    }

    public override IAsyncEnumerable<FlowtideGenericObject<User>> FullLoadAsync()
    {
        // Return initial data...
    }
}

RunDeltaLoad is a long-running method that controls its own loop, instead of returning an enumerable per interval. Data is submitted through transactions which manage the checkpoint lock, ensuring all items in one transaction end up in the same checkpoint.

Trigger data reloads programatically

The generic data source also registers triggers that allows the user to notify the stream when a reload should happen.

The following triggers are registered:

full_load - Does a full load on all running generic data sources
delta_load - Does a delta load on all running generic data sources
full_load_{tableName} - Full load for a specific source
delta_load_{tableName} - Delta load for a specific source

Example on calling a trigger:

await stream.CallTrigger("delta_load", default);

Calling the triggers programatically can be useful if having an interval would cause too much latency for the data.

SQL Table Provider

There is a table provider as well for the generic data source, that can be used to easily import table metadata from a class.

Example:

sqlPlanBuilder.AddGenericDataTable<User>("users");

If you are starting Flowtide with dependency injection, a table provider is added automatically, so this step is not required.

Generic data sink

The generic data sink allow the implementation of a sink that reads the rows as C# classes. This limits the stream to only send those specific columns, but it can be useful im cases such as API integrations where there is a strict schema.

Implementation example

Create a class that inherits from GenericDataSink.

internal class TestDataSink : GenericDataSink<User>
{
    public override Task<List<string>> GetPrimaryKeyNames()
    {
        return Task.FromResult(new List<string> { "{primaryKeyColumnName}" });
    }

    public override async Task OnChanges(IAsyncEnumerable<FlowtideGenericWriteObject<User>> changes, Watermark watermark, bool isInitialData, CancellationToken cancellationToken)
    {
        await foreach(var userChange in changes)
        {
            if (!userChange.IsDeleted)
            {
                // Do upsert to destination
            }
            else
            {
                // Do delete against destination
            }
        }
    }
}

Add the generic data sink to the ConnectorManager:

connectorManager.AddCustomSink("{tableName}", (rel) => new testDataSink());

Fetching existing data

On initial startup, fetching existing data from the destination is optional. If you do not override this method, Flowtide assumes the destination is empty and simply sends all current stream data to OnChanges as initial inserts.

However, if you want Flowtide to calculate a delta based on the current state of the destination system (e.g., to only push changes, and to emit delete operations for any rows in the destination that are no longer in the incoming stream), you can override the GetExistingData method:

public override IAsyncEnumerable<User> GetExistingData()
{
    // Fetch all existing users from the destination
    return _userRepository.GetAllUsersAsync();
}

If you override this method:

During initial load, Flowtide compares the existing destination data with the incoming stream data.
It will automatically trigger insert/update operations for new or changed data, and delete operations for any rows in the destination that are no longer present in the incoming stream data.

Note

GetExistingData is only executed when the stream starts up for the first time. Once the initial synchronization is complete and the first checkpoint has been committed, it will not be called again (even on subsequent stream restarts). From that point onward, the sink relies entirely on Flowtide's internally persisted state to calculate subsequent changes.

Custom sink sample

A runnable console application demonstrating a custom sink and source is available under samples/CustomSinkSample. It shows how initial sink synchronization with GetExistingData works and how the sink handles data updates.

Adding custom converters

It is possible to register custom converters to convert .NET objects into the column format, and back. This is done by overriding the GetCustomConverters method in both generic sources and sinks.

One example is for instance to use strings for enums instead of integers. Example:

internal class MyDataSource : GenericDataSource<MyClass>
{
    ...

    public override IEnumerable<IObjectColumnResolver> GetCustomConverters()
    {
        // Returns a new converter resolver for enums that will use strings instead of integers
        yield return new EnumResolver(enumAsStrings: true);
    }

    ...
}

Table of Contents

Custom Data Source

Generic Data Source

Implementation example

Push-based delta loading

Trigger data reloads programatically

SQL Table Provider

Generic data sink

Implementation example

Fetching existing data

Note

Custom sink sample

Adding custom converters