DEV Community: sisiodos

Connecting RDBs and Search Engines — Chapter 5

sisiodos — Sat, 10 May 2025 22:38:23 +0000

Chapter 5: Implementing a CDC Join Pipeline with Flink SQL

In this chapter, we build a full pipeline that captures CDC streams from PostgreSQL via Debezium, joins the data using Flink SQL, and stores the results in OpenSearch.

Architecture Overview

PostgreSQL
   ↓ (CDC)
Debezium (Kafka Connect)
   ↓ (debezium-json)
Kafka Topic (orders, products)
   ↓
Flink SQL
   ↓
OpenSearch (orders_with_products)

PostgreSQL Table Setup

Add the following to postgres/01-init.sql:

-- Products table
CREATE TABLE products (
  product_id VARCHAR(32) PRIMARY KEY,
  product_name VARCHAR(255),
  category_id VARCHAR(32)
);
ALTER TABLE products REPLICA IDENTITY FULL;

-- Initial product data
INSERT INTO products (product_id, product_name, category_id) VALUES
('P001', 'Sneaker X', 'C001'),
('P002', 'Jacket Y', 'C002');

-- Orders table
CREATE TABLE orders (
  order_id VARCHAR(32) PRIMARY KEY,
  order_time TIMESTAMP,
  customer_id VARCHAR(32),
  product_id VARCHAR(32),
  quantity INT,
  price NUMERIC(10,2)
);
ALTER TABLE orders REPLICA IDENTITY FULL;

-- Initial order data
INSERT INTO orders (order_id, order_time, customer_id, product_id, quantity, price) VALUES
('O1001', '2025-04-27 10:00:00', 'CUST01', 'P001', 1, 9800.00),
('O1002', '2025-04-27 10:05:00', 'CUST02', 'P002', 2, 15800.00);

-- Grant permissions to Debezium user
GRANT SELECT ON products, orders TO debezium;

-- Add tables to existing publication
ALTER PUBLICATION debezium_pub ADD TABLE products, orders;

Register Debezium Connector

After Kafka Connect is running, register the connector using a script such as scripts/02-debezium-table-table-join.sh. Ensure the config includes:

"table.include.list": "public.testtable,public.products,public.orders"

Create OpenSearch Index

Prepare opensearch/orders_with_products-mapping.json:

{
  "mappings": {
    "properties": {
      "order_id": { "type": "keyword" },
      "product_id": { "type": "keyword" },
      "product_name": {
        "type": "text",
        "fields": { "raw": { "type": "keyword" } }
      },
      "category_id": { "type": "keyword" },
      "quantity": { "type": "integer" },
      "price": { "type": "double" }
    }
  }
}

Create the index:

curl -X PUT "http://localhost:9200/orders_with_products" \
  -H "Content-Type: application/json" \
  -d @opensearch/orders_with_products-mapping.json

✅ product_name uses a multi-field (text + keyword) for full-text and exact match support.

Example Flink SQL

Create flink/sql/table-table-join.sql:

-- Products table
CREATE TABLE products (
  product_id STRING,
  product_name STRING,
  category_id STRING,
  PRIMARY KEY (product_id) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'topic' = 'dbserver1.public.products',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format' = 'debezium-json',
  'scan.startup.mode' = 'earliest-offset'
);

-- Orders table
CREATE TABLE orders (
  order_id STRING,
  order_time STRING,
  customer_id STRING,
  product_id STRING,
  quantity INT,
  price DOUBLE
) WITH (
  'connector' = 'kafka',
  'topic' = 'dbserver1.public.orders',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format' = 'debezium-json',
  'scan.startup.mode' = 'earliest-offset'
);

-- OpenSearch Sink
CREATE TABLE orders_with_products (
  order_id STRING,
  product_id STRING,
  product_name STRING,
  category_id STRING,
  quantity INT,
  price DOUBLE,
  PRIMARY KEY (order_id, product_id) NOT ENFORCED
) WITH (
  'connector' = 'opensearch-2',
  'hosts' = 'http://opensearch:9200',
  'allow-insecure' = 'true',
  'index' = 'orders_with_products',
  'document-id.key-delimiter' = '$',
  'sink.bulk-flush.max-size' = '1mb'
);

-- Join View
CREATE VIEW orders_with_products_view AS
SELECT
  o.order_id,
  o.product_id,
  p.product_name,
  p.category_id,
  o.quantity,
  o.price
