ETL Pipelines & Data Integration

<h2 id="etl-pipelines--data-integration" class="position-relative d-flex align-items-center group"> ETL Pipelines &amp; Data 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="etl-pipelines--data-integration" aria-haspopup="dialog" aria-label="Share link: ETL Pipelines &amp; Data Integration"> 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>ETL (Extract, Transform, Load) pipelines are critical for integrating data from diverse sources into your Geode graph database. Whether migrating from a relational database, aggregating data from multiple APIs, or streaming real-time events, well-designed ETL processes ensure data arrives in Geode accurately, efficiently, and reliably. <h3 id="understanding-etl-in-graph-context" class="position-relative d-flex align-items-center group"> Understanding ETL in Graph Context <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-etl-in-graph-context" aria-haspopup="dialog" aria-label="Share link: Understanding ETL in Graph Context"> Share link </button> </h3>Traditional ETL processes are designed for tabular data moving between relational databases. Graph ETL requires additional considerations: <ul> <li>Relationship Discovery: Identifying connections between entities during transformation</li> <li>Node Deduplication: Ensuring the same entity isn’t created multiple times</li> <li>Edge Creation: Linking nodes based on foreign key relationships or business logic</li> <li>Graph Schema Mapping: Converting normalized tables to denormalized graph properties</li> <li>Traversal Optimization: Organizing data for efficient graph queries</li> </ul> <h3 id="the-etl-process" class="position-relative d-flex align-items-center group"> The ETL Process <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="the-etl-process" aria-haspopup="dialog" aria-label="Share link: The ETL Process"> Share link </button> </h3> <h4 id="extract-pulling-data-from-sources" class="position-relative d-flex align-items-center group"> Extract: Pulling Data from Sources <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="extract-pulling-data-from-sources" aria-haspopup="dialog" aria-label="Share link: Extract: Pulling Data from Sources"> Share link </button> </h4>Extract phase reads data from source systems without modifying the original data. Common Sources: <ul> <li>Relational databases (PostgreSQL, MySQL, Oracle)</li> <li>NoSQL databases (MongoDB, Cassandra)</li> <li>REST APIs</li> <li>CSV/JSON files</li> <li>Message queues (Kafka, RabbitMQ)</li> <li>Data warehouses (Snowflake, BigQuery)</li> </ul> <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Python - Extract from PostgreSQL import asyncpg from geode_client import Client async def extract_from_postgres(pg_pool): """Extract users and orders from PostgreSQL.""" async with pg_pool.acquire() as conn: # Extract users users = await conn.fetch(""" SELECT user_id, email, first_name, last_name, created_at, country, status FROM users WHERE updated_at > $1 """, last_sync_time) # Extract orders orders = await conn.fetch(""" SELECT order_id, user_id, total_amount, currency, status, created_at FROM orders WHERE updated_at > $1 """, last_sync_time) # Extract order items (relationships) order_items = await conn.fetch(""" SELECT order_id, product_id, quantity, unit_price FROM order_items WHERE updated_at > $1 """, last_sync_time) return { "users": users, "orders": orders, "order_items": order_items } </code></pre></div>Go example with database/sql: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-go" data-lang="go">// Go - Extract from MySQL package main import ( "context" "database/sql" "time" _ "github.com/go-sql-driver/mysql" ) type User struct { ID string Email string FirstName string LastName string CreatedAt time.Time } func extractUsers(ctx context.Context, db *sql.DB, since time.Time) ([]User, error) { rows, err := db.QueryContext(ctx, ` SELECT user_id, email, first_name, last_name, created_at FROM users WHERE updated_at > ? ORDER BY updated_at ASC `, since) if err != nil { return nil, err } defer rows.Close() var users []User for rows.Next() { var u User if err := rows.Scan(&u.ID, &u.Email, &u.FirstName, &u.LastName, &u.CreatedAt); err != nil { return nil, err } users = append(users, u) } return users, rows.Err() } </code></pre></div> <h4 id="transform-converting-to-graph-structure" class="position-relative d-flex align-items-center group"> Transform: Converting to Graph 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="transform-converting-to-graph-structure" aria-haspopup="dialog" aria-label="Share link: Transform: Converting to Graph Structure"> Share link </button> </h4>Transform phase converts source data into nodes, edges, and properties suitable for Geode. <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Python - Transform relational data to graph structure from typing import Dict, List, Any from datetime import datetime def transform_user(raw_user: Dict[str, Any]) -> Dict[str, Any]: """Transform PostgreSQL user record to Geode Person node.""" return { "labels": ["Person", "Customer"], "properties": { "id": raw_user['user_id'], "email": raw_user['email'].lower().strip(), "name": f"{raw_user['first_name']} {raw_user['last_name']}", "first_name": raw_user['first_name'], "last_name": raw_user['last_name'], "country": raw_user['country'], "status": raw_user['status'], "created_at": raw_user['created_at'].isoformat(), "imported_at": datetime.utcnow().isoformat() } } def transform_order(raw_order: Dict[str, Any]) -> Dict[str, Any]: """Transform order record to Geode Order node.""" return { "labels": ["Order"], "properties": { "id": raw_order['order_id'], "total": float(raw_order['total_amount']), "currency": raw_order['currency'], "status": raw_order['status'], "created_at": raw_order['created_at'].isoformat() } } def transform_order_relationship(raw_order: Dict[str, Any]) -> Dict[str, Any]: """Create PLACED edge between Person and Order.""" return { "type": "PLACED", "from_id": raw_order['user_id'], "from_label": "Person", "to_id": raw_order['order_id'], "to_label": "Order", "properties": { "placed_at": raw_order['created_at'].isoformat() } } def transform_order_item_relationship(raw_item: Dict[str, Any]) -> Dict[str, Any]: """Create CONTAINS edge between Order and Product.""" return { "type": "CONTAINS", "from_id": raw_item['order_id'], "from_label": "Order", "to_id": raw_item['product_id'], "to_label": "Product", "properties": { "quantity": raw_item['quantity'], "unit_price": float(raw_item['unit_price']) } } </code></pre></div>Rust example with type safety: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-rust" data-lang="rust">// Rust - Transform with type-safe builders use serde::{Deserialize, Serialize}; use chrono::{DateTime, Utc}; #[derive(Deserialize)] struct RawUser { user_id: String, email: String, first_name: String, last_name: String, created_at: DateTime<Utc>, } #[derive(Serialize)] struct PersonNode { labels: Vec<String>, properties: PersonProperties, } #[derive(Serialize)] struct PersonProperties { id: String, email: String, name: String, created_at: String, } fn transform_user(raw: RawUser) -> PersonNode { PersonNode { labels: vec!["Person".into(), "Customer".into()], properties: PersonProperties { id: raw.user_id, email: raw.email.to_lowercase().trim().to_string(), name: format!("{} {}", raw.first_name, raw.last_name), created_at: raw.created_at.to_rfc3339(), }, } } </code></pre></div> <h4 id="load-inserting-into-geode" class="position-relative d-flex align-items-center group"> Load: Inserting into Geode <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="load-inserting-into-geode" aria-haspopup="dialog" aria-label="Share link: Load: Inserting into Geode"> Share link </button> </h4>Load phase efficiently inserts transformed data into Geode with proper error handling and retry logic. <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Python - Batch load with transaction support from geode_client import Client import asyncio async def load_to_geode(client: Client, nodes: List[Dict], edges: List[Dict]): """Load nodes and edges to Geode in batches.""" batch_size = 1000 async with client.connection() as tx: await tx.begin() # Load nodes in batches for i in range(0, len(nodes), batch_size): batch = nodes[i:i + batch_size] # Use MERGE to handle duplicates idempotently await tx.execute( """UNWIND $batch AS row MERGE (n:Person {id: row.id}) SET n.email = row.email, n.name = row.name, n.created_at = row.created_at, n.updated_at = NOW() """, {"batch": [node['properties'] for node in batch]} ) print(f"Loaded {i + len(batch)}/{len(nodes)} nodes") # Load edges after all nodes exist for i in range(0, len(edges), batch_size): batch = edges[i:i + batch_size] await tx.execute( """UNWIND $batch AS row MATCH (from:Person {id: row.from_id}) MATCH (to:Order {id: row.to_id}) MERGE (from)-[r:PLACED]->(to) SET r.placed_at = row.placed_at """, {"batch": [edge['properties'] for edge in batch]} ) print(f"Loaded {i + len(batch)}/{len(edges)} edges") await tx.commit() </code></pre></div> <h3 id="incremental-vs-full-load" class="position-relative d-flex align-items-center group"> Incremental vs Full Load <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="incremental-vs-full-load" aria-haspopup="dialog" aria-label="Share link: Incremental vs Full Load"> Share link </button> </h3> <h4 id="full-load" class="position-relative d-flex align-items-center group"> Full Load <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="full-load" aria-haspopup="dialog" aria-label="Share link: Full Load"> Share link </button> </h4>Complete replacement of data, typically used for initial migration: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">async def full_load_etl(pg_pool, geode_client): """Full load: replace all data.""" # Extract all data data = await extract_all_from_postgres(pg_pool) # Transform nodes = [transform_user(user) for user in data['users']] edges = [transform_order_relationship(order) for order in data['orders']] # Clear existing data (optional) await geode_client.execute("MATCH (n:Person) DELETE n") # Load all await load_to_geode(geode_client, nodes, edges) </code></pre></div> <h4 id="incremental-load-change-data-capture" class="position-relative d-flex align-items-center group"> Incremental Load (Change Data Capture) <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="incremental-load-change-data-capture" aria-haspopup="dialog" aria-label="Share link: Incremental Load (Change Data Capture)"> Share link </button> </h4>Only load changed records since last sync: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python">import json from datetime import datetime, timedelta from pathlib import Path class IncrementalETL: def __init__(self, state_file: str = "etl_state.json"): self.state_file = Path(state_file) def get_last_sync_time(self) -> datetime: """Get timestamp of last successful sync.""" if self.state_file.exists(): state = json.loads(self.state_file.read_text()) return datetime.fromisoformat(state['last_sync']) return datetime.min # First run, sync all data def save_sync_time(self, timestamp: datetime): """Save successful sync timestamp.""" self.state_file.write_text(json.dumps({ 'last_sync': timestamp.isoformat() })) async def incremental_load(self, pg_pool, geode_client): """Load only changed records since last sync.""" sync_start = datetime.utcnow() last_sync = self.get_last_sync_time() print(f"Syncing changes since {last_sync}") # Extract only changed records data = await extract_from_postgres(pg_pool, since=last_sync) print(f"Extracted {len(data['users'])} users, {len(data['orders'])} orders") # Transform nodes = [transform_user(user) for user in data['users']] edges = [transform_order_relationship(order) for order in data['orders']] # Load (MERGE handles inserts and updates) await load_to_geode(geode_client, nodes, edges) # Save state only after successful load self.save_sync_time(sync_start) print(f"Sync completed successfully") </code></pre></div> <h3 id="orchestration-with-apache-airflow" class="position-relative d-flex align-items-center group"> Orchestration with Apache Airflow <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="orchestration-with-apache-airflow" aria-haspopup="dialog" aria-label="Share link: Orchestration with Apache Airflow"> Share link </button> </h3>Apache Airflow is industry-standard for scheduling and monitoring ETL pipelines. <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Airflow DAG for daily ETL from airflow import DAG from airflow.operators.python import PythonOperator from airflow.utils.dates import days_ago from datetime import timedelta import asyncio default_args = { 'owner': 'data-team', 'depends_on_past': False, 'email': ['[email protected]'], 'email_on_failure': True, 'email_on_retry': False, 'retries': 3, 'retry_delay': timedelta(minutes=5), } dag = DAG( 'geode_etl_daily', default_args=default_args, description='Daily ETL from PostgreSQL to Geode', schedule_interval='0 2 * * *', # Run at 2 AM daily start_date=days_ago(1), tags=['etl', 'geode'], catchup=False ) def extract_task(**context): """Extract data from source.""" data = asyncio.run(extract_from_postgres(pg_pool)) context['task_instance'].xcom_push(key='extracted_data', value=data) return f"Extracted {len(data['users'])} users" def transform_task(**context): """Transform extracted data.""" data = context['task_instance'].xcom_pull(key='extracted_data') nodes = [transform_user(u) for u in data['users']] edges = [transform_order_relationship(o) for o in data['orders']] context['task_instance'].xcom_push(key='transformed_data', value={'nodes': nodes, 'edges': edges}) return f"Transformed {len(nodes)} nodes, {len(edges)} edges" def load_task(**context): """Load data to Geode.""" data = context['task_instance'].xcom_pull(key='transformed_data') asyncio.run(load_to_geode(geode_client, data['nodes'], data['edges'])) return f"Loaded successfully" def validate_task(**context): """Validate data quality after load.""" result = asyncio.run(geode_client.execute(""" MATCH (p:Person) WHERE p.email IS NULL OR p.email !~ '^[^@]+@[^@]+\.[^@]+$' RETURN count(p) AS invalid_count """)) invalid = result.bindings[0]['invalid_count'] if invalid > 0: raise ValueError(f"Found {invalid} persons with invalid emails") return f"Validation passed" # Define task dependencies extract = PythonOperator(task_id='extract', python_callable=extract_task, dag=dag) transform = PythonOperator(task_id='transform', python_callable=transform_task, dag=dag) load = PythonOperator(task_id='load', python_callable=load_task, dag=dag) validate = PythonOperator(task_id='validate', python_callable=validate_task, dag=dag) # Pipeline: extract -> transform -> load -> validate extract >> transform >> load >> validate </code></pre></div> <h3 id="real-time-streaming-etl" class="position-relative d-flex align-items-center group"> Real-Time Streaming ETL <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="real-time-streaming-etl" aria-haspopup="dialog" aria-label="Share link: Real-Time Streaming ETL"> Share link </button> </h3>For real-time data integration, use streaming ETL with Kafka or similar: <div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Python - Streaming ETL with Kafka from kafka import KafkaConsumer from geode_client import Client import json async def streaming_etl(): """Process events from Kafka in real-time.""" consumer = KafkaConsumer( 'user-events', bootstrap_servers=['localhost:9092'], value_deserializer=lambda m: json.loads(m.decode('utf-8')), group_id='geode-etl', auto_offset_reset='earliest' ) client = Client(host="localhost", port=3141) async with client.connection() as conn: for message in consumer: event = message.value try: if event['type'] == 'user.created': node = transform_user(event['data']) await conn.execute( """MERGE (p:Person {id: $id}) SET p += $properties""", {"id": node['properties']['id'], "properties": node['properties']} ) elif event['type'] == 'order.placed': order_node = transform_order(event['data']) relationship = transform_order_relationship(event['data']) await conn.execute( """MERGE (o:Order {id: $order_id}) SET o += $order_props WITH o MATCH (p:Person {id: $user_id}) MERGE (p)-[r:PLACED]->(o) SET r.placed_at = $placed_at""", { "order_id": order_node['properties']['id'], "order_props": order_node['properties'], "user_id": relationship['from_id'], "placed_at": relationship['properties']['placed_at'] } ) # Commit offset after successful processing consumer.commit() except Exception as e: print(f"Error processing event: {e}") # Don't commit offset, retry on next poll </code></pre></div> <h3 id="error-handling-and-recovery" class="position-relative d-flex align-items-center group"> Error Handling and Recovery <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-and-recovery" aria-haspopup="dialog" aria-label="Share link: Error Handling and Recovery"> Share link </button> </h3><div class="highlight"><pre tabindex="0" class="chroma"><code class="language-python" data-lang="python"># Python - Robust error handling import logging from tenacity import retry, stop_after_attempt, wait_exponential logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10) ) async def load_batch_with_retry(client, batch, batch_num): """Load batch with exponential backoff retry.""" try: await client.execute( "UNWIND $batch AS row\n" "MERGE (n:Person {id: row.id})\n" "SET n += row.properties", {"batch": batch} ) logger.info(f"Batch {batch_num} loaded successfully") except Exception as e: logger.error(f"Batch {batch_num} failed: {e}") raise async def resilient_etl(pg_pool, geode_client): """ETL with comprehensive error handling.""" failed_batches = [] try: data = await extract_from_postgres(pg_pool) except Exception as e: logger.error(f"Extract failed: {e}") send_alert("ETL Extract Failure", str(e)) raise nodes = [transform_user(u) for u in data['users']] # Process batches with individual error handling batch_size = 1000 for i in range(0, len(nodes), batch_size): batch = nodes[i:i + batch_size] batch_num = i // batch_size + 1 try: await load_batch_with_retry(geode_client, batch, batch_num) except Exception as e: logger.error(f"Batch {batch_num} failed after retries: {e}") failed_batches.append((batch_num, batch)) # Continue processing other batches # Report failures if failed_batches: logger.warning(f"{len(failed_batches)} batches failed") send_alert(f"ETL Partial Failure", f"{len(failed_batches)} batches failed") else: logger.info("ETL completed successfully") </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><ol> <li>Idempotent Operations: Use MERGE instead of INSERT to handle re-runs safely</li> <li>Incremental Loading: Only process changed data to reduce load time</li> <li>Batch Processing: Load data in batches (1000-10000 records) for efficiency</li> <li>Transaction Management: Use transactions to ensure atomicity</li> <li>Error Recovery: Implement retry logic with exponential backoff</li> <li>Data Validation: Validate both before and after loading</li> <li>Monitoring: Track ETL metrics (records processed, duration, errors)</li> <li>State Management: Track last sync time for incremental loads</li> <li>Deduplication: Handle duplicate records gracefully</li> <li>Schema Versioning: Version your transformation logic</li> </ol> <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>Parallel Processing: Load independent batches in parallel Connection Pooling: Reuse database connections Bulk Operations: Use batch inserts instead of individual queries Index Management: Ensure target nodes have indexes on lookup keys <h3 id="troubleshooting" class="position-relative d-flex align-items-center group"> Troubleshooting <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="troubleshooting" aria-haspopup="dialog" aria-label="Share link: Troubleshooting"> Share link </button> </h3>Slow extraction: Add indexes on source database timestamp columns Memory issues: Reduce batch size or use streaming approach Duplicate nodes: Use MERGE instead of INSERT, ensure unique constraints Missing relationships: Verify all referenced nodes exist before creating edges <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/import" >Import</a> - Data import utilities</li> <li><a href="/tags/migration" >Migration</a> - Schema and data migration</li> <li><a href="/tags/data-quality" >Data Quality</a> - Quality checks in ETL</li> <li><a href="/tags/import/" >Batch Processing</a> - Batch operation patterns</li> <li><a href="/tags/transactions" >Transactions</a> - Transactional loading</li> <li><a href="/tags/performance" >Performance</a> - ETL performance optimization</li> </ul>

Popular

Related Articles

Data Import Guide

Data Export Guide