Streaming Binlog Events with BinLogStreamReader #

When you need MySQL row changes as decoded events in Python — for change-data capture, for building a custom validator over row images, or for driving recovery logic finer than the mysqlbinlog CLI allows — BinLogStreamReader from python-mysql-replication is the tool. But the naive stream that iterates events and forgets its position loses its place on every restart, re-processes duplicates, and stalls the first time a table’s schema changes underneath it. This page builds a durable stream: it registers with a unique server_id, resumes from a persisted GTID position, filters to the schemas and events that matter, and survives schema changes. It is the practical, data-plane counterpart to the library selection in Python Binlog Parsing Library Reference.

Context & Prerequisites #

This implements the data-plane role described in Python Binlog Parsing Library Reference. You need Python 3.10+, python-mysql-replication installed, MySQL 8.0.22+ with binlog_format=ROW (a hard requirement for row-event decoding, per ROW vs STATEMENT vs MIXED Formats), gtid_mode=ON, and a dedicated account holding REPLICATION SLAVE and REPLICATION CLIENT, scoped per Security & Access Frameworks.

Step-by-Step Implementation #

1. Register with a unique server_id and filters #

Give the reader an id distinct from every replica, and filter to only the events and schemas you consume so the primary sends less.

# Python 3.10+
from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.row_event import WriteRowsEvent, UpdateRowsEvent, DeleteRowsEvent

def make_reader(settings: dict, server_id: int, schemas: list[str], gtid: str | None):
    return BinLogStreamReader(
        connection_settings=settings,
        server_id=server_id,                      # unique among all replicas/readers
        only_events=[WriteRowsEvent, UpdateRowsEvent, DeleteRowsEvent],
        only_schemas=schemas,
        auto_position=gtid,                        # resume from a GTID set, if we have one
        blocking=True,
        resume_stream=True,
    )

PITR relevance: filtering server-side reduces the event volume the reader must process, keeping reader lag — and therefore any recovery signal derived from it — low.

2. Persist the GTID position after each processed batch #

Record the GTID up to which events are durably handled, atomically, so a restart resumes exactly there.

# Python 3.10+
import json, os
from pathlib import Path

POS = Path("/var/lib/binlog-reader/position.json")

def save_position(gtid: str) -> None:
    tmp = POS.with_suffix(".tmp")
    tmp.write_text(json.dumps({"gtid": gtid}))
    os.replace(tmp, POS)          # atomic; a crash never leaves a torn position

def load_position() -> str | None:
    return json.loads(POS.read_text())["gtid"] if POS.exists() else None

PITR relevance: a durable GTID position makes the stream idempotent across restarts — you reprocess at most a bounded tail, never skip events, which is the same exactly-once discipline the archiver uses.

3. Consume events and advance the position #

Process each row, then advance the saved position to the reader’s current GTID.

# Python 3.10+
def run(reader, handle_row) -> None:
    try:
        for event in reader:
            for row in event.rows:
                handle_row(event.schema, event.table, type(event).__name__, row)
            if reader.log_pos and (gtid := getattr(reader, "auto_position", None)):
                save_position(gtid)      # advance only after the batch is handled
    finally:
        reader.close()

PITR relevance: advancing the position only after handling guarantees that a crash mid-batch resumes before the unhandled events, preserving completeness of any derived change record.

4. Handle schema changes without stalling #

A DDL on a streamed table changes its column layout; refresh the reader’s table metadata so subsequent row events map correctly.

PITR relevance: an unhandled schema change is the most common cause of a silently stalled reader; detecting the DDL and refreshing metadata keeps the stream — and its lag metric — alive.

Configuration Snippet & Reference Table #

Verification Checklist #

• The reader’s server_id is unique among all replicas and readers.
• binlog_format=ROW and gtid_mode=ON are confirmed before streaming.
• The GTID position is persisted atomically after each handled batch.
• A restart resumes from the saved position with no skipped events and only a bounded reprocessed tail.
• Server-side only_events/only_schemas filters are set to minimize volume.
• A DDL on a streamed table does not stall the reader (metadata refresh handled).

Gotchas & Version-Specific Caveats #

Requested position already purged. Could not find first log file name in binary log index file means the resume GTID was purged from the primary; the reader must restart from an available position, which is exactly why event streaming and file archiving are complementary — the archive holds what the primary has purged.

Duplicate server_id. Two readers (or a reader and a replica) sharing an id causes the primary to drop one connection; assign distinct ids.

Schema changes need metadata refresh. After a table’s DDL, cached column metadata is stale; failing to refresh either mis-maps rows or stalls the reader.

Row-based logging is mandatory. Under statement or mixed logging there are no row events to decode; the reader yields nothing useful. Enforce binlog_format=ROW upstream.

Frequently Asked Questions #

Related #

• Python Binlog Parsing Library Reference — where this library fits among the three.
• mysql-connector-python vs PyMySQL vs python-mysql-replication for Binlog Parsing — the selection guide.
• mysqlbinlog Replay Scripting — the applying counterpart to this reading tool.

Back to Python Binlog Parsing Library Reference.