Event Streaming

<h2 id="event-streaming" class="position-relative d-flex align-items-center group"> Event Streaming <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-streaming" aria-haspopup="dialog" aria-label="Share link: Event Streaming"> 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>Event streaming is a paradigm for capturing, storing, and processing continuous flows of events in real-time. Geode’s event streaming capabilities enable building reactive, scalable applications that respond to changes as they happen, powering use cases from real-time analytics to complex event processing. <h3 id="event-streaming-fundamentals" class="position-relative d-flex align-items-center group"> Event Streaming Fundamentals <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-streaming-fundamentals" aria-haspopup="dialog" aria-label="Share link: Event Streaming Fundamentals"> Share link </button> </h3> <h4 id="what-is-event-streaming" class="position-relative d-flex align-items-center group"> What is Event Streaming? <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-event-streaming" aria-haspopup="dialog" aria-label="Share link: What is Event Streaming?"> Share link </button> </h4>Event streaming treats data as a continuous flow of events: Events - Immutable facts about what happened Streams - Ordered sequences of events Producers - Systems that emit events Consumers - Systems that process events <h4 id="event-stream-characteristics" class="position-relative d-flex align-items-center group"> Event Stream Characteristics <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-stream-characteristics" aria-haspopup="dialog" aria-label="Share link: Event Stream Characteristics"> Share link </button> </h4>Key properties of event streams: <ul> <li>Ordered - Events maintain temporal ordering</li> <li>Immutable - Events cannot be changed once written</li> <li>Append-Only - New events added to stream end</li> <li>Retained - Events stored for configurable duration</li> <li>Replayable - Consumers can replay historical events</li> </ul> <h3 id="geode-event-streaming" class="position-relative d-flex align-items-center group"> Geode Event Streaming <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-event-streaming" aria-haspopup="dialog" aria-label="Share link: Geode Event Streaming"> Share link </button> </h3> <h4 id="creating-event-streams" class="position-relative d-flex align-items-center group"> Creating Event 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="creating-event-streams" aria-haspopup="dialog" aria-label="Share link: Creating Event Streams"> Share link </button> </h4>Define streams to capture graph events: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Create stream for user activities CREATE STREAM user_activities AS SELECT event_time, user_id, action_type, properties FROM user_actions EMIT CHANGES; -- Create stream with windowing CREATE STREAM page_views AS SELECT TUMBLING_WINDOW(5 MINUTES) AS window, page_url, COUNT(*) AS view_count FROM page_view_events GROUP BY page_url EMIT CHANGES; </code></pre></div> <h4 id="publishing-events" class="position-relative d-flex align-items-center group"> Publishing 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="publishing-events" aria-haspopup="dialog" aria-label="Share link: Publishing Events"> Share link </button> </h4>Publish events to streams: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">from geode_client import Client async def publish_user_action(client, user_id, action): """Publish user action event""" event = { "event_time": datetime.now().isoformat(), "user_id": user_id, "action_type": action, "properties": { "page": "/products", "duration_ms": 1500 } } await client.publish_event("user_activities", event) </code></pre></div> <h4 id="consuming-events" class="position-relative d-flex align-items-center group"> Consuming 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-events" aria-haspopup="dialog" aria-label="Share link: Consuming Events"> Share link </button> </h4>Subscribe to stream events: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def process_user_activities(client): """Process user activity stream""" async with client.subscribe_stream("user_activities") as stream: async for event in stream: print(f"User {event['user_id']} performed {event['action_type']}") # React to specific actions if event['action_type'] == 'purchase': await handle_purchase(event) elif event['action_type'] == 'signup': await send_welcome_email(event['user_id']) </code></pre></div> <h3 id="stream-processing-patterns" class="position-relative d-flex align-items-center group"> Stream Processing 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="stream-processing-patterns" aria-haspopup="dialog" aria-label="Share link: Stream Processing Patterns"> Share link </button> </h3> <h4 id="filter-streams" class="position-relative d-flex align-items-center group"> Filter 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="filter-streams" aria-haspopup="dialog" aria-label="Share link: Filter Streams"> Share link </button> </h4>Filter events based on criteria: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Stream of high-value purchases CREATE STREAM high_value_purchases AS SELECT * FROM purchases WHERE amount > 1000 EMIT CHANGES; </code></pre></div> <h4 id="transform-streams" class="position-relative d-flex align-items-center group"> Transform 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="transform-streams" aria-haspopup="dialog" aria-label="Share link: Transform Streams"> Share link </button> </h4>Transform event data: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Enrich events with user data CREATE STREAM enriched_purchases AS SELECT p.purchase_id, p.amount, u.name AS customer_name, u.tier AS customer_tier FROM purchases p JOIN users u ON p.user_id = u.id EMIT CHANGES; </code></pre></div> <h4 id="aggregate-streams" class="position-relative d-flex align-items-center group"> Aggregate 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="aggregate-streams" aria-haspopup="dialog" aria-label="Share link: Aggregate Streams"> Share link </button> </h4>Compute aggregations over time: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Real-time revenue by product CREATE STREAM revenue_by_product AS SELECT product_id, SUM(amount) AS total_revenue, COUNT(*) AS purchase_count FROM purchases GROUP BY product_id WINDOW TUMBLING (SIZE 1 HOUR) EMIT CHANGES; </code></pre></div> <h4 id="join-streams" class="position-relative d-flex align-items-center group"> Join 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="join-streams" aria-haspopup="dialog" aria-label="Share link: Join Streams"> Share link </button> </h4>Combine multiple streams: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Join page views with purchases CREATE STREAM conversion_events AS SELECT v.user_id, v.page_url, p.purchase_id, p.amount, (p.event_time - v.event_time) AS time_to_purchase FROM page_views v JOIN purchases p ON v.user_id = p.user_id WITHIN 1 HOUR EMIT CHANGES; </code></pre></div> <h3 id="windowing-operations" class="position-relative d-flex align-items-center group"> Windowing Operations <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="windowing-operations" aria-haspopup="dialog" aria-label="Share link: Windowing Operations"> Share link </button> </h3> <h4 id="tumbling-windows" class="position-relative d-flex align-items-center group"> Tumbling Windows <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="tumbling-windows" aria-haspopup="dialog" aria-label="Share link: Tumbling Windows"> Share link </button> </h4>Fixed-size, non-overlapping windows: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Count events per 5-minute window CREATE STREAM event_counts AS SELECT TUMBLING_WINDOW(5 MINUTES) AS window, event_type, COUNT(*) AS count FROM events GROUP BY event_type EMIT CHANGES; </code></pre></div>Tumbling windows: <ul> <li>Fixed size (e.g., 5 minutes)</li> <li>No overlap between windows</li> <li>Each event belongs to exactly one window</li> </ul> <h4 id="hopping-windows" class="position-relative d-flex align-items-center group"> Hopping Windows <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="hopping-windows" aria-haspopup="dialog" aria-label="Share link: Hopping Windows"> Share link </button> </h4>Fixed-size windows with overlap: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Sliding 10-minute window, advancing every 1 minute CREATE STREAM sliding_averages AS SELECT HOPPING_WINDOW(SIZE 10 MINUTES, ADVANCE 1 MINUTE) AS window, AVG(response_time_ms) AS avg_response_time FROM api_requests EMIT CHANGES; </code></pre></div>Hopping windows: <ul> <li>Fixed size with configurable advance</li> <li>Windows overlap</li> <li>Each event may appear in multiple windows</li> </ul> <h4 id="session-windows" class="position-relative d-flex align-items-center group"> Session Windows <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="session-windows" aria-haspopup="dialog" aria-label="Share link: Session Windows"> Share link </button> </h4>Dynamic windows based on inactivity: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- User sessions (15 minute timeout) CREATE STREAM user_sessions AS SELECT SESSION_WINDOW(15 MINUTES) AS session, user_id, COUNT(*) AS actions_in_session, COLLECT_LIST(action_type) AS actions FROM user_activities GROUP BY user_id EMIT CHANGES; </code></pre></div>Session windows: <ul> <li>Variable size based on activity</li> <li>Gap threshold defines session boundary</li> <li>Useful for user behavior analysis</li> </ul> <h3 id="stream-processing" class="position-relative d-flex align-items-center group"> Stream Processing <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-processing" aria-haspopup="dialog" aria-label="Share link: Stream Processing"> Share link </button> </h3> <h4 id="stateless-processing" class="position-relative d-flex align-items-center group"> Stateless Processing <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="stateless-processing" aria-haspopup="dialog" aria-label="Share link: Stateless Processing"> Share link </button> </h4>Process each event independently: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def stateless_processor(event): """Process event without state""" # Validate event if not validate_event(event): return # Transform event transformed = { "id": event['id'], "timestamp": event['timestamp'], "value": event['value'] * 1.5 # Apply transformation } # Publish to output stream await publish(transformed) </code></pre></div> <h4 id="stateful-processing" class="position-relative d-flex align-items-center group"> Stateful Processing <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="stateful-processing" aria-haspopup="dialog" aria-label="Share link: Stateful Processing"> Share link </button> </h4>Maintain state across events: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">class StatefulProcessor: def __init__(self): self.state = {} async def process(self, event): """Process event with state""" user_id = event['user_id'] # Retrieve user state if user_id not in self.state: self.state[user_id] = { "total_purchases": 0, "total_spent": 0 } # Update state self.state[user_id]["total_purchases"] += 1 self.state[user_id]["total_spent"] += event['amount'] # Check for milestone if self.state[user_id]["total_spent"] > 10000: await notify_vip_status(user_id) </code></pre></div> <h4 id="complex-event-processing-cep" class="position-relative d-flex align-items-center group"> Complex Event Processing (CEP) <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="complex-event-processing-cep" aria-haspopup="dialog" aria-label="Share link: Complex Event Processing (CEP)"> Share link </button> </h4>Detect patterns across multiple events: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Detect suspicious behavior pattern CREATE STREAM suspicious_activity AS SELECT user_id, COLLECT_LIST(action_type) AS pattern FROM user_activities MATCH_RECOGNIZE ( PARTITION BY user_id ORDER BY event_time MEASURES FIRST(A.event_time) AS start_time, LAST(B.event_time) AS end_time PATTERN (A B+ C) DEFINE A AS A.action_type = 'login_failure', B AS B.action_type = 'login_failure', C AS C.action_type = 'login_success' AND C.ip_address != A.ip_address ) EMIT CHANGES; </code></pre></div> <h3 id="event-sourcing" class="position-relative d-flex align-items-center group"> Event Sourcing <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-sourcing" aria-haspopup="dialog" aria-label="Share link: Event Sourcing"> Share link </button> </h3> <h4 id="event-store" class="position-relative d-flex align-items-center group"> Event Store <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-store" aria-haspopup="dialog" aria-label="Share link: Event Store"> Share link </button> </h4>Store all events as source of truth: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Event-sourced account balance CREATE EVENT_STORE account_transactions ( account_id STRING, event_type STRING, amount DECIMAL, balance DECIMAL, event_time TIMESTAMP ) PARTITION BY account_id; -- Publish transaction events INSERT INTO account_transactions VALUES ( 'acct123', 'deposit', 500.00, 1500.00, NOW() ); </code></pre></div> <h4 id="event-replay" class="position-relative d-flex align-items-center group"> Event Replay <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-replay" aria-haspopup="dialog" aria-label="Share link: Event Replay"> Share link </button> </h4>Rebuild state from events: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def rebuild_account_state(account_id): """Rebuild account state from events""" balance = 0 # Replay all events events = await client.query( "SELECT * FROM account_transactions WHERE account_id = $1 ORDER BY event_time", [account_id] ) for event in events: if event['event_type'] == 'deposit': balance += event['amount'] elif event['event_type'] == 'withdrawal': balance -= event['amount'] return balance </code></pre></div> <h4 id="snapshots" class="position-relative d-flex align-items-center group"> Snapshots <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="snapshots" aria-haspopup="dialog" aria-label="Share link: Snapshots"> Share link </button> </h4>Optimize replay with periodic snapshots: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">-- Create snapshot table CREATE TABLE account_snapshots ( account_id STRING PRIMARY KEY, balance DECIMAL, snapshot_time TIMESTAMP, last_event_id STRING ); -- Periodically save snapshots INSERT INTO account_snapshots SELECT account_id, SUM(CASE WHEN event_type = 'deposit' THEN amount WHEN event_type = 'withdrawal' THEN -amount END) AS balance, NOW() AS snapshot_time, MAX(event_id) AS last_event_id FROM account_transactions GROUP BY account_id; </code></pre></div> <h3 id="stream-integration" class="position-relative d-flex align-items-center group"> Stream Integration <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-integration" aria-haspopup="dialog" aria-label="Share link: Stream Integration"> Share link </button> </h3> <h4 id="kafka-integration" class="position-relative d-flex align-items-center group"> Kafka Integration <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="kafka-integration" aria-haspopup="dialog" aria-label="Share link: Kafka Integration"> Share link </button> </h4>Integrate with Apache Kafka: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">from kafka import KafkaProducer, KafkaConsumer # Publish Geode events to Kafka async def publish_to_kafka(geode_event): producer = KafkaProducer( bootstrap_servers=['localhost:9092'], value_serializer=lambda v: json.dumps(v).encode('utf-8') ) producer.send('user-activities', geode_event) producer.flush() # Consume Kafka events to Geode async def consume_from_kafka(): consumer = KafkaConsumer( 'user-activities', bootstrap_servers=['localhost:9092'], value_deserializer=lambda m: json.loads(m.decode('utf-8')) ) for message in consumer: await geode_client.publish_event('user_activities', message.value) </code></pre></div> <h4 id="pulsar-integration" class="position-relative d-flex align-items-center group"> Pulsar Integration <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="pulsar-integration" aria-haspopup="dialog" aria-label="Share link: Pulsar Integration"> Share link </button> </h4>Connect with Apache Pulsar: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">import pulsar async def pulsar_to_geode(): client = pulsar.Client('pulsar://localhost:6650') consumer = client.subscribe('user-events', 'geode-subscription') while True: msg = consumer.receive() try: event = json.loads(msg.data()) await geode_client.publish_event('user_activities', event) consumer.acknowledge(msg) except Exception as e: consumer.negative_acknowledge(msg) </code></pre></div> <h3 id="performance-optimization" class="position-relative d-flex align-items-center group"> Performance Optimization <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="performance-optimization" aria-haspopup="dialog" aria-label="Share link: Performance Optimization"> Share link </button> </h3> <h4 id="batching" class="position-relative d-flex align-items-center group"> 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="batching" aria-haspopup="dialog" aria-label="Share link: Batching"> Share link </button> </h4>Improve throughput with batching: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def batch_processor(stream_name, batch_size=100): """Process events in batches""" batch = [] async with client.subscribe_stream(stream_name) as stream: async for event in stream: batch.append(event) if len(batch) >= batch_size: await process_batch(batch) batch = [] </code></pre></div> <h4 id="parallel-processing" class="position-relative d-flex align-items-center group"> Parallel Processing <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="parallel-processing" aria-haspopup="dialog" aria-label="Share link: Parallel Processing"> Share link </button> </h4>Scale with parallel consumers: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">import asyncio async def parallel_consumer(stream_name, num_workers=4): """Process stream with multiple workers""" async def worker(worker_id, partition): async with client.subscribe_stream( stream_name, partition=partition ) as stream: async for event in stream: await process_event(event, worker_id) # Start workers for each partition tasks = [ worker(i, i) for i in range(num_workers) ] await asyncio.gather(*tasks) </code></pre></div> <h4 id="backpressure" class="position-relative d-flex align-items-center group"> Backpressure <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="backpressure" aria-haspopup="dialog" aria-label="Share link: Backpressure"> Share link </button> </h4>Handle slow consumers: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def backpressure_handler(stream_name): """Handle backpressure gracefully""" buffer = asyncio.Queue(maxsize=1000) async def producer(): async with client.subscribe_stream(stream_name) as stream: async for event in stream: await buffer.put(event) async def consumer(): while True: event = await buffer.get() await process_event(event) buffer.task_done() await asyncio.gather(producer(), consumer()) </code></pre></div> <h3 id="monitoring-streams" class="position-relative d-flex align-items-center group"> Monitoring 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="monitoring-streams" aria-haspopup="dialog" aria-label="Share link: Monitoring Streams"> 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>Track stream health: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">SELECT stream_name, events_per_second, bytes_per_second, consumer_lag, oldest_event_age FROM SYSTEM.streams WHERE consumer_lag > 10000; </code></pre></div> <h4 id="consumer-monitoring" class="position-relative d-flex align-items-center group"> Consumer Monitoring <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-monitoring" aria-haspopup="dialog" aria-label="Share link: Consumer Monitoring"> Share link </button> </h4>Monitor consumer performance: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-gql" data-lang="gql">SELECT consumer_id, stream_name, throughput, error_rate, last_heartbeat FROM SYSTEM.stream_consumers WHERE error_rate > 0.01; </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><ul> <li>Use consistent event schemas</li> <li>Include sufficient context</li> <li>Version events for evolution</li> <li>Keep events immutable</li> <li>Use timestamps for ordering</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><ul> <li>Choose appropriate retention</li> <li>Partition for scalability</li> <li>Monitor and alert on lag</li> <li>Implement backpressure handling</li> <li>Test failure scenarios</li> </ul> <h4 id="processing-guarantees" class="position-relative d-flex align-items-center group"> Processing Guarantees <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="processing-guarantees" aria-haspopup="dialog" aria-label="Share link: Processing Guarantees"> Share link </button> </h4><ul> <li>Implement idempotent processing</li> <li>Use checkpointing for progress</li> <li>Handle exactly-once semantics</li> <li>Design for eventual consistency</li> <li>Monitor processing lag</li> </ul> <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/cdc" >CDC</a> - Change data capture</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/real-time" >Real-Time</a> - Real-time processing</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://engineering.linkedin.com/distributed-systems/log-what-every-software-engineer-should-know-about-real-time-datas-unifying" aria-label="The Log: What every software engineer should know – opens in new window" target="_blank" rel="noopener noreferrer" >The Log: What every software engineer should know ↗ </a> </li> <li><a href="https://www.confluent.io/blog/event-streaming-platform-1" aria-label="Event Streaming Patterns – opens in new window" target="_blank" rel="noopener noreferrer" >Event Streaming Patterns ↗ </a> </li> <li><a href="https://www.confluent.io/resources/kafka-the-definitive-guide/" aria-label="Kafka: The Definitive Guide – opens in new window" target="_blank" rel="noopener noreferrer" >Kafka: The Definitive Guide ↗ </a> </li> </ul>

Popular

Related Articles