From Binlog to Breakthrough: Real-Time Change Data Capture Pipeline with PyFlink and MySQL

Apache Flink has emerged as the gold standard for stateful stream processing, and with the rise of PyFlink, we can now harness this power without diving deep into Java. Today, we’re walking through a lightweight prototype that bridges MySQL 8 and PyFlink to create a real-time data tap.

containerized mysql8 & pyflink instance

The Concept: Streaming via Binlogs

Traditional data extraction usually involves “polling” a database with expensive SELECT queries. CDC flips this on its head. By tailing the MySQL Binlog (binary log), we can stream every INSERT, UPDATE, and DELETE as an event.

For this prototype, we’ve built a containerized environment using a test_db.products table as our source. Any change made to a product’s price or name in MySQL is instantly "pushed" to our Flink application.

The Blueprint: Dockerized CDC

To keep the experiment clean and reproducible, we use a two-container setup via Docker Compose:

MySQL 8 Instance: Configured specifically for CDC with ROW level binlogging and GTID mode enabled.

flink_cdc_app: A python:3.11-slim container pre-loaded with the OpenJDK JRE and the necessary Flink-MySQL connectors.

The Configuration Secret Sauce

To make this work, the MySQL container isn’t just a standard database. We ensure the following flags are set in our docker-compose.yml:

• --server-id=223345: A unique ID for the Flink "replica."
• --log-bin=mysql-bin: Enables the binary log.
• --binlog_format=ROW: Essential for CDC to capture the actual data changes.

services:
  mysql:
    image: mysql:8.0
    container_name: flink-cdc-mysql
    command:
      - --server-id=223345
      - --log-bin=mysql-bin
      - --binlog_format=ROW
      - --binlog_row_image=FULL
      - --gtid_mode=ON
      - --enforce_gtid_consistency=ON
    environment:
      MYSQL_ROOT_PASSWORD: 123456
      MYSQL_DATABASE: test_db
    ports:
      - "3306:3306"
    volumes:
      - ./docker/mysql/init:/docker-entrypoint-initdb.d
    healthcheck:
      test: ["CMD", "mysqladmin", "ping", "-h", "localhost", "-p123456"]
      interval: 5s
      timeout: 5s
      start_period: 180s
      retries: 60

  pyflink-cdc:
    build:
      context: .
      dockerfile: Dockerfile
    container_name: pyflink-cdc-app
    depends_on:
      mysql:
        condition: service_healthy
    environment:
      MYSQL_HOST: mysql
      MYSQL_PORT: "3306"
      MYSQL_USER: root
      MYSQL_PASSWORD: 123456
      MYSQL_DATABASE: test_db
      MYSQL_TABLE: products
      CDC_CONNECTOR_JAR: file:///opt/flink/connectors/flink-sql-connector-mysql-cdc-3.4.0.jar
      MYSQL_JDBC_JAR: file:///opt/flink/connectors/mysql-connector-j-8.4.0.jar

The bind mount ./docker/mysql/init/01-init.sql → /docker-entrypoint-initdb.d initializes the dummy data for our test.

CREATE DATABASE IF NOT EXISTS test_db;
USE test_db;

CREATE TABLE IF NOT EXISTS products (
    id INT PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    price DECIMAL(10, 2) NOT NULL
);

INSERT INTO products (id, name, price) VALUES
    (1, 'keyboard', 49.99),
    (2, 'mouse', 19.99),
    (3, 'monitor', 199.99)
ON DUPLICATE KEY UPDATE
    name = VALUES(name),
    price = VALUES(price);

The Heart of the App: flink_cdc_app.py

The logic of our streaming pipeline is surprisingly concise. It performs four critical operations:

1. Environment Setup: Initializes a Streaming Table Environment.
2. Fault Tolerance: Enables checkpointing every 3 seconds, ensuring that if the app crashes, it knows exactly where it left off in the binlog.
3. Jar Integration: Dynamically links the flink-sql-connector-mysql-cdc and the mysql-connector-j drivers.
4. The CDC Source: Defines a virtual table that points to our physical MySQL table.

import os

from pyflink.table import EnvironmentSettings, TableEnvironment

env_settings = EnvironmentSettings.in_streaming_mode()
t_env = TableEnvironment.create(env_settings)