FROM orders o
INNER JOIN products p ON o.product_id = p.product_id;

-- Insert into OpenSearch
INSERT INTO orders_with_products
SELECT * FROM orders_with_products_view;

Run the Flink Job

Execute from inside the Flink container:

sql-client.sh -f /opt/flink/sql/table-table-join.sql

Or via Docker:

docker compose exec flink-jobmanager sql-client.sh -f /opt/flink/sql/table-table-join.sql

Validation Steps

Verify CDC data in Kafka topics:

docker compose exec kafka kafka-console-consumer --bootstrap-server localhost:9092 --topic dbserver1.public.orders --from-beginning
docker compose exec kafka kafka-console-consumer --bootstrap-server localhost:9092 --topic dbserver1.public.products --from-beginning

Verify data in OpenSearch:

curl -X GET "http://localhost:9200/orders_with_products/_search?pretty"

With custom query:

curl -s -X GET "http://localhost:9200/orders_with_products/_search?pretty" \
  -H 'Content-Type: application/json' \
  -d '{ "query": { "match_all": {} } }'

For convenience, use a script like scripts/opensearch-query.sh:

#!/bin/bash

osq() {
  local index="${1:-_all}"
  local size="${2:-10}"

  curl -s -X GET "http://localhost:9200/${index}/_search?pretty" \
    -H 'Content-Type: application/json' \
    -d "{\"query\": { \"match_all\": {} }, \"size\": ${size} }"
}

Example usage:

source scripts/opensearch-query.sh
osq orders_with_products
osq orders_with_products 20
osq orders_with_products 100 | jq '.hits.hits[]._id'

In Chapter 6 and beyond, we will explore topics such as deduplication, batch processing, DLQ (Dead Letter Queue), and OpenSearch index partitioning strategies for production-grade pipelines.

Connecting RDBs and Search Engines — Chapter 4 Part 2

sisiodos — Sat, 10 May 2025 22:38:11 +0000

Chapter 4 (Part 2): Integrating Kafka CDC Data with OpenSearch Using Flink

In this chapter, we use Flink SQL to process CDC data in Debezium JSON format from Kafka and write it to OpenSearch.

1. Architecture Overview

graph TD
  subgraph Source
    PG[PostgreSQL]
  end
  subgraph CDC
    DBZ[Debezium Connect]
  end
  subgraph Stream
    TOPIC[Kafka Topic]
  end
  subgraph Flink
    FLINK[Flink SQL Kafka Source]
    OS_SINK[Flink OpenSearch Sink]
  end
  subgraph Search
    OS[OpenSearch]
  end

  PG --> DBZ --> TOPIC --> FLINK --> OS_SINK --> OS

Flink consumes CDC events from Kafka, transforms them, and upserts the data into OpenSearch.

2. Prerequisites

Ensure the following OSS components are running (e.g., via Docker Compose):

PostgreSQL
Apache Kafka
Kafka Connect (Debezium)
ZooKeeper
Flink (1.19)
OpenSearch (2.13)

💡 See docker-compose.yaml for details.

3. Flink JAR Setup

Place the following JARs in flink/plugins/kafka/, flink/plugins/opensearch directories.

Required Connector JARs

flink-sql-connector-kafka-3.3.0-1.19.jar
flink-sql-connector-opensearch2-2.0.0-1.19.jar

💡 Note: Why use the plugins directory?

Flink's SQL Client loads connectors via a plugin classloader.
Placing JARs directly into flink/lib can lead to class conflicts when used alongside the DataStream API.
Using the plugins/{connector} directory to place connector JARs is a safer approach that prevents
classloading issues.

4. Create OpenSearch Index

Create opensearch/test-index-mapping.json:

{
  "mappings": {
    "properties": {
      "doc_id": { "type": "keyword" },
      "id": { "type": "integer" },
      "message": { "type": "text" }
    }
  }
}

Run:

curl -X PUT "http://localhost:9200/test-index" \
  -H "Content-Type: application/json" \
  -d @opensearch/test-index-mapping.json

✅ Defining mappings helps avoid automatic keyword field generation.

5. Create Flink SQL Script

Save the following as flink/sql/cdc_to_opensearch.sql:

