Amazon DynamoDB

PgOutput2Json supports copying modified PostgreSQL rows to Amazon DynamoDB tables. By default, rows are copied only when they change, using logical replication and compact JSON messages. Optionally, initial data copy can be enabled with WithInitialDataCopy(true) when configuring the builder.

The PgOutput2Json library will create DynamoDB tables for each table included in logical replication.

Tables are created the first time a row belonging to the respective table is changed. The first PK column is used as partition key. The other PK columns are joined with | into a string, stored in the sort key attribute SK. An array column is stored as a string with its JSON text.

Important: First set up PostgreSQL as described in Getting started.

Create a .NET Worker Service

Set up a new .NET Worker Service and add the following package reference:

dotnet add package PgOutput2Json.DynamoDb

In your Worker.cs, use the following code to configure change propagation to Amazon DynamoDB:

using Amazon;
using Amazon.DynamoDBv2;
using Amazon.Runtime;
using Amazon.Runtime.CredentialManagement;
using PgOutput2Json;

public class Worker : BackgroundService
{
    private readonly ILoggerFactory _loggerFactory;

    public Worker(ILoggerFactory loggerFactory)
    {
        _loggerFactory = loggerFactory;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // This code assumes PostgreSQL is running on localhost
        using var pgOutput2Json = PgOutput2JsonBuilder.Create()
            .WithLoggerFactory(_loggerFactory)
            .WithPgConnectionString("server=localhost;database=my_database;username=pgoutput2json;password=_your_password_here_")
            .WithPgPublications("my_publication")
            .UseDynamoDb(options =>
            {
                options.ClientConfig = new AmazonDynamoDBConfig
                {
                    ServiceURL = "http://localhost:8000",                              // for local DynamoDB, change/remove for AWS
                    UseHttp = true,                                                    // for local DynamoDb, remove for AWS
                    DefaultAWSCredentials = new BasicAWSCredentials("dummy", "dummy"), // for local DynamoDB, remove for AWS
                    MaxErrorRetry = 3,                                                 // retry failed requests up to 3 times
                };
            })
            .Build();

        await pgOutput2Json.StartAsync(stoppingToken);
    }
}