Skip to content

A tutorial for help setup a sample ETL Postgres CDC -> Kafka Connect -> ES

License

Notifications You must be signed in to change notification settings

uuhnaut69/postgres-cdc-kafka-connect-es-setup-guide

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

4 Commits
 
 
 
 

Repository files navigation

postgres-cdc-kafka-connect-es-setup-guide

A tutorial setup ETL Postgres CDC -> Kafka Connect -> ES

Topology


                   +-------------+
                   |             |
                   |   Postgres  |
                   |             |
                   +------+------+
                          |
                          |
                          |
          +---------------v------------------+
          |                                  |
          |           Kafka Connect          |
          |    (Debezium, ES connectors)     |
          |                                  |
          +---------------+------------------+
                          |
                          |
                          |
                          |
                  +-------v--------+
                  |                |
                  | Elasticsearch  |
                  |                |
                  +----------------+

Installation Guide

Installing the wal2json plugin

wal2json: To encode changes in JSON format

  • Go to lib folder of Postgres app: /usr/pgsql-9.6/lib/ (The url maybe difference depend on your installation by homebrew or using Postgres App)

  • Set the export path as shown below: export PATH="$PATH:/usr/pgsql-9.6/bin"

  • Enter the wal2json installation commands.

    git clone https://github.com/eulerto/wal2json -b master --single-branch \
    && cd wal2json \
    && git checkout d2b7fef021c46e0d429f2c1768de361069e58696 \
    && make && make install \
    && cd .. \
    && rm -rf wal2json
    

Or Using Docker Images: Here

Enable Replication on the PostgreSQL server

  • Open postgresql.conf file. Find and edit as below configuration.
    # LOGGING
    log_min_error_statement = fatal
    # CONNECTION
    listen_addresses = '*'
    # MODULES
    shared_preload_libraries = 'wal2json'
    # REPLICATION
    wal_level = logical             # minimal, archive, hot_standby, or logical (change requires restart)
    max_wal_senders = 1             # max number of walsender processes (change requires restart)
    #wal_keep_segments = 4          # in logfile segments, 16MB each; 0 disables
    #wal_sender_timeout = 60s       # in milliseconds; 0 disables
    max_replication_slots = 1       # max number of replication slots (change requires restart)
    

Install Kafka Connect (Confluent Platform)

  • Installation Confluent tutorial can be found here

Install the connector

  • Debezium Connector Postgresql:
confluent-hub install debezium/debezium-connector-postgresql:latest 
  • Elasticsearch Sink Connector:
confluent-hub install confluentinc/kafka-connect-elasticsearch:latest 

( Must use ES version >= 4.0 unless when delete operation on postgres, record being deleted won't be deleted on ES)

Connect & Run

  • Register Postgres CDC connector
[POST] http://localhost:8083/connectors/

Body request:

{
    "name": "jpwork-connector",
    "config": {
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
        "tasks.max": "1",
        "plugin.name": "pgoutput",
        "database.hostname": "127.0.0.1",
        "database.port": "5432",
        "database.user": "postgres",
        "database.password": "postgres",
        "database.dbname" : "postgres",
        "database.server.name": "postgres",
        "schema.whitelist": "public"
    }
}
[POST] http://localhost:8083/connectors/

Body request:

{
    "name": "elastic-sink",
    "config": {
        "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
        "tasks.max": "1",
        "topics": "postgres.public.category",
        "connection.url": "http://localhost:9200",
        "transforms": "unwrap,key",
        "transforms.unwrap.type": "io.debezium.transforms.UnwrapFromEnvelope",
        "transforms.unwrap.drop.tombstones": "false",
        "transforms.unwrap.drop.deletes": "false",
        "transforms.key.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
        "transforms.key.field": "id",
        "key.ignore": "false",
        "type.name": "category",
        "behavior.on.null.values": "delete"
    }
}

About

A tutorial for help setup a sample ETL Postgres CDC -> Kafka Connect -> ES

Topics

Resources

License

Stars

Watchers

Forks

Releases

No releases published

Packages

No packages published