CREATE TABLE cdc_source (
  id INT,
  message STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'topic' = 'dbserver1.public.testtable',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format' = 'debezium-json',
  'scan.startup.mode' = 'earliest-offset'
);

CREATE TABLE opensearch_sink (
  doc_id STRING,
  id INT,
  message STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'opensearch-2',
  'hosts' = 'http://opensearch:9200',
  'allow-insecure' = 'true',
  'index' = 'test-index',
  'document-id.key-delimiter' = '$',
  'sink.bulk-flush.max-size' = '1mb'
);

INSERT INTO opensearch_sink
SELECT
  SHA256(CAST(id AS STRING)) AS doc_id,
  id,
  message
FROM cdc_source;

6. Run the Flink Job

docker compose exec flink-jobmanager bash
sql-client.sh -f /opt/flink/sql/cdc_to_opensearch.sql -l /opt/flink/plugins/kafka -l /opt/flink/plugins/

7. Verify OpenSearch Output

curl -X GET "http://localhost:9200/test-index/_search?pretty"

Sample output:

"hits" : [
  {
    "_index" : "test-index",
    "_id" : "c4ca4238a0b923820dcc509a6f75849b",
    "_source" : {
      "doc_id" : "c4ca4238a0b923820dcc509a6f75849b",
      "id" : 1,
      "message" : "CDC test row"
    }
  }
]

In this chapter, we built and verified a pipeline using Flink SQL to upsert CDC data into OpenSearch.

(Coming soon: Chapter 5 — Implementing a CDC Join Pipeline with Flink SQL)

Connecting RDBs and Search Engines — Chapter 4 Part 1

sisiodos — Sat, 10 May 2025 22:37:58 +0000

Chapter 4 (Part 1): Outputting Kafka CDC Data to Console with Flink

In this chapter, we will process the CDC data delivered to Kafka using Flink SQL and print the output to the console. Before persisting to OpenSearch, we visually verify that Flink is correctly consuming and processing the data from Kafka.

1. Prerequisites

Ensure the following components are already running:

PostgreSQL
Apache Kafka
Kafka Connect
ZooKeeper
Flink

Refer to Chapter 3 for details on setting up the Debezium → Kafka pipeline.

2. Architecture Overview

graph TD
  subgraph source
    PG[PostgreSQL]
  end
  subgraph Change Data Capture
    DBZ[Debezium Connect]
  end
  subgraph stream platform
    TOPIC[Kafka Topic]
  end
  subgraph stream processing
    FLINK[Flink SQL Kafka Source]
    PRINT[Flink Print Sink Console]
  end

  PG --> DBZ --> TOPIC --> FLINK --> PRINT

We verify that CDC events flow from Kafka to Flink and appear in the standard output in a format like +I[...].

3. Add Kafka Connector to Flink

Add the Kafka SQL connector JAR to Flink:

flink-sql-connector-kafka-3.3.0-1.19.jar

⚠️ To avoid interference with Flink core libraries, place the JAR in flink/plugins/{connector}.

`docker-compose.yaml` Excerpt

flink-jobmanager:
  image: flink:1.19
  command: ["/bin/bash", "/jobmanager-entrypoint.sh"]
  volumes:
    - ./flink/jobmanager-entrypoint.sh:/jobmanager-entrypoint.sh
    - ./flink/sql:/opt/flink/sql
    - ./flink/plugins/kafka:/opt/flink/plugins/kafka

💡 Note: Why use the plugins directory?

Flink's SQL Client loads connectors via a plugin classloader.
Placing JARs directly into flink/lib can lead to class conflicts when used alongside the DataStream API.
Using the plugins/{connector} directory to place connector JARs is a safer approach that prevents
classloading issues.

4. Write Flink SQL Script

Save the following SQL in flink/sql/cdc_to_console.sql:

CREATE TABLE cdc_source (
  id INT,
  message STRING,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'kafka',
  'topic' = 'dbserver1.public.testtable',
  'properties.bootstrap.servers' = 'kafka:9092',
  'format' = 'debezium-json',
  'scan.startup.mode' = 'earliest-offset'
);

CREATE TABLE print_sink (
  id INT,
  message STRING
) WITH (
  'connector' = 'print'
);

INSERT INTO print_sink
SELECT * FROM cdc_source;

Explanation of Flink SQL Tables

Kafka Source Table: `cdc_source`

