postgres-kafka-demo

Fully reproducible step-by-step demo on how to stream tables from Postgres to Kafka/KSQL back to Postgres.

I walk through this tutorial and others here on GitHub and on my Medium blog. Here is a friend link for open access to the article: Data Stream Processing for Newbies with Kafka, KSQL, and Postgres. I'll always add friend links on my GitHub tutorials for free Medium access if you don't have a paid Medium membership (referral link).

If you find any of this useful, I always appreciate contributions to my Saturday morning fancy coffee fund!

All components are containerized so that the only things you need to run through this demo are Docker and docker-compose.

Data

The data used here was originally taken from the Graduate Admissions open dataset available on Kaggle. The admit csv files are records of students and test scores with their chances of college admission. The research csv files contain a flag per student for whether or not they have research experience.

Components

The following technologies are used through Docker containers:

Kafka, the streaming platform
Zookeeper, Kafka's best friend
KSQL server, which we will use to create real-time updating tables
Kafka's schema registry, needed to use the Avro data format
Kafka Connect, pulled from debezium, which will source and sink data back and forth through Kafka
Postgres, pulled from debezium, tailored for use with Connect

Most of the containers are pulled directly from official Docker Hub images. The debezium connect image used here needs some additional packages, so I've built a debezium connect image that I've made available on DockerHub. It can also be built from the included Dockerfile.

Build the connect image (optional)

docker build -t debezium-connect -f debezium.Dockerfile .

Bring up the entire environment

docker-compose up -d

check postgres wal_level

select * from pg_settings where name ='wal_level';
or via psql
postgres@localhost:postgres> SHOW wal_level

Loading data into Postgres

We will bring up a container with a psql command line, mount our local data files inside, create a database called students, and load the data on students' chance of admission into the admission table.

docker run -it --rm --network=postgres-kafka-demo_default \
         -v $PWD:/home/data/ \
         postgres:11.0 psql -h postgres -U postgres

Password = postgres

Open the powershell available with the postgres container and at the command line: psql -h localhost -U postgres -p 5432

or docker exec -it b87b96279fd6 psql -h postgres -U postgres

CREATE DATABASE students;
\connect students;

Load our admission data table:

CREATE TABLE admission
(student_id INTEGER, gre INTEGER, toefl INTEGER, cpga DOUBLE PRECISION, admit_chance DOUBLE PRECISION,
CONSTRAINT student_id_pk PRIMARY KEY (student_id));

\copy admission FROM '/home/data/admit_1.csv' DELIMITER ',' CSV HEADER

Load the research data table with:

CREATE TABLE research
(student_id INTEGER, rating INTEGER, research INTEGER,
PRIMARY KEY (student_id));

\copy research FROM '/home/data/research_1.csv' DELIMITER ',' CSV HEADER

Connect Postgres database as a source to Kafka

load the json file to kafka connect (port 8083)

The postgres-source.json file contains the configuration settings needed to sink all of the students database to Kafka.

curl -X POST -H "Accept:application/json" -H "Content-Type: application/json" --data @postgres-source.json http://localhost:8083/connectors

The connector 'postgres-source' should show up when curling for the list of existing connectors: curl --location --request GET 'http://localhost:8083/connectors'
--header 'Accept: application/json'

TO DELETE THE CONNECTOR --curl -X DELETE http://localhost:8083/connectors/postgres-source

curl -H "Accept:application/json" localhost:8083/connectors/

The two tables in the students database will now show up as topics in Kafka. You can check this by entering the Kafka container:

open debezium connect container

docker exec -it <kafka-container-id> /bin/bash
docker exec -it fb350d45a429 /bin/bash  //get container id from docker ps

and listing the available topics:

/usr/bin/kafka-topics --list --zookeeper zookeeper:2181

Create tables in KSQL

Bring up a KSQL server command line client as a container:

docker run --network postgres-kafka-demo_default \
           --interactive --tty --rm \
           confluentinc/cp-ksql-cli:latest \
           http://ksql-server:8088

