Change Data Capture (CDC)

<h2 id="change-data-capture-cdc" class="position-relative d-flex align-items-center group"> Change Data Capture (CDC) <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="change-data-capture-cdc" aria-haspopup="dialog" aria-label="Share link: Change Data Capture (CDC)"> Share link </button> </h2><div id="headingShareModal" class="heading-share-modal" role="dialog" aria-modal="true" aria-labelledby="headingShareTitle" hidden> <div class="hsm-dialog" role="document"> <div class="hsm-header"> <h2 id="headingShareTitle" class="h6 mb-0 fw-bold">Share this section</h2> <button type="button" class="hsm-close" aria-label="Close"> </button> </div> <div class="hsm-body"> <label for="headingShareInput" class="form-label small text-muted mb-1 text-uppercase fw-bold" style="font-size: 0.7rem; letter-spacing: 0.5px;">Permalink</label> <div class="input-group mb-4 hsm-url-group"> <input id="headingShareInput" type="text" class="form-control font-monospace" readonly aria-readonly="true" style="font-size: 0.85rem;" /> <button class="btn btn-primary hsm-copy" type="button" aria-label="Copy" title="Copy"> </button> </div> <div class="small fw-bold mb-2 text-muted text-uppercase" style="font-size: 0.7rem; letter-spacing: 0.5px;">Share via</div> <div class="hsm-share-grid"> <a id="share-twitter" class="btn btn-outline-secondary w-100" target="_blank" rel="noopener noreferrer"> Twitter </a> <a id="share-linkedin" class="btn btn-outline-secondary w-100" target="_blank" rel="noopener noreferrer"> LinkedIn </a> <a id="share-facebook" class="btn btn-outline-secondary w-100" target="_blank" rel="noopener noreferrer"> Facebook </a> </div> </div> </div> </div> <style> .heading-share-modal { position: fixed; inset: 0; display: flex; justify-content: center; align-items: center; background: rgba(0, 0, 0, 0.6); z-index: 1050; padding: 1rem; backdrop-filter: blur(4px); -webkit-backdrop-filter: blur(4px); } .heading-share-modal[hidden] { display: none !important; } .hsm-dialog { max-width: 420px; width: 100%; background: var(--bs-body-bg, #fff); color: var(--bs-body-color, #212529); border: 1px solid var(--bs-border-color, rgba(0,0,0,0.1)); border-radius: 1rem; box-shadow: 0 25px 50px -12px rgba(0, 0, 0, 0.25); overflow: hidden; animation: hsm-fade-in 0.2s ease-out; } @keyframes hsm-fade-in { from { opacity: 0; transform: scale(0.95); } to { opacity: 1; transform: scale(1); } } [data-bs-theme="dark"] .hsm-dialog { background: #1e293b; border-color: rgba(255,255,255,0.1); color: #f8f9fa; } .hsm-header { display: flex; justify-content: space-between; align-items: center; padding: 1rem 1.5rem; border-bottom: 1px solid var(--bs-border-color, rgba(0,0,0,0.1)); background: rgba(0,0,0,0.02); } [data-bs-theme="dark"] .hsm-header { background: rgba(255,255,255,0.02); border-color: rgba(255,255,255,0.1); } .hsm-close { background: transparent; border: none; color: inherit; opacity: 0.5; padding: 0.25rem 0.5rem; border-radius: 0.25rem; font-size: 1.2rem; line-height: 1; transition: opacity 0.2s; } .hsm-close:hover { opacity: 1; } .hsm-body { padding: 1.5rem; } .hsm-url-group { display: flex !important; align-items: stretch; } .hsm-url-group .form-control { flex: 1; min-width: 0; margin: 0; background: var(--bs-secondary-bg, #f8f9fa); border-color: var(--bs-border-color, #dee2e6); border-top-right-radius: 0; border-bottom-right-radius: 0; height: 42px; } .hsm-url-group .btn { flex: 0 0 auto; margin: 0; margin-left: -1px; border-top-left-radius: 0; border-bottom-left-radius: 0; height: 42px; display: flex; align-items: center; justify-content: center; padding: 0 1.25rem; z-index: 2; } [data-bs-theme="dark"] .hsm-url-group .form-control { background: #0f172a; border-color: #334155; color: #e2e8f0; } .hsm-share-grid { display: flex; flex-direction: column; gap: 0.5rem; } .hsm-share-grid .btn { display: flex; align-items: center; justify-content: center; font-size: 0.9rem; padding: 0.6rem; border-color: var(--bs-border-color); width: 100%; } [data-bs-theme="dark"] .hsm-share-grid .btn { color: #e2e8f0; border-color: #475569; } [data-bs-theme="dark"] .hsm-share-grid .btn:hover { background: #334155; border-color: #cbd5e1; } </style> <script> (function(){ const modal = document.getElementById('headingShareModal'); if(!modal) return; const input = modal.querySelector('#headingShareInput'); const copyBtn = modal.querySelector('.hsm-copy'); const twitter = modal.querySelector('#share-twitter'); const linkedin = modal.querySelector('#share-linkedin'); const facebook = modal.querySelector('#share-facebook'); const closeBtn = modal.querySelector('.hsm-close'); let lastFocus=null; let trapBound=false; function buildUrl(id){ return window.location.origin + window.location.pathname + '#' + id; } function isOpen(){ return !modal.hasAttribute('hidden'); } function hydrate(id){ const url=buildUrl(id); input.value=url; const enc=encodeURIComponent(url); const text=encodeURIComponent(document.title); if(twitter) twitter.href=`https://twitter.com/intent/tweet?url=${enc}&text=${text}`; if(linkedin) linkedin.href=`https://www.linkedin.com/sharing/share-offsite/?url=${enc}`; if(facebook) facebook.href=`https://www.facebook.com/sharer/sharer.php?u=${enc}`; } function openModal(id){ lastFocus=document.activeElement; hydrate(id); if(!isOpen()){ modal.removeAttribute('hidden'); } requestAnimationFrame(()=>{ input.focus(); }); trapFocus(); } function closeModal(){ if(!isOpen()) return; modal.setAttribute('hidden',''); if(lastFocus && typeof lastFocus.focus==='function') lastFocus.focus(); } function copyCurrent(){ try{ navigator.clipboard.writeText(input.value).then(()=>feedback(true),()=>fallback()); } catch(e){ fallback(); } } function fallback(){ input.select(); try{ document.execCommand('copy'); feedback(true);}catch(e){ feedback(false);} } function feedback(ok){ if(!copyBtn) return; const icon=copyBtn.querySelector('i'); if(!icon) return; const prev=copyBtn.getAttribute('data-prev')||icon.className; if(!copyBtn.getAttribute('data-prev')) copyBtn.setAttribute('data-prev',prev); icon.className= ok ? 'fa-duotone fa-clipboard-check':'fa-duotone fa-circle-exclamation'; setTimeout(()=>{ icon.className=prev; },1800); } function handleShareClick(e){ e.preventDefault(); const btn=e.currentTarget; const id=btn.getAttribute('data-share-target'); if(id) openModal(id); } function bindShareButtons(){ document.querySelectorAll('.h-share').forEach(btn=>{ if(!btn.dataset.hShareBound){ btn.addEventListener('click', handleShareClick); btn.dataset.hShareBound='1'; } }); } bindShareButtons(); if(document.readyState==='loading'){ document.addEventListener('DOMContentLoaded', bindShareButtons); } else { requestAnimationFrame(bindShareButtons); } document.addEventListener('click', function(e){ const shareBtn=e.target.closest && e.target.closest('.h-share'); if(shareBtn && !shareBtn.dataset.hShareBound){ handleShareClick.call(shareBtn, e); } }, true); document.addEventListener('click', e=>{ if(e.target===modal) closeModal(); if(e.target.closest && e.target.closest('.hsm-close')){ e.preventDefault(); closeModal(); } if(copyBtn && (e.target===copyBtn || (e.target.closest && e.target.closest('.hsm-copy')))) { e.preventDefault(); copyCurrent(); } }); document.addEventListener('keydown', e=>{ if(e.key==='Escape' && isOpen()) closeModal(); }); function trapFocus(){ if(trapBound) return; trapBound=true; modal.addEventListener('keydown', f=>{ if(f.key==='Tab' && isOpen()){ const focusable=[...modal.querySelectorAll('a[href],button,input,textarea,select,[tabindex]:not([tabindex="-1"])')].filter(el=>!el.hasAttribute('disabled')); if(!focusable.length) return; const first=focusable[0]; const last=focusable[focusable.length-1]; if(f.shiftKey && document.activeElement===first){ f.preventDefault(); last.focus(); } else if(!f.shiftKey && document.activeElement===last){ f.preventDefault(); first.focus(); } } }); } if(closeBtn) closeBtn.addEventListener('click', e=>{ e.preventDefault(); closeModal(); }); })(); </script>Change Data Capture (CDC) is a design pattern that identifies and captures changes made to data in a database and delivers those changes to downstream systems in real-time. Geode’s CDC implementation enables applications to react to data modifications as they occur, powering use cases like real-time analytics, data synchronization, and event-driven architectures. <h3 id="understanding-cdc" class="position-relative d-flex align-items-center group"> Understanding CDC <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="understanding-cdc" aria-haspopup="dialog" aria-label="Share link: Understanding CDC"> Share link </button> </h3> <h4 id="what-is-cdc" class="position-relative d-flex align-items-center group"> What is CDC? <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="what-is-cdc" aria-haspopup="dialog" aria-label="Share link: What is CDC?"> Share link </button> </h4>CDC monitors database changes and publishes events for: Insert Events - New nodes or edges created Update Events - Property modifications Delete Events - Nodes or edges removed Schema Events - Structural changes to graphs <h4 id="why-use-cdc" class="position-relative d-flex align-items-center group"> Why Use CDC? <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="why-use-cdc" aria-haspopup="dialog" aria-label="Share link: Why Use CDC?"> Share link </button> </h4>CDC enables powerful scenarios: <ul> <li>Real-Time Analytics - Update dashboards instantly</li> <li>Cache Invalidation - Maintain cache consistency</li> <li>Data Synchronization - Replicate to other systems</li> <li>Audit Trails - Track all data modifications</li> <li>Event-Driven Architecture - Trigger business workflows</li> </ul> <h3 id="geode-cdc-architecture" class="position-relative d-flex align-items-center group"> Geode CDC Architecture <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="geode-cdc-architecture" aria-haspopup="dialog" aria-label="Share link: Geode CDC Architecture"> Share link </button> </h3> <h4 id="change-event-structure" class="position-relative d-flex align-items-center group"> Change Event Structure <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="change-event-structure" aria-haspopup="dialog" aria-label="Share link: Change Event Structure"> Share link </button> </h4>Each change produces a detailed event: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-json" data-lang="json">{ "event_id": "evt_abc123", "timestamp": "2026-01-24T10:30:00Z", "graph": "social", "operation": "INSERT", "entity_type": "node", "entity_id": "person:123", "before": null, "after": { "labels": ["Person"], "properties": { "name": "Alice", "age": 25, "created": "2026-01-24T10:30:00Z" } }, "transaction_id": "tx_xyz789", "user": "[email protected]" } </code></pre></div> <h4 id="enabling-cdc" class="position-relative d-flex align-items-center group"> Enabling CDC <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="enabling-cdc" aria-haspopup="dialog" aria-label="Share link: Enabling CDC"> Share link </button> </h4>Enable CDC for specific graphs or globally: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Enable CDC for entire graph ALTER GRAPH social SET CDC ENABLED; -- Enable CDC for specific labels ALTER GRAPH social SET CDC ENABLED FOR LABELS (Person, Post); -- Enable CDC for specific operations ALTER GRAPH social SET CDC ENABLED FOR OPERATIONS (INSERT, UPDATE); </code></pre></div> <h4 id="cdc-streams" class="position-relative d-flex align-items-center group"> CDC Streams <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="cdc-streams" aria-haspopup="dialog" aria-label="Share link: CDC Streams"> Share link </button> </h4>Changes flow through named streams: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Create CDC stream CREATE CDC STREAM user_changes ON GRAPH social FOR LABELS (Person) WHERE operation IN ('INSERT', 'UPDATE') OPTIONS ( format: 'json', include_before: true, include_after: true, buffer_size: 10000 ); </code></pre></div> <h3 id="consuming-change-events" class="position-relative d-flex align-items-center group"> Consuming Change Events <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="consuming-change-events" aria-haspopup="dialog" aria-label="Share link: Consuming Change Events"> Share link </button> </h3> <h4 id="polling-consumer" class="position-relative d-flex align-items-center group"> Polling Consumer <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="polling-consumer" aria-haspopup="dialog" aria-label="Share link: Polling Consumer"> Share link </button> </h4>Pull events on demand: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Read events from stream SELECT * FROM CDC_STREAM('user_changes') WHERE timestamp > NOW() - INTERVAL '1 hour' ORDER BY timestamp ASC LIMIT 1000; </code></pre></div> <h4 id="push-consumer" class="position-relative d-flex align-items-center group"> Push Consumer <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="push-consumer" aria-haspopup="dialog" aria-label="Share link: Push Consumer"> Share link </button> </h4>Subscribe for real-time delivery: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">from geode_client import Client async def handle_change(event): print(f"Change detected: {event['operation']} on {event['entity_id']}") if event['operation'] == 'INSERT': # Handle new entity await process_new_entity(event['after']) elif event['operation'] == 'UPDATE': # Handle modification await process_update(event['before'], event['after']) elif event['operation'] == 'DELETE': # Handle deletion await process_deletion(event['before']) async def main(): client = Client(host="localhost", port=3141) last_seen = "1970-01-01T00:00:00Z" # Poll application-managed ChangeLog entries async with client.connection() as conn: while True: result, _ = await conn.query( """ MATCH (e:ChangeLog) WHERE e.emitted_at > $since RETURN e.operation AS operation, e.entity_id AS entity_id, e.before AS before, e.after AS after, e.emitted_at AS emitted_at ORDER BY emitted_at """, {"since": last_seen}, ) for row in result.rows: event = { "operation": row["operation"].raw_value, "entity_id": row["entity_id"].raw_value, "before": row["before"].raw_value, "after": row["after"].raw_value, } await handle_change(event) last_seen = row["emitted_at"].raw_value </code></pre></div> <h4 id="event-filtering" class="position-relative d-flex align-items-center group"> Event Filtering <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="event-filtering" aria-haspopup="dialog" aria-label="Share link: Event Filtering"> Share link </button> </h4>Filter events before delivery: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Only high-value users CREATE CDC STREAM vip_users ON GRAPH social FOR LABELS (Person) WHERE after.properties.lifetime_value > 10000; -- Only specific property changes CREATE CDC STREAM email_changes ON GRAPH social FOR LABELS (Person) WHERE before.properties.email != after.properties.email; </code></pre></div> <h3 id="cdc-patterns" class="position-relative d-flex align-items-center group"> CDC Patterns <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="cdc-patterns" aria-haspopup="dialog" aria-label="Share link: CDC Patterns"> Share link </button> </h3> <h4 id="database-replication" class="position-relative d-flex align-items-center group"> Database Replication <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="database-replication" aria-haspopup="dialog" aria-label="Share link: Database Replication"> Share link </button> </h4>Replicate changes to other databases: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def replicate_to_postgres(event): """Replicate Geode changes to PostgreSQL""" if event['operation'] == 'INSERT': await pg_conn.execute( "INSERT INTO users (id, name, age) VALUES ($1, $2, $3)", event['entity_id'], event['after']['properties']['name'], event['after']['properties']['age'] ) elif event['operation'] == 'UPDATE': await pg_conn.execute( "UPDATE users SET name = $2, age = $3 WHERE id = $1", event['entity_id'], event['after']['properties']['name'], event['after']['properties']['age'] ) elif event['operation'] == 'DELETE': await pg_conn.execute( "DELETE FROM users WHERE id = $1", event['entity_id'] ) async def main(): last_seen = "1970-01-01T00:00:00Z" async with client.connection() as conn: while True: result, _ = await conn.query( """ MATCH (e:ChangeLog) WHERE e.emitted_at > $since RETURN e.operation AS operation, e.entity_id AS entity_id, e.after AS after, e.emitted_at AS emitted_at ORDER BY emitted_at """, {"since": last_seen}, ) for row in result.rows: event = { "operation": row["operation"].raw_value, "entity_id": row["entity_id"].raw_value, "after": row["after"].raw_value, } await replicate_to_postgres(event) last_seen = row["emitted_at"].raw_value </code></pre></div> <h4 id="cache-invalidation" class="position-relative d-flex align-items-center group"> Cache Invalidation <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="cache-invalidation" aria-haspopup="dialog" aria-label="Share link: Cache Invalidation"> Share link </button> </h4>Keep caches synchronized: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-javascript" data-lang="javascript">const redis = require('redis'); const { createClient } = require('@geodedb/client'); const cache = redis.createClient(); async function pollChanges() { const geode = await createClient('quic://localhost:3141'); let lastSeen = '1970-01-01T00:00:00Z'; while (true) { const rows = await geode.queryAll( ` MATCH (e:ChangeLog) WHERE e.emitted_at > $since RETURN e.operation AS operation, e.entity_id AS entity_id, e.after AS after, e.emitted_at AS emitted_at ORDER BY emitted_at `, { params: { since: lastSeen } } ); for (const row of rows) { const cacheKey = `user:${row.entity_id}`; if (row.operation === 'DELETE') { await cache.del(cacheKey); } else { await cache.setEx(cacheKey, 3600, JSON.stringify(row.after)); } lastSeen = row.emitted_at; } } } pollChanges(); </code></pre></div> <h4 id="search-index-synchronization" class="position-relative d-flex align-items-center group"> Search Index Synchronization <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="search-index-synchronization" aria-haspopup="dialog" aria-label="Share link: Search Index Synchronization"> Share link </button> </h4>Update search indexes in real-time: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">from elasticsearch import Elasticsearch es = Elasticsearch(['localhost:9200']) async def sync_to_elasticsearch(event): """Sync changes to Elasticsearch""" index = 'users' doc_id = event['entity_id'] if event['operation'] == 'DELETE': es.delete(index=index, id=doc_id) else: doc = { 'name': event['after']['properties']['name'], 'age': event['after']['properties']['age'], 'updated_at': event['timestamp'] } es.index(index=index, id=doc_id, document=doc) last_seen = "1970-01-01T00:00:00Z" async with client.connection() as conn: while True: result, _ = await conn.query( """ MATCH (e:ChangeLog) WHERE e.emitted_at > $since RETURN e.operation AS operation, e.entity_id AS entity_id, e.after AS after, e.emitted_at AS emitted_at ORDER BY emitted_at """, {"since": last_seen}, ) for row in result.rows: event = { "operation": row["operation"].raw_value, "entity_id": row["entity_id"].raw_value, "after": row["after"].raw_value, "timestamp": row["emitted_at"].raw_value, } await sync_to_elasticsearch(event) last_seen = row["emitted_at"].raw_value </code></pre></div> <h4 id="materialized-view-updates" class="position-relative d-flex align-items-center group"> Materialized View Updates <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="materialized-view-updates" aria-haspopup="dialog" aria-label="Share link: Materialized View Updates"> Share link </button> </h4>Maintain derived data: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Trigger to update materialized view CREATE TRIGGER update_user_stats ON CDC_STREAM user_changes WHEN operation = 'INSERT' EXECUTE GQL INSERT INTO stats (user_count, last_updated) VALUES ((SELECT COUNT(*) FROM Person), NOW()) ON CONFLICT (id) DO UPDATE SET user_count = EXCLUDED.user_count, last_updated = EXCLUDED.last_updated; </code></pre></div> <h3 id="cdc-performance" class="position-relative d-flex align-items-center group"> CDC Performance <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="cdc-performance" aria-haspopup="dialog" aria-label="Share link: CDC Performance"> Share link </button> </h3> <h4 id="event-batching" class="position-relative d-flex align-items-center group"> Event Batching <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="event-batching" aria-haspopup="dialog" aria-label="Share link: Event Batching"> Share link </button> </h4>Process events in batches for efficiency: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def process_batch(events): """Process events in batches""" # Group by operation inserts = [e for e in events if e['operation'] == 'INSERT'] updates = [e for e in events if e['operation'] == 'UPDATE'] deletes = [e for e in events if e['operation'] == 'DELETE'] # Batch insert if inserts: await batch_insert(inserts) # Batch update if updates: await batch_update(updates) # Batch delete if deletes: await batch_delete(deletes) last_seen = "1970-01-01T00:00:00Z" batch = [] async with client.connection() as conn: while True: result, _ = await conn.query( """ MATCH (e:ChangeLog) WHERE e.emitted_at > $since RETURN e.operation AS operation, e.entity_id AS entity_id, e.after AS after, e.emitted_at AS emitted_at ORDER BY emitted_at """, {"since": last_seen}, ) for row in result.rows: batch.append({ "operation": row["operation"].raw_value, "entity_id": row["entity_id"].raw_value, "after": row["after"].raw_value, }) last_seen = row["emitted_at"].raw_value if len(batch) >= 100: await process_batch(batch) batch = [] </code></pre></div> <h4 id="event-compression" class="position-relative d-flex align-items-center group"> Event Compression <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="event-compression" aria-haspopup="dialog" aria-label="Share link: Event Compression"> Share link </button> </h4>Reduce bandwidth with compression: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Enable compression for CDC stream ALTER CDC STREAM user_changes SET OPTIONS ( compression: 'gzip', compression_level: 6 ); </code></pre></div> <h4 id="checkpoint-management" class="position-relative d-flex align-items-center group"> Checkpoint Management <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="checkpoint-management" aria-haspopup="dialog" aria-label="Share link: Checkpoint Management"> Share link </button> </h4>Track processing progress: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">class CheckpointManager: def __init__(self, stream_name): self.stream_name = stream_name self.checkpoint = self.load_checkpoint() def load_checkpoint(self): # Load last processed event ID return redis.get(f"cdc:checkpoint:{self.stream_name}") def save_checkpoint(self, event_id): redis.set(f"cdc:checkpoint:{self.stream_name}", event_id) async def process_from_checkpoint(self): last_seen = self.checkpoint or "0" async with client.connection() as conn: while True: result, _ = await conn.query( """ MATCH (e:ChangeLog) WHERE e.event_id > $since RETURN e.event_id AS event_id, e.operation AS operation, e.after AS after ORDER BY e.event_id """, {"since": last_seen}, ) for row in result.rows: event = { "event_id": row["event_id"].raw_value, "operation": row["operation"].raw_value, "after": row["after"].raw_value, } await process_event(event) self.save_checkpoint(event["event_id"]) last_seen = event["event_id"] </code></pre></div> <h3 id="cdc-reliability" class="position-relative d-flex align-items-center group"> CDC Reliability <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="cdc-reliability" aria-haspopup="dialog" aria-label="Share link: CDC Reliability"> Share link </button> </h3> <h4 id="at-least-once-delivery" class="position-relative d-flex align-items-center group"> At-Least-Once Delivery <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="at-least-once-delivery" aria-haspopup="dialog" aria-label="Share link: At-Least-Once Delivery"> Share link </button> </h4>Events may be delivered multiple times: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Use idempotent operations async def process_event_idempotent(event): # Check if already processed if await is_processed(event['event_id']): return # Process event await handle_event(event) # Mark as processed await mark_processed(event['event_id']) </code></pre></div> <h4 id="exactly-once-semantics" class="position-relative d-flex align-items-center group"> Exactly-Once Semantics <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="exactly-once-semantics" aria-haspopup="dialog" aria-label="Share link: Exactly-Once Semantics"> Share link </button> </h4>Achieve exactly-once with deduplication: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def process_with_dedup(event): async with db.connection() as tx: await tx.begin() # Check for duplicate if await tx.exists("SELECT 1 FROM processed WHERE event_id = $1", event['event_id']): return # Process event await handle_event(event, tx) # Record processing await tx.execute( "INSERT INTO processed (event_id, timestamp) VALUES ($1, $2)", event['event_id'], datetime.now() ) </code></pre></div> <h4 id="error-handling" class="position-relative d-flex align-items-center group"> Error Handling <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="error-handling" aria-haspopup="dialog" aria-label="Share link: Error Handling"> Share link </button> </h4>Handle failures gracefully: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def robust_processor(event): max_retries = 3 retry_count = 0 while retry_count < max_retries: try: await process_event(event) return except TransientError as e: retry_count += 1 await asyncio.sleep(2 ** retry_count) # Exponential backoff except FatalError as e: # Send to dead letter queue await send_to_dlq(event, str(e)) return # Max retries exceeded await send_to_dlq(event, "Max retries exceeded") </code></pre></div> <h3 id="monitoring-cdc" class="position-relative d-flex align-items-center group"> Monitoring CDC <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="monitoring-cdc" aria-haspopup="dialog" aria-label="Share link: Monitoring CDC"> Share link </button> </h3> <h4 id="stream-metrics" class="position-relative d-flex align-items-center group"> Stream Metrics <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="stream-metrics" aria-haspopup="dialog" aria-label="Share link: Stream Metrics"> Share link </button> </h4>Monitor CDC stream health: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Query CDC stream metrics SELECT stream_name, events_published, events_consumed, lag_events, lag_seconds, last_event_timestamp FROM SYSTEM.cdc_streams; </code></pre></div> <h4 id="consumer-lag" class="position-relative d-flex align-items-center group"> Consumer Lag <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="consumer-lag" aria-haspopup="dialog" aria-label="Share link: Consumer Lag"> Share link </button> </h4>Track consumer processing lag: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">SELECT consumer_id, stream_name, last_processed_event, lag_events, throughput_events_per_sec FROM SYSTEM.cdc_consumers WHERE lag_events > 1000; </code></pre></div> <h4 id="alerting" class="position-relative d-flex align-items-center group"> Alerting <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="alerting" aria-haspopup="dialog" aria-label="Share link: Alerting"> Share link </button> </h4>Set up alerts for CDC issues: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-yaml" data-lang="yaml">alerts: - name: CDC lag too high condition: lag_events > 10000 action: notify_oncall - name: CDC consumer offline condition: last_heartbeat < NOW() - INTERVAL '5 minutes' action: page_oncall </code></pre></div> <h3 id="best-practices" class="position-relative d-flex align-items-center group"> Best Practices <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="best-practices" aria-haspopup="dialog" aria-label="Share link: Best Practices"> Share link </button> </h3> <h4 id="event-design" class="position-relative d-flex align-items-center group"> Event Design <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="event-design" aria-haspopup="dialog" aria-label="Share link: Event Design"> Share link </button> </h4>Design events carefully: <ul> <li>Include enough context to process independently</li> <li>Keep events immutable</li> <li>Use consistent event schemas</li> <li>Version event formats</li> <li>Include causality information</li> </ul> <h4 id="stream-management" class="position-relative d-flex align-items-center group"> Stream Management <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="stream-management" aria-haspopup="dialog" aria-label="Share link: Stream Management"> Share link </button> </h4>Manage streams effectively: <ul> <li>Use separate streams for different purposes</li> <li>Configure appropriate retention periods</li> <li>Monitor and scale stream capacity</li> <li>Implement backpressure handling</li> </ul> <h4 id="consumer-patterns" class="position-relative d-flex align-items-center group"> Consumer Patterns <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="consumer-patterns" aria-haspopup="dialog" aria-label="Share link: Consumer Patterns"> Share link </button> </h4>Implement robust consumers: <ul> <li>Process events idempotently</li> <li>Implement error handling and retries</li> <li>Track and checkpoint progress</li> <li>Monitor consumer lag</li> <li>Scale consumers horizontally</li> </ul> <h3 id="advanced-features" class="position-relative d-flex align-items-center group"> Advanced Features <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="advanced-features" aria-haspopup="dialog" aria-label="Share link: Advanced Features"> Share link </button> </h3> <h4 id="change-event-enrichment" class="position-relative d-flex align-items-center group"> Change Event Enrichment <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="change-event-enrichment" aria-haspopup="dialog" aria-label="Share link: Change Event Enrichment"> Share link </button> </h4>Add context to events: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Enrich events with related data CREATE CDC STREAM enriched_posts ON GRAPH social FOR LABELS (Post) WITH ENRICHMENT ( SELECT p.*, u.name AS author_name FROM Post p JOIN Person u ON p.author_id = u.id ); </code></pre></div> <h4 id="event-transformation" class="position-relative d-flex align-items-center group"> Event Transformation <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="event-transformation" aria-haspopup="dialog" aria-label="Share link: Event Transformation"> Share link </button> </h4>Transform events before delivery: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">CREATE CDC STREAM transformed_users ON GRAPH social FOR LABELS (Person) WITH TRANSFORMATION ( SELECT entity_id, after.properties.name AS user_name, UPPER(after.properties.email) AS email_upper, after.properties.age AS age ); </code></pre></div> <h4 id="multi-graph-cdc" class="position-relative d-flex align-items-center group"> Multi-Graph CDC <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="multi-graph-cdc" aria-haspopup="dialog" aria-label="Share link: Multi-Graph CDC"> Share link </button> </h4>Capture changes across multiple graphs: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">CREATE CDC STREAM all_changes ON GRAPHS (social, commerce, analytics) FOR ALL LABELS; </code></pre></div> <h3 id="related-topics" class="position-relative d-flex align-items-center group"> Related Topics <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="related-topics" aria-haspopup="dialog" aria-label="Share link: Related Topics"> Share link </button> </h3><ul> <li><a href="/tags/streaming" >Event Streaming</a> - Event streaming architectures</li> <li><a href="/tags/pubsub" >Pub/Sub</a> - Publish-subscribe messaging</li> <li><a href="/tags/events" >Events</a> - Event-driven architecture</li> <li><a href="/tags/replication" >Replication</a> - Data replication strategies</li> </ul> <h3 id="learn-more" class="position-relative d-flex align-items-center group"> Learn More <button type="button" class="h-share btn btn-link p-0 text-decoration-none link-secondary opacity-50 hover-opacity-100 transition-all ms-1" data-share-target="learn-more" aria-haspopup="dialog" aria-label="Share link: Learn More"> Share link </button> </h3><ul> <li><a href="https://www.confluent.io/learn/change-data-capture/" aria-label="Change Data Capture Patterns – opens in new window" target="_blank" rel="noopener noreferrer" >Change Data Capture Patterns ↗ </a> </li> <li><a href="https://debezium.io/" aria-label="Debezium CDC Platform – opens in new window" target="_blank" rel="noopener noreferrer" >Debezium CDC Platform ↗ </a> </li> <li><a href="https://martinfowler.com/articles/201701-event-driven.html" aria-label="Event-Driven Architecture – opens in new window" target="_blank" rel="noopener noreferrer" >Event-Driven Architecture ↗ </a> </li> </ul>

Popular

Related Articles

Deployment and Integration Guide

Fraud and Anomaly Detection