Property	Description
`connector = 'kafka'`	Reads data from a Kafka topic
`format = 'debezium-json'`	Handles JSON messages in Debezium format
`scan.startup.mode = 'earliest-offset'`	Reads from the earliest offset
`PRIMARY KEY (...) NOT ENFORCED`	Defines a primary key without enforcement

Console Sink Table: `print_sink`

Uses Flink's internal print connector to write output to stdout.

5. Run the Flink SQL Job

docker compose exec flink-jobmanager bash
sql-client.sh -f /opt/flink/sql/cdc_to_console.sql -l /opt/flink/plugins/kafka

Verify Running Jobs

docker compose exec flink-jobmanager bash
flink list

Expected output:

------------------ Running/Restarting Jobs -------------------
<job-id> : insert-into_default_catalog.default_database.print_sink (RUNNING)

6. Check the Output

docker compose logs flink-taskmanager

Expected output:

flink-taskmanager-1 | +I[1, CDC test row]

Why TaskManager?

The print sink outputs to the logs of the TaskManager running the job.
Use docker compose logs flink-taskmanager to view the output.

7. Troubleshooting

SQL Client Doesn’t Start

Check if JobManager is running:

docker compose logs flink-jobmanager

No CDC Output Appears

Is the Kafka topic name correct?
Is there data in the Kafka topic?
Is scan.startup.mode set to earliest-offset?

Check topic content:

kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic dbserver1.public.testtable \
  --from-beginning

Bonus: Observing Parallelism

If you scale Flink to use multiple TaskManagers, you’ll see output distributed across their logs.

This allows you to observe parallel execution, slot allocation, and subtask distribution.

In this chapter, we confirmed the flow from Kafka → Flink → console output. Next, we will write the results to OpenSearch for persistence.

(Coming soon: Chapter 4 Part 2 — Integrating Kafka CDC Data with OpenSearch Using Flink)

Connecting RDBs and Search Engines — Chapter 3

sisiodos — Sat, 10 May 2025 22:37:45 +0000

Chapter 3: Verifying the Flow from PostgreSQL to Debezium to Kafka

In this chapter, we will test the end-to-end process of capturing change data from PostgreSQL with Debezium and delivering it to Kafka.

1. Starting the Environment

Before beginning this chapter, ensure that all necessary services are running via Docker Compose:

docker compose up -d

Verify that the following components are running:

PostgreSQL
Apache Kafka
Debezium Connect (Kafka Connect)
ZooKeeper

See docker-compose.yaml in the repository for service configuration details.

2. PostgreSQL Setup

First, we create the table that Debezium will monitor. Add the following to postgres/00-init.sql:

-- Create Debezium user
CREATE ROLE debezium WITH LOGIN PASSWORD 'dbz' REPLICATION;

-- Create target table
CREATE TABLE testtable (
  id INTEGER PRIMARY KEY,
  message VARCHAR(255)
);
INSERT INTO testtable (id, message) VALUES (1, 'CDC test row');

-- Grant SELECT permission
GRANT SELECT ON testtable TO debezium;

-- Create publication (for pgoutput plugin)
CREATE PUBLICATION debezium_pub FOR TABLE testtable;

This script runs once only during initial container creation.
To apply changes to the script, you must delete persistent volumes and restart:

docker compose down --volumes

3. PostgreSQL Container Configuration

To enable CDC using WAL (Write-Ahead Log), add the following to the docker-compose.yaml configuration:

command: >
  postgres
  -c wal_level=logical
  -c max_replication_slots=4
  -c max_wal_senders=4

To restart PostgreSQL without losing data:

docker compose restart postgres

To rerun initialization SQL, you’ll need to remove volumes:

docker compose down --volumes

4. Registering the Debezium Connector

Once Debezium Connect is running, register the connector with the following command (Debezium 1.9 format):

curl -X POST "localhost:8083/connectors" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "postgres-connector",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "database.hostname": "postgres",
      "database.port": "5432",
      "database.user": "debezium",
      "database.password": "dbz",
      "database.dbname": "postgres",
      "database.server.name": "dbserver1",
      "topic.prefix": "dbserver1",
      "plugin.name": "pgoutput",
      "publication.name": "debezium_pub",
      "slot.name": "debezium_slot",
      "slot.drop.on.stop": "true",
      "table.include.list": "public.testtable",
      "snapshot.mode": "initial",
      "tombstones.on.delete": "false",
      "key.converter": "org.apache.kafka.connect.json.JsonConverter",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "key.converter.schemas.enable": "false",
      "value.converter.schemas.enable": "false"
    }
  }'

