From be2bbcbd92642560ae544223518b1b1e6fbc65b0 Mon Sep 17 00:00:00 2001 From: mo Date: Fri, 26 Jun 2026 09:21:17 +0000 Subject: [PATCH] feat(cdc): app-level Cassandra+Neo4j change events to Kafka (->S3) on generate --- infra/airflow/scripts/cdc_emit.py | 66 +++++++++++++++++++ .../generate_cassandra_telemetry_data.py | 22 ++++++- .../scripts/generate_neo4j_graph_data.py | 16 +++++ 3 files changed, 103 insertions(+), 1 deletion(-) create mode 100644 infra/airflow/scripts/cdc_emit.py diff --git a/infra/airflow/scripts/cdc_emit.py b/infra/airflow/scripts/cdc_emit.py new file mode 100644 index 0000000..b6bc4c5 --- /dev/null +++ b/infra/airflow/scripts/cdc_emit.py @@ -0,0 +1,66 @@ +"""Lightweight change-event emitter to Kafka for sources without a native +Debezium connector (Cassandra 4.1 with cdc disabled, Neo4j 4.4 Community). + +This performs an application-level dual-write: as the generators insert data, +each new record is also published to Kafka as a Debezium-style change event. +The platform's all-topics S3 consumer then archives these to object storage, +so Cassandra/Neo4j changes flow DB -> Kafka -> S3 just like the CDC sources. + +Best-effort: never raises, so data generation cannot be broken by Kafka issues. +""" +import json +import os +import time + +KAFKA_BOOTSTRAP = os.getenv("CDC_KAFKA_BOOTSTRAP", "10.0.21.36:9092") + +_producer = None + + +def _default(o): + # make datetimes / uuids / Decimals JSON-serializable + try: + import datetime + if isinstance(o, (datetime.datetime, datetime.date)): + return o.isoformat() + except Exception: + pass + return str(o) + + +def _get_producer(): + global _producer + if _producer is None: + from kafka import KafkaProducer + _producer = KafkaProducer( + bootstrap_servers=KAFKA_BOOTSTRAP, + value_serializer=lambda v: json.dumps(v, default=_default).encode("utf-8"), + linger_ms=50, + acks=1, + retries=3, + request_timeout_ms=20000, + ) + return _producer + + +def emit_changes(topic, db, table, records): + """Publish a list of dict records as change events. Returns count sent.""" + if not records: + return 0 + try: + p = _get_producer() + now = int(time.time() * 1000) + for r in records: + evt = { + "op": "c", + "ts_ms": now, + "source": {"connector": "app-cdc", "db": db, "table": table}, + "after": r, + } + p.send(topic, evt) + p.flush(timeout=30) + print(f"[cdc] emitted {len(records)} change events -> {topic}") + return len(records) + except Exception as e: + print(f"[cdc] emit warning ({topic}): {str(e)[:200]}") + return 0 diff --git a/infra/airflow/scripts/generate_cassandra_telemetry_data.py b/infra/airflow/scripts/generate_cassandra_telemetry_data.py index bd7ae4e..d8fe6b3 100644 --- a/infra/airflow/scripts/generate_cassandra_telemetry_data.py +++ b/infra/airflow/scripts/generate_cassandra_telemetry_data.py @@ -9,6 +9,13 @@ import random from datetime import datetime, timedelta import uuid import sys +import os +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +try: + from cdc_emit import emit_changes +except Exception: + def emit_changes(*a, **k): + return 0 # Database connection details DB_HOST = "10.0.21.51" @@ -57,12 +64,25 @@ def main(): ) total_generated = 0 + cdc_buffer = [] + CDC_TOPIC = "cassandra_telemetry.cdc.device_metrics" for i in range(TARGET_ROWS): - session.execute(insert_stmt, generate_fake_device_metric()) + row = generate_fake_device_metric() + session.execute(insert_stmt, row) total_generated += 1 + cdc_buffer.append({ + "device_id": row[0], "metric_ts": row[1], "metric_type": row[2], + "metric_value": row[3], "payload": row[4], + }) + if len(cdc_buffer) >= 1000: + emit_changes(CDC_TOPIC, "telemetry", "device_metrics", cdc_buffer) + cdc_buffer = [] if total_generated % 1000 == 0: print(f"Generated {total_generated} rows...") + if cdc_buffer: + emit_changes(CDC_TOPIC, "telemetry", "device_metrics", cdc_buffer) + session.shutdown() cluster.shutdown() diff --git a/infra/airflow/scripts/generate_neo4j_graph_data.py b/infra/airflow/scripts/generate_neo4j_graph_data.py index dcf3176..59a5f82 100644 --- a/infra/airflow/scripts/generate_neo4j_graph_data.py +++ b/infra/airflow/scripts/generate_neo4j_graph_data.py @@ -8,6 +8,16 @@ from neo4j import GraphDatabase import random import uuid import sys +import os +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +try: + from cdc_emit import emit_changes +except Exception: + def emit_changes(*a, **k): + return 0 + +NODES_TOPIC = "neo4j_graph.cdc.nodes" +RELS_TOPIC = "neo4j_graph.cdc.relationships" # Database connection details DB_HOST = "10.0.21.51" @@ -121,6 +131,7 @@ def main(): batch=batch ) total_products += len(batch) + emit_changes(NODES_TOPIC, "graph", "Product", batch) batch = [] if total_products % 50000 == 0: @@ -143,6 +154,7 @@ def main(): batch=batch ) total_products += len(batch) + emit_changes(NODES_TOPIC, "graph", "Product", batch) print(f"Generated {total_products} product nodes.") @@ -172,6 +184,7 @@ def main(): batch=batch ) total_suppliers += len(batch) + emit_changes(NODES_TOPIC, "graph", "Supplier", batch) batch = [] if batch: @@ -189,6 +202,7 @@ def main(): batch=batch ) total_suppliers += len(batch) + emit_changes(NODES_TOPIC, "graph", "Supplier", batch) print(f"Generated {total_suppliers} supplier nodes.") @@ -213,6 +227,7 @@ def main(): if len(batch) >= BATCH_SIZE: _flush_rels(session, batch) total_relationships += len(batch) + emit_changes(RELS_TOPIC, "graph", "RELATIONSHIP", batch) batch = [] if total_relationships % 50000 == 0: @@ -221,6 +236,7 @@ def main(): if batch: _flush_rels(session, batch) total_relationships += len(batch) + emit_changes(RELS_TOPIC, "graph", "RELATIONSHIP", batch) print(f"Created {total_relationships} relationships.")