t_env.get_config().set("execution.checkpointing.interval", "3000ms")
cdc_connector_jar = os.getenv(
    "CDC_CONNECTOR_JAR",
    "file:///opt/flink/connectors/flink-sql-connector-mysql-cdc-3.4.0.jar",
)
mysql_jdbc_jar = os.getenv(
    "MYSQL_JDBC_JAR",
    "file:///opt/flink/connectors/mysql-connector-j-8.4.0.jar",
)
t_env.get_config().set("pipeline.jars", ";".join([cdc_connector_jar, mysql_jdbc_jar]))

mysql_host = os.getenv("MYSQL_HOST", "mysql")
mysql_port = os.getenv("MYSQL_PORT", "3306")
mysql_user = os.getenv("MYSQL_USER", "root")
mysql_password = os.getenv("MYSQL_PASSWORD", "123456")
mysql_database = os.getenv("MYSQL_DATABASE", "test_db")
mysql_table = os.getenv("MYSQL_TABLE", "products")

t_env.execute_sql("""
    CREATE TABLE mysql_cdc_source (
        id INT,
        name STRING,
        price DECIMAL(10, 2),
        PRIMARY KEY (id) NOT ENFORCED
    ) WITH (
        'connector' = 'mysql-cdc',
        'hostname' = '{hostname}',
        'port' = '{port}',
        'username' = '{username}',
        'password' = '{password}',
        'database-name' = '{database}',
        'table-name' = '{table}',
        'scan.startup.mode' = 'initial'
    )
""".format(
    hostname=mysql_host,
    port=mysql_port,
    username=mysql_user,
    password=mysql_password,
    database=mysql_database,
    table=mysql_table,
))

t_env.execute_sql("SELECT * FROM mysql_cdc_source").print()

By setting scan.startup.mode to initial, the app performs a snapshot of the existing data first, then transitions seamlessly into "tailing" new changes. We need both the flink sql connector mysql cdc jar and mysql native connectivity jar for our prototype. Prepare the Dockerfile accordingly:

FROM python:3.11-slim

ENV PYTHONUNBUFFERED=1
WORKDIR /app

RUN apt-get update \
    && apt-get install -y --no-install-recommends curl ca-certificates openjdk-21-jre \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

RUN mkdir -p /opt/flink/connectors \
    && curl -fsSL \
        -o /opt/flink/connectors/flink-sql-connector-mysql-cdc-3.4.0.jar \
        https://repo1.maven.org/maven2/org/apache/flink/flink-sql-connector-mysql-cdc/3.4.0/flink-sql-connector-mysql-cdc-3.4.0.jar \
    && curl -fsSL \
        -o /opt/flink/connectors/mysql-connector-j-8.4.0.jar \
        https://repo1.maven.org/maven2/com/mysql/mysql-connector-j/8.4.0/mysql-connector-j-8.4.0.jar

COPY flink_cdc_app.py /app/flink_cdc_app.py

CMD ["python", "/app/flink_cdc_app.py"]

Observations from the Lab

When we launch the stack, the magic happens in the logs. On startup, PyFlink reads our initial inventory (Keyboards, Mice, Monitors).

initial test_db.products contents

The moment we run an update in MySQL: UINSERT INTO products (id, name, price) VALUES (10, 'webcam', 89.99); UPDATE products SET price = 79.99 WHERE id = 10;The PyFlink console immediately prints the change event. It doesn’t wait for a cron job or a refresh; the event is pushed through the pipeline in milliseconds. Notice how flink captures the different operations under op column: INSERT, UPDATE, DELETE etc.

change data captures

Lessons Learned

• The Good: The setup is incredibly reactive. Using Checkpointing makes the pipeline resilient, and the Table API makes SQL-like operations on streams very intuitive.
• The Basic: For this prototype, we printed to stdout. In a production scenario, you would swap .print() for a Sink—writing the changes to Kafka, an S3 Data Lake, or an Elasticsearch index.

Conclusion

Building a real-time CDC pipeline no longer requires a massive infrastructure team and thousands of lines of Java. With Apache Flink we can build robust, “always-on” data feeds that keep downstream systems in perfect sync with the source of truth.