About `snapshot.mode`

The snapshot.mode option controls whether to load the full contents of the database at connector startup.

initial: Load all existing data once, then capture changes (default)
never: Skip initial load; capture only subsequent changes

In this guide, we use initial, so existing rows are sent to Kafka with op: r (read).

5. Checking Connector Status

Check the connector status:

curl http://localhost:8083/connectors/postgres-connector/status

Expected output:

{
  "name": "postgres-connector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "10.4.1.81:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "10.4.1.81:8083"
    }
  ],
  "type": "source"
}

6. Checking Kafka Topics

List Kafka topics:

# Enter Kafka container
docker compose exec kafka bash

# List topics
kafka-topics --bootstrap-server localhost:9092 --list

Expected output:

dbserver1.public.testtable

7. Viewing CDC Events from Kafka

Use the following to consume events from the topic:

kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic dbserver1.public.testtable \
  --from-beginning

Sample snapshot output:

{
  "before": null,
  "after": {
    "id": 1,
    "message": "CDC test row"
  },
  "op": "r"
}

8. Confirming Live Data Changes

Add a new row in PostgreSQL:

# Log into PostgreSQL
docker compose exec postgres psql -U postgres

# Insert new row
INSERT INTO testtable (id, message) VALUES (2, 'inserted row');

Expected Kafka output:

{
  "before": null,
  "after": {
    "id": 2,
    "message": "inserted row"
  },
  "op": "c"
}

Debezium JSON Event Format

Debezium uses the following format for CDC messages:

`op` value	Meaning
`r`	Snapshot read (initial data)
`c`	Insert (create)
`u`	Update
`d`	Delete

For example, the following JSON means a new row was inserted:

{
  "before": null,
  "after": {
    "id": 2,
    "message": "inserted row"
  },
  "op": "c"
}

Troubleshooting

Topic Not Created in Kafka

If dbserver1.public.testtable doesn’t appear, check:

That debezium_pub exists in PostgreSQL:

SELECT * FROM pg_publication;

That publication.name and slot.name are correctly set in the connector config
That the 00-init.sql script ran (reset if needed):

docker compose down --volumes
docker compose up -d

That Kafka Connect logs are error-free:

docker compose logs connect

No Events Displayed from Kafka

If CDC events aren’t visible in Kafka:

Ensure the connector is RUNNING (see step 5)
Ensure the inserted row has a unique ID
Make sure you’re consuming from the beginning:

kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic dbserver1.public.testtable \
  --from-beginning

In this chapter, we confirmed the flow of change data from PostgreSQL to Kafka using Debezium. In the next chapter, we'll use Flink to process this data.

(Coming soon: Chapter 4 Part 1 — Outputting Kafka CDC Data to Console with Flink)

Connecting RDBs and Search Engines — Chapter 2

sisiodos — Sat, 10 May 2025 22:36:31 +0000

Chapter 2: Understanding the Architecture and Component Relationships

Below is a diagram representing the architecture discussed in this chapter:

graph TD
  subgraph Source
    PG[PostgreSQL]
  end

  subgraph Change Data Capture
    Connect[Debezium Connect]
  end

  subgraph Stream Platform
    ZK[ZooKeeper] --> K[Kafka]
  end

  subgraph Stream Processing
    subgraph Flink
      JM[JobManager] --> TM[TaskManager]
    end
  end

  subgraph Sink
    OS[OpenSearch]
  end

  PG --> Connect --> K --> TM --> OS

This chapter explains how each component interacts to enable real-time data flow into the search engine.

PostgreSQL

PostgreSQL serves as the relational database and source of change data. Initial data is inserted via 00-init.sql, and subsequent changes (INSERT, UPDATE, DELETE) are detected by Debezium.

Debezium leverages PostgreSQL's WAL (Write-Ahead Log) to capture changes. WAL is a transaction log that records all data modifications before they're applied to the actual tables. By reading the WAL, changes can be captured without impacting application logic.

Debezium Connect

Debezium operates as a connector on top of the Kafka Connect framework. Kafka Connect simplifies data integration with Kafka through reusable connectors. These connectors can be registered and managed via Kafka Connect's REST API (covered in the next chapter).

