PostgreSQL To Azure Event Hubs

​

Scenario

​

Prerequisites

1. Enable logical decoding in PostgreSQL. Ensure that the following settings/GUCs are properly configured:
   1. wal_level: logical
   2. max_wal_senders: >1
   3. max_replication_slots: 4
2. Enable replication access for a PostgreSQL user - ALTER USER pg_user REPLICATION;
3. Ensure that both tables have primary keys. Composite primaries are also fine. If not, make sure your tables have REPLICA IDENTITY FULL.
4. If you are using PostgreSQL on the cloud, below links capture how to enable logical replication for each cloud:
   1. AWS RDS and Aurora PostgreSQL
   2. Azure Database for PostgreSQL - Flexible Server
   3. GCP Cloud SQL PostgreSQL

​

Step 1: Add PostgreSQL and Event Hubs Peers

-- Connect to PeerDB
psql "port=9900 host=localhost password=peerdb"

-- Add PostgreSQL and Event Hubs peers
CREATE PEER postgres_peer FROM POSTGRES (...);
CREATE PEER eventhubs_peer FROM EVENTHUBS (...);

​

Step 2: Real-Time CDC from PostgreSQL to Event Hubs

CREATE MIRROR IF NOT EXISTS <mirror-name>
FROM <postgres-peer-name> TO <eventhubs-peer-name>
WITH TABLE MAPPING(
    <source-schema>.<table>:<namespace_name>.<eventhub_name>.<partition_key_column>,
    ... -- Repeat as required for multiple tables
)
WITH(
    max_batch_size = <number>,
    publication_name = '<publication-name>'
);

CREATE MIRROR IF NOT EXISTS test_eh_mirror
FROM test_pg_peer TO test_eh_peer
WITH TABLE MAPPING(
    schema1.table1:mynamespace1.hub1.id,
    schema1.table2:mynamespace2.hub2.id
    -- Add more tables as required
)
WITH(
    do_initial_copy = false,
    max_batch_size = 300000,
    publication_name = 'test_publication'
);

• mirror-name: Desired name for the mirror.
• postgres-peer-name: Name of the PostgreSQL peer.
• eventhubs-peer-name: Name of the Event Hubs group peer.
• namespace-name: Name of the namespace in which you wish to sync to an eventhub.
• eventhub-name: Name of the eventhub in which you wish to sync the data. PeerDB creates the eventhub for you if it doesn’t exist already.
• partition_key_column: Column in the source table whose values will be used to route data into the partitions of the event hub.
• max_batch_size: Maximum number of records in a batch.
• publication_name: Name of the publication.

​

Step 3: Validate the Mirror

SELECT COUNT(*) FROM postgres_peer.public.transactions;

​

Step 4: Monitor the MIRROR

​

Step 5: DROP MIRROR

-- drop the mirror
DROP MIRROR real_time_cdc;