docker-compose -f docker-compose-ksql.yml exec ksql-cli ksql http://ksql-server:8088 docker exec -it 545fc5ac48b7 ksql-cli ksql http://ksql-server:8088

To see your updates, a few settings need to be configured by first running:

set 'commit.interval.ms'='2000'; set 'cache.max.bytes.buffering'='10000000'; set 'auto.offset.reset'='earliest';


### Mirror Postgres tables

The Postgres table topics will be visible in KSQL, and we will create
KSQL streams to auto update KSQL tables mirroring the Postgres tables:

SHOW TOPICS;

CREATE STREAM admission_src (student_id INTEGER, gre INTEGER, toefl INTEGER, cpga DOUBLE, admit_chance DOUBLE)WITH (KAFKA_TOPIC='dbserver1.public.admission', VALUE_FORMAT='AVRO');

CREATE STREAM admission_src_rekey WITH (PARTITIONS=1) AS
SELECT * FROM admission_src PARTITION BY student_id;

SHOW STREAMS;

CREATE TABLE admission (student_id INTEGER, gre INTEGER, toefl INTEGER, cpga DOUBLE, admit_chance DOUBLE)
WITH (KAFKA_TOPIC='ADMISSION_SRC_REKEY', VALUE_FORMAT='AVRO', KEY='student_id');

SHOW TABLES;

CREATE STREAM research_src (student_id INTEGER, rating INTEGER, research INTEGER)
WITH (KAFKA_TOPIC='dbserver1.public.research', VALUE_FORMAT='AVRO');

CREATE STREAM research_src_rekey WITH (PARTITIONS=1) AS
SELECT * FROM research_src PARTITION BY student_id;

CREATE TABLE research (student_id INTEGER, rating INTEGER, research INTEGER)
WITH (KAFKA_TOPIC='RESEARCH_SRC_REKEY', VALUE_FORMAT='AVRO', KEY='student_id');


Currently KSQL uses uppercase casing convention for stream, table, and field
names.

### Create downstream tables

We will create a new KSQL streaming table to join students' chance of
admission with research experience.

CREATE TABLE research_boost AS
SELECT a.student_id as student_id,
a.admit_chance as admit_chance,
r.research as research
FROM admission a
LEFT JOIN research r on a.student_id = r.student_id;


and another table calculating the average chance of admission for
students with and without research experience:

CREATE TABLE research_ave_boost AS
SELECT research, SUM(admit_chance)/COUNT(admit_chance) as ave_chance
FROM research_boost
WITH (KAFKA_TOPIC='research_ave_boost', VALUE_FORMAT='delimited', KEY='research')
GROUP BY research;


## Add a connector to sink a KSQL table back to Postgres

The postgres-sink.json configuration file will create a RESEARCH_AVE_BOOST
table and send the data back to Postgres.

curl -X POST -H "Accept:application/json" -H "Content-Type: application/json" --data @postgres-sink.json http://localhost:8083/connectors


## Update the source Postgres tables and watch the Postgres sink table update

The RESEARCH_AVE_BOOST table should now be available in Postgres to query:

SELECT "AVE_CHANCE" FROM "RESEARCH_AVE_BOOST" WHERE cast("RESEARCH" as INT)=0;


With these data the average admission chance will be 65.19%.

Note that the tables are forced to upper case and case sensitive.
The research field needs to be cast because it has been typed as text
instead of integer, which may be a bug in KSQL or Connect.

Add some new data to the admission and research tables in Postgres:

\copy admission FROM '/home/data/admit_2.csv' DELIMITER ',' CSV HEADER \copy research FROM '/home/data/research_2.csv' DELIMITER ',' CSV HEADER


With the same query above on the RESEARCH_AVE_BOOST table, the
average chance of admission for students without research experience
has been updated to 63.49%.

paolo-tn/postgres-kafka-demo