In this setup, the Debezium PostgreSQL connector reads from WAL and emits change events to Kafka.

Kafka + ZooKeeper

Kafka is a high-throughput messaging system used for processing and relaying real-time data. Debezium publishes change events to Kafka topics, which are then consumed by Flink or other downstream systems.

Kafka manages data by topics. Each consumer subscribes to one or more topics to receive data streams. In this case, Flink uses the topic as a streaming source. Topic design is a crucial part of data modeling in streaming systems.

ZooKeeper assists Kafka by managing configuration and cluster state. In this setup, we use a single-node ZooKeeper for simplicity.

Flink JobManager / TaskManager

Flink is a distributed stream processing engine developed under the Apache project. It supports both real-time and batch processing with low latency and high throughput.

Flink jobs consume data from Kafka, perform transformations, and output it to OpenSearch.

JobManager handles job orchestration, while TaskManager executes tasks in parallel.

OpenSearch

OpenSearch is an open-source search and analytics engine based on Elasticsearch. It can store structured and semi-structured data, and provides fast full-text search and aggregation capabilities.

Data is stored in indexes, similar to tables in relational databases. Each index holds multiple documents, which represent individual records.

Stored data can be queried and visualized using tools like OpenSearch Dashboards.

Explanation

There are two main reasons behind this architecture. Let’s examine them:

Normalized Data vs. Search Performance

Some may wonder: “Why not query PostgreSQL directly? Isn’t OpenSearch redundant?”

This architecture deliberately separates the source and the search engine because of differing performance and design goals.

Relational databases like PostgreSQL often use normalized data models to ensure consistency and avoid redundancy. While beneficial for updates, these models require joins across multiple tables, which can slow down search queries.

In contrast, OpenSearch allows data to be denormalized and restructured for optimal search performance. This avoids costly joins and improves response times.

Thus, by separating the update-optimized PostgreSQL from the search-optimized OpenSearch, and managing the transformation in between, the overall system becomes more efficient.

The Value of Real-Time Stream Processing

In this architecture, PostgreSQL ensures consistency, while OpenSearch provides high-performance search over transformed data.

The pipeline—Debezium detecting changes, Kafka transporting them, Flink transforming them, and OpenSearch indexing them—enables real-time propagation of updates.

The benefits of this streaming approach include:

Real-time responsiveness: Unlike batch jobs, changes propagate immediately, allowing up-to-date search results.
Asynchronous replication: Changes can be replicated without impacting application/database performance.
Flexible transformation: Flink enables complex filtering, joins, and aggregations tailored to search use cases.

Change Data Capture (CDC) with stream processing is a key design pattern in modern data infrastructure.

In the next chapter, we'll create tables in PostgreSQL, generate changes, and observe how events are streamed to Kafka topics.

(Coming soon: Chapter 3 — Verifying the Flow from PostgreSQL to Debezium to Kafka)

Connecting RDBs and Search Engines — Chapter 1

sisiodos — Sat, 10 May 2025 22:35:18 +0000

Chapter 1: Setting Up the Environment with Docker Compose

Purpose of This Chapter

The goal of this chapter is to help you build and understand a data pipeline that streams change data from PostgreSQL through stream processing and into OpenSearch, all within your local environment.

About the Code (Repeat)

The docker-compose.yaml file and PostgreSQL initialization scripts used in this chapter are available in the following repository:

👉 https://github.com/sisiodos/rdb-to-search-pipeline-with-flink

If you haven't cloned the repository yet, you can set it up with:

git clone https://github.com/sisiodos/rdb-to-search-pipeline-with-flink.git
cd rdb-to-search-pipeline-with-flink
docker compose up -d

The docker/ directory contains the configuration files, and the postgres/ directory contains SQL scripts for initializing sample data.

Architecture Overview

Here’s a brief explanation of each technology used in this pipeline, so even readers unfamiliar with these tools can follow along:

PostgreSQL: An open-source relational database. It serves as the source of data changes in this setup.
Debezium: A CDC (Change Data Capture) tool that detects changes in the database and publishes them to Kafka in real time.
Kafka: A high-throughput messaging system that temporarily stores the change data from Debezium and relays it to Flink.
Flink: A distributed stream processing engine. It processes the change data from Kafka and sends it to OpenSearch.
OpenSearch: A search engine where the processed data from Flink is stored and made searchable.

