Fully reproducible step-by-step demo on how to stream tables from Postgres to Kafka/KSQL back to Postgres.
All components are containerized so that the only things you need to run through this demo are Docker and docker-compose.
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.
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.
docker build -t debezium-connect -f debezium.Dockerfile .
docker-compose up -d
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
At the command line:
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
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 -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:
docker exec -it <kafka-container-id> /bin/bash
and listing the available topics:
/usr/bin/kafka-topics --list --zookeeper zookeeper:2181
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
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';
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.
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;
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
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%.