I can't retrieve the updated or written field names; all I get are things like UNKNOWN_COL0.

import base64
from datetime import date, datetime
import json
import traceback
import pymysql
from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.row_event import (
DeleteRowsEvent,
UpdateRowsEvent,
WriteRowsEvent,
TableMapEvent
)

def default(obj):
if isinstance(obj, datetime):
return obj.isoformat()
elif isinstance(obj, date):
return obj.isoformat()
elif isinstance(obj, bytes):
return base64.b64encode(obj).decode('ascii')
raise TypeError(f"Object of type {obj.__class__.__name__} is not JSON serializable")

class BinlogListener:
def __init__(self, mysql_settings):
self.mysql_settings = mysql_settings
self.stream = None
self.table_map = {}

def start_stream(self):
events = [DeleteRowsEvent, UpdateRowsEvent, WriteRowsEvent]

self.stream = BinLogStreamReader(
connection_settings=self.mysql_settings,
server_id=101,
only_events=events,
resume_stream=True,
blocking=True,
)

def process_events(self):
if self.stream is None:
self.start_stream()

for binlogevent in self.stream:
try:
binlogevent.dump()
for row in binlogevent.rows:

event = {"schema": binlogevent.schema, "table": binlogevent.table}
print(event)
if isinstance(binlogevent, DeleteRowsEvent):
event["action"] = "delete"
event["data"] = row["values"]

elif isinstance(binlogevent, UpdateRowsEvent):
event["action"] = "update"
event["data"] = row["after_values"]

elif isinstance(binlogevent, WriteRowsEvent):
event["action"] = "insert"
event["data"] = row["values"]

print(json.dumps(event, default=default))

except Exception as e:
traceback.print_exc()

def stop_stream(self):
if self.stream is not None:
self.stream.close()
self.stream = None

Start with the BinLogStreamReader setup and row handling shown in the report, then inspect how TableMapEvent metadata relates to the values returned for UpdateRowsEvent and WriteRowsEvent. Done means written and updated row data exposes the actual MySQL field names instead of names such as UNKNOWN_COL0; add a regression test if a suitable event test location exists.

Written by the indexing model from the issue text.