This pipeline demonstrates the end-to-end flow of capturing changes from PostgreSQL and delivering them to OpenSearch via Kafka and Flink.

Building the Environment with Docker Compose

Note: This guide assumes a local environment (Mac / Windows / Linux) with Docker Desktop installed. If you're unfamiliar with Docker CLI or Compose, refer to the official Docker documentation.

Docker Compose is a tool for defining and managing multi-container Docker applications. With a single docker-compose.yaml file, you can describe multiple services, including images, ports, environment variables, and dependencies.

In this project, the docker-compose.yaml file defines the following open-source services:

PostgreSQL (sample database)
Debezium Connect (captures changes from PostgreSQL and sends to Kafka)
Kafka + ZooKeeper (messaging infrastructure)
Flink (JobManager and TaskManager)
OpenSearch (search engine)

Starting the Services

To start all services:

docker compose up -d

Stopping the Services

To stop all services:

docker compose down

Managing Individual Services

You can control individual services (e.g., pause, restart, check logs) like this:

docker compose restart kafka            # Restart Kafka service only
docker compose stop connect             # Stop Connect service only
docker compose start connect            # Start Connect service only
docker compose logs flink-taskmanager   # View logs from Flink TaskManager

Pause and Resume

To temporarily suspend or resume all services:

docker compose pause
docker compose unpause

This is useful when you want to reduce CPU or I/O usage or pause processing during log inspection.

Remove Persistent Volumes (Full Cleanup)

To completely clean up the environment, including persistent volumes (e.g., PostgreSQL data), use:

docker compose down --volumes

This removes all container volumes and resets everything to its initial state. See the official documentation for more details.

Connecting RDBs and Search Engines — Preface

sisiodos — Sat, 10 May 2025 22:33:28 +0000

Introduction

Why I Wrote This Guide

In today's system architectures, streaming massive amounts of data into a search engine in real time has become increasingly common. However, the underlying mechanisms—stream processing, data transformation, and integration across systems—pose many hidden complexities.

This guide organizes practical knowledge about OSS-based data pipelines using Kafka, Flink, and OpenSearch, and presents it in a way that allows anyone to learn by actually running and observing the system.

Intended Audience

This guide is aimed at individuals who:

Prefer to experiment and validate systems locally rather than in the cloud
Are interested in stream processing and real-time data integration
Are new to Kafka, Flink, or OpenSearch
Want to try out OSS-based architectures using Docker Compose
Are interested in designing and verifying a data infrastructure but don't know where to start

Even beginners can safely follow along, as the guide explains the background and architecture of each component in a clear and thorough way.

What You'll Learn

After completing this guide, you will be able to:

Build a Kafka → Flink → OpenSearch pipeline using Docker Compose
Understand the basics of Flink's Event Time and Watermark processing, and how to inspect logs
Gain insights into detecting and visualizing data ingestion delays into the search engine

Required Knowledge

This guide assumes basic familiarity with:

Linux command-line operations
Basic Docker usage

Guide Structure

The book is organized into the following chapters:

Setting up the environment using Docker Compose
Understanding the architecture and relationship between components
Extracting change data with Debezium and outputting it to Kafka
Running Flink jobs and analyzing logs
Sending data to a search engine and verifying it

Each chapter includes diagrams and runnable examples to bridge theory with practice.

About the Code

All code, SQL scripts, and Docker Compose configuration files in this guide are available on GitHub:

👉 GitHub repository

Each chapter corresponds to a runnable configuration, so you can follow along and execute each step as you read.

If Docker Desktop is installed, you can get started with:

git clone https://github.com/sisiodos/rdb-to-search-pipeline-with-flink.git
cd rdb-to-search-pipeline-with-flink
docker compose up -d

Chapter-specific files and commands are indicated in the guide to help you follow and try them hands-on.

Future Expansion (Planning Notes)

While this guide focuses on the fundamentals of connecting stream processing with search engines using Flink, we are also planning to explore the following advanced topics:

