c8-source-postgres

Singer tap that extracts data from a PostgreSQL database and produces JSON-formatted data following the Singer spec.

How to use it

The recommended method of running this tap is to use it with Macrometa data connectors. If you want to run this Singer Tap independently please read further.

Install and Run

First, make sure Python 3 is installed on your system or follow these installation instructions for Mac or Ubuntu .

It's recommended to use a virtualenv:

  python3 -m venv venv
  pip install c8-source-postgres

or

  make venv

Create a config.json

{
  "host": "localhost",
  "port": 5432,
  "user": "postgres",
  "password": "secret",
  "dbname": "db"
}

These are the same basic configuration properties used by the PostgreSQL command-line client (psql).

Full list of options in config.json:

Run the tap in Discovery Mode

c8-source-postgres --config config.json --discover                # Should dump a Catalog to stdout
c8-source-postgres --config config.json --discover > catalog.json # Capture the Catalog

Add Metadata to the Catalog

Each entry under the Catalog's "stream" key will need the following metadata:

{
  "streams": [
    {
      "stream_name": "my_topic"
      "metadata": [{
        "breadcrumb": [],
        "metadata": {
          "selected": true,
          "replication-method": "LOG_BASED",
        }
      }]
    }
  ]
}

The replication method can be one of FULL_TABLE, INCREMENTAL or LOG_BASED.

Note: Log based replication requires a few adjustments in the source postgres database, please read further for more information.

Run the tap in Sync Mode

c8-source-postgres --config config.json --catalog catalog.json

The tap will write bookmarks to stdout which can be captured and passed as an optional --state state.json parameter to the tap for the next sync.

Log Based replication requirements

• PostgreSQL database's running PostgreSQL versions 9.4.x or greater. To avoid a critical PostgreSQL bug, use at least one of the following minor versions:

  • PostgreSQL 12.0
  • PostgreSQL 11.2
  • PostgreSQL 10.7
  • PostgreSQL 9.6.12
  • PostgreSQL 9.5.16
  • PostgreSQL 9.4.21

• A connection to the master instance. Log-based replication will only work by connecting to the master instance.

• wal2json plugin: To use Log Based for your PostgreSQL integration, you must install the wal2json plugin. The wal2json plugin outputs JSON objects for logical decoding, which the tap then uses to perform Log-based Replication. Steps for installing the plugin vary depending on your operating system. Instructions for each operating system type are in the wal2json’s GitHub repository:

  • Unix-based operating systems
  • Windows

• postgres config file: Locate the database configuration file (usually postgresql.conf) and define the parameters as follows:

  wal_level=logical
  max_replication_slots=5
  max_wal_senders=5
  

  Restart your PostgreSQL service to ensure the changes take effect.

  Note: For max_replication_slots and max_wal_senders, we’re defaulting to a value of 5. This should be sufficient unless you have a large number of read replicas connected to the master instance.

• Existing replication slot: Log based replication requires a dedicated logical replication slot. In PostgreSQL, a logical replication slot represents a stream of database changes that can then be replayed to a client in the order they were made on the original server. Each slot streams a sequence of changes from a single database.

  Login to the master instance as a superuser and using the wal2json plugin, create a logical replication slot:

    SELECT *
    FROM pg_create_logical_replication_slot('macrometa_<database_name>', 'wal2json');
  

  Note: Replication slots are specific to a given database in a cluster. If you want to connect multiple databases - whether in one integration or several - you’ll need to create a replication slot for each database.

To run tests:

1. Install python test dependencies in a virtual env:

 make venv

2. You need to have a postgres database to run the tests and export its credentials:

  export C8_SOURCE_POSTGRES_HOST=<postgres-host>
  export C8_SOURCE_POSTGRES_PORT=<postgres-port>
  export C8_SOURCE_POSTGRES_SECONDARY_HOST=<postgres-replica-host>
  export C8_SOURCE_POSTGRES_SECONDARY_PORT=<postgres-replica-port>
  export C8_SOURCE_POSTGRES_USER=<postgres-user>
  export C8_SOURCE_POSTGRES_PASSWORD=<postgres-password>

You can make use of the local docker-compose to spin up a test database by running make start_db

Test objects will be created in the postgres database.

3. To run the tests:

  make test

To run pylint:

1. Install python dependencies and run python linter

  make venv
  make pylint