Flink parallel execution and TaskManager slot configuration
Data distribution and visualization across multiple TaskManagers
Order guarantee and consistency design for Flink → OpenSearch writes
- Upsert control using _id and op fields
Deduplication strategies for handling duplicate messages
Checkpointing, exactly-once processing, and TwoPhaseCommitSink
Design for stable operation and backpressure mitigation
- Analyzing bottlenecks via metrics like currentSendBackPressure, busyTimeMsPerSecond, I/O wait
- Tuning parallelism, slot placement, resource balancing
- Operator chaining, buffer timeouts, Watermark tuning
- Managing large state (joins, deduplication): key selection, TTL, partial aggregation
- Tuning RocksDB StateBackend and spill control
- Using Async I/O Sink, tuning bulk flush/batch size
- Monitoring and visualization using Flink Web UI and Grafana + Prometheus
- Alerting on checkpoint growth and state bloat

These topics go beyond the basics and involve more architectural thinking.

Behind the Scenes

This guide was written in collaboration with OpenAI's conversational AI, ChatGPT-4o. From structure and technical verification to fine-tuning expressions, AI played a major role as a thinking partner throughout the writing process.

That said, the final architecture, structure, verification, and all editorial decisions are the author's responsibility.

The act of documenting and delivering technical insight is no longer the sole domain of human effort. I hope this guide serves as an example of what's possible when human and AI collaborate—and supports you in your own technical explorations.

For what it's worth, during the writing process, ChatGPT-4o used "I" as its pronoun. Somewhere along the way, it became something like a co-author. I leave that note here, quietly.

(Coming soon: Chapter 1 — Setting Up the Environment with Docker Compose)

DEV Community: sisiodos

Connecting RDBs and Search Engines — Chapter 5

Chapter 5: Implementing a CDC Join Pipeline with Flink SQL

Architecture Overview

PostgreSQL Table Setup

Register Debezium Connector

Create OpenSearch Index

Example Flink SQL

Run the Flink Job

Validation Steps

Connecting RDBs and Search Engines — Chapter 4 Part 2

Chapter 4 (Part 2): Integrating Kafka CDC Data with OpenSearch Using Flink

1. Architecture Overview

2. Prerequisites

3. Flink JAR Setup

Required Connector JARs

4. Create OpenSearch Index

5. Create Flink SQL Script

6. Run the Flink Job

7. Verify OpenSearch Output

Connecting RDBs and Search Engines — Chapter 4 Part 1

Chapter 4 (Part 1): Outputting Kafka CDC Data to Console with Flink

1. Prerequisites

2. Architecture Overview

3. Add Kafka Connector to Flink

docker-compose.yaml Excerpt

4. Write Flink SQL Script

Explanation of Flink SQL Tables

Kafka Source Table: cdc_source

Console Sink Table: print_sink

5. Run the Flink SQL Job

Verify Running Jobs

6. Check the Output

Why TaskManager?

7. Troubleshooting

SQL Client Doesn’t Start

No CDC Output Appears

Bonus: Observing Parallelism

Connecting RDBs and Search Engines — Chapter 3

Chapter 3: Verifying the Flow from PostgreSQL to Debezium to Kafka

1. Starting the Environment

2. PostgreSQL Setup

3. PostgreSQL Container Configuration

4. Registering the Debezium Connector

About snapshot.mode

5. Checking Connector Status

6. Checking Kafka Topics

7. Viewing CDC Events from Kafka

8. Confirming Live Data Changes

Debezium JSON Event Format

Troubleshooting

Topic Not Created in Kafka

No Events Displayed from Kafka

Connecting RDBs and Search Engines — Chapter 2

Chapter 2: Understanding the Architecture and Component Relationships

PostgreSQL

Debezium Connect

Kafka + ZooKeeper

Flink JobManager / TaskManager

OpenSearch

Explanation

Normalized Data vs. Search Performance

The Value of Real-Time Stream Processing

Connecting RDBs and Search Engines — Chapter 1

Chapter 1: Setting Up the Environment with Docker Compose

Purpose of This Chapter

About the Code (Repeat)

Architecture Overview

Building the Environment with Docker Compose

Starting the Services

Stopping the Services

Managing Individual Services

Pause and Resume

Remove Persistent Volumes (Full Cleanup)

Related Links

Connecting RDBs and Search Engines — Preface

Introduction

Why I Wrote This Guide

Intended Audience

What You'll Learn

`docker-compose.yaml` Excerpt

Kafka Source Table: `cdc_source`

Console Sink Table: `print_sink`

About `snapshot.mode`