Databases

Supply chains are graphs. Suppliers feed into warehouses, warehouses feed into distribution centers, and distribution centers feed into retailers. When we model them that way — as nodes and relationships rather than rows and columns — we unlock a set of tools that gives us the ability to ask questions about connectivity, paths, and the structural importance of individual nodes. In this article, we'll build a supply chain, load it into Neo4j via Apache Spark, use NetworkX to identify the most critical nodes in the network, and then simulate a real-world disruption to find alternative routes. The full source code is available on GitHub. The Stack Each tool in the stack does what it does best: ToolRoleApache Spark (local mode)Data generation, transformation, and loading into Neo4jNeo4j (remote, AuraDB)Graph storage and native variable-length path queriesNetworkXBetweenness centrality - identifying the most critical nodesPlotlyInteractive visualization throughout One tool conspicuously absent from this list is Neo4j's Graph Data Science (GDS) library. We'll come back to why and what to reach for when you outgrow the approach described in this article. Setting Up Neo4j AuraDB AuraDB is Neo4j's fully managed cloud database. A free tier is available with no credit card required. Sign up at console.neo4j.io/graphacademy.Create a new AuraDB Free instance.When the instance is created, download or note the credentials - the connection URI, username, and password. Neo4j only shows the password once, so save it somewhere safe.Once the instance is running, open the built-in Query tab and verify connectivity: cypher MATCH (n) RETURN count(n) . This should return 0. We are ready to load data. Before starting Jupyter, export the connection details as environment variables in your shell: Shell export NEO4J_URI=neo4j+s://xxxx.databases.neo4j.io export NEO4J_USERNAME=your_username_here export NEO4J_PASSWORD=your_password_here export NEO4J_DATABASE=your_database_name_here The notebook reads these at startup and raises an error immediately if any are missing. The Data Model The supply chain has four layers connected by SHIPS_TO relationships: Plain Text Suppliers -> Warehouses -> Distribution Centers -> Retailers Each SHIPS_TO relationship carries three properties: cost (shipping cost in dollars)distance (km)capacity (maximum units per shipment) We'll generate a synthetic but reproducible dataset using Faker and NumPy with a fixed random seed, giving us 20 suppliers, 12 warehouses, 10 distribution centers, and 30 retailers with 125 routes across all three layers. Loading the Graph With Spark Spark earns its place in the pipeline by handling the loading step. The Neo4j Spark Connector translates Spark DataFrames into Cypher MERGE statements under the hood, handling the graph write for us: Python spark = ( SparkSession.builder .master("local[*]") .appName("SupplyChainResilience") .config("spark.jars.packages", SPARK_CONNECTOR) .config("neo4j.url", NEO4J_URI) .config("neo4j.authentication.basic.username", NEO4J_USERNAME) .config("neo4j.authentication.basic.password", NEO4J_PASSWORD) .getOrCreate() ) The connector JAR resolves automatically from Maven Central on first run. In a real pipeline, this step would read from S3, a data warehouse, or a Kafka topic and stream records into Neo4j continuously. One important detail is that we'll clear the database before each load using Cypher's IN TRANSACTIONS syntax so each run starts from a clean slate: Cypher MATCH (n) CALL (n) { DETACH DELETE n } IN TRANSACTIONS OF 10000 ROWS We'll then confirm the database is empty before writing new data to the database. Betweenness Centrality With NetworkX Betweenness centrality answers a specific question: if we looked at every possible shortest path between every pair of nodes in the network, how often does each node appear on one of those paths? A node with high betweenness acts as a bridge through which many shortest paths pass. If it disappears, many routes break. A node with low betweenness is peripheral - the network barely notices if it goes offline. We'll pull the graph out of Neo4j via Spark into a NetworkX DiGraph and compute centrality using shipping cost as the edge weight, so the algorithm finds shortest paths by lowest cost rather than fewest hops: Python edges_sdf = ( spark.read.format("org.neo4j.spark.DataSource") .option("query", "MATCH (a)-[r:SHIPS_TO]->(b) " "RETURN coalesce(a.id, a.name) AS source, " " coalesce(b.id, b.name) AS target, " " r.cost AS cost") .load() ) edges_pd = edges_sdf.toPandas() G = nx.DiGraph() for _, row in edges_pd.iterrows(): G.add_edge(row["source"], row["target"], weight = row["cost"]) centrality = nx.betweenness_centrality(G, weight = "cost", normalized = True) Figure 1 shows the full supply chain network before any disruption. Each node type is color-coded: suppliers in blue, warehouses in orange, distribution centers in teal, and retailers in red-orange. The density of connections between layers gives a first impression of where bottlenecks might exist. Figure 1. Full Supply Chain Network Once computed, we'll write the scores back into Neo4j via Spark so Cypher queries can use centrality as a filter or sort key without recomputing it every time. Figure 2 shows the top 15 nodes ranked by betweenness centrality. The length of each bar reflects how often that node appears on a shortest path between other nodes in the network. A longer bar indicates a node that carries a disproportionate share of shortest-path traffic. Figure 2. Top 15 Nodes by Betweenness Centrality Why Not GDS? Neo4j's Graph Data Science (GDS) library has a native gds.betweenness.stream() procedure that runs the same algorithm inside the database using advanced processing. For our small-node demo dataset, NetworkX is instant and requires no additional setup. But nx.betweenness_centrality() runs in O(n * m) time and loads the entire graph into memory. At tens of thousands of nodes, both of those properties become problems. That is exactly where GDS comes in. If you are using Neo4j AuraDB, the same algorithm is available through Aura Graph Analytics — a service that connects directly to your AuraDB instance. The rest of the notebook — Spark for data loading, Plotly for visualization, native Cypher for shortest path — works identically on AuraDB without any changes. Simulating a Disruption With centrality scores computed, we'll identify the highest-scoring node that is a Supplier or Warehouse and mark it as disrupted in Neo4j: Python with driver.session(database = NEO4J_DATABASE) as session: session.run( "MATCH (n {id: $id}) SET n.disrupted = true", id=disrupted_id ) We'll deliberately restrict disruption to Suppliers and Warehouses. Distribution centers are fewer in number, and each carries more routing burden, making them more likely to be sole bridges whose removal severs the network entirely. A warehouse disruption is a more realistic scenario and produces richer alternative-route results. Finding Alternative Routes With Native Cypher With the disrupted node flagged, we'll use Neo4j's built-in variable-length path matching to find alternative routes that avoid it: Cypher MATCH (s:Supplier), (r:Retailer) WHERE s.disrupted IS NULL AND r.disrupted IS NULL MATCH path = (s)-[:SHIPS_TO*..6]->(r) WHERE none(n IN nodes(path) WHERE n.disrupted IS NOT NULL AND n.disrupted = true) WITH s, r, path, reduce( cost = 0.0, rel IN relationships(path) | cost + rel.cost ) AS total_cost ORDER BY total_cost ASC RETURN s.id AS source, r.id AS target, [n IN nodes(path) | coalesce(n.id, n.name)] AS path_nodes, round(total_cost, 2) AS total_cost, length(path) AS hops LIMIT 10 This query is available on both local Neo4j and AuraDB with no additional plugins required. A typical result looks like this: Plain Text source target total_cost hops S006 R001 94.35 3 S007 R026 148.86 3 S007 R017 155.19 3 S019 R026 165.88 3 The cheapest alternative route bypasses the disrupted node entirely at a total shipping cost of $94.35. Note that MATCH (s:Supplier), (r:Retailer) creates a cartesian product for every Supplier/Retailer pair, which is fine for our small dataset. For larger graphs, you would normally constrain the source and destination. The network after disruption is shown in Figure 3. The disrupted node is highlighted in red, and the best alternative route is shown in green, tracing the lowest-cost path from supplier to retailer that avoids the failed node entirely. Figure 3. Best Alternative Route After Disruption Figure 4 compares the top alternative routes by total shipping cost and number of hops. A route with more hops may still be cheaper - the cost comparison makes that trade-off explicit and gives logistics planners a clear basis for decision-making. Figure 4. Alternative Route Cost and Hop Comparison Gotchas and Lessons Learned This project required some debugging. Here are the issues worth knowing about before you try this yourself. Java Version Compatibility PySpark 3.5.x officially supports several versions of Java. However, Java 23 removed javax.security.auth.Subject.getSubject(), which Spark's Hadoop dependency calls during startup. On Java 23 or later, this produces a cryptic UnsupportedOperationException: getSubject is not supported error and Spark never starts. The solution is to install Java 21 LTS alongside any existing Java installation and point PySpark at it before starting Jupyter. Here, for example, using Homebrew on Apple hardware: Shell brew install openjdk@21 export JAVA_HOME=/opt/homebrew/opt/openjdk@21 export PATH="/opt/homebrew/opt/openjdk@21/bin:$PATH" Any existing Java installation is unaffected outside that shell session. The Neo4j Spark Connector 6.x support for Spark 4.x is in active development, so upgrading PySpark to avoid the Java issue is a future option. Relationship Write Deadlocks When writing relationships via the Neo4j Spark Connector with multiple Spark partitions, concurrent writes can deadlock inside Neo4j as transactions compete for the same node locks. The error looks like this: Plain Text ForsetiClient can't acquire EXCLUSIVE NODE_RELATIONSHIP_GROUP_DELETE because it would form a deadlock wait cycle The solution is to call .coalesce(1) on the DataFrame before writing relationships, which forces Spark to write them sequentially from a single partition: Python sdf.coalesce(1).write.format("org.neo4j.spark.DataSource") ... Node writes do not need this because they do not acquire the same lock types. Stale Data Between Runs In the Jupyter notebook's write configuration, the Spark Connector's Overwrite mode merges on node keys but does not remove relationships that existed in a previous run but are absent from the current one. If the dataset size changes between runs, old relationships accumulate alongside new ones, interfering with the graph structure. The solution is to clear the database at the start of every load run rather than relying on Overwrite to clean up after itself. Always confirm the clear succeeded with a node count check before writing. The none() Predicate and Missing Properties This was the subtlest issue of the project. Our disruption query used: Cypher WHERE none(n IN nodes(path) WHERE n.disrupted = true) This returned zero results even when paths clearly existed, and the disrupted node was correctly flagged. In Neo4j, when a node doesn't have a disrupted property at all, n.disrupted = true evaluates to null rather than false. The none() predicate then treats every node as potentially disrupted and filters out all paths. This is exactly how Cypher's three-valued logic works. The solution is an explicit IS NOT NULL check: Cypher WHERE none(n IN nodes(path) WHERE n.disrupted IS NOT NULL AND n.disrupted = true) shortestPath() and Alternative Routes Initially, Neo4j's shortestPath() function was used to find alternative routes. It returned zero results. The reason is that shortestPath() finds the path with fewest hops first, then applies the WHERE none(...) filter. It computes a single shortest path rather than exploring alternative candidates, and filtering on disrupted nodes can eliminate that path without considering longer valid alternatives. The solution is to use a plain variable-length path match with an explicit hop limit instead. This lets the WHERE clause filter while still returning valid results: Cypher MATCH path = (s)-[:SHIPS_TO*..6]->(r) WHERE none(n IN nodes(path) WHERE ...) Guaranteed Connectivity in Generated Data With purely random route generation, it's possible for a single node to end up as the only connection between two layers - a so-called sole bridge. Disrupting that node severs the network completely and leaves no alternative routes to find. The solution is to generate routes with a guaranteed minimum connectivity. So, every source node gets at least two outbound routes, and every target node gets at least two inbound routes before random fill: Python def make_routes(sources, targets, n_routes, min_out=2, min_in=2): # Guarantee every source has at least min_out outbound routes for s in src_ids: sample = rng.choice(tgt_ids, size = min(min_out, len(tgt_ids)), replace = False) for t in sample: if (s, t) not in seen: rows.append(make_row(s, t)) # Guarantee every target has at least min_in inbound routes for t in tgt_ids: sample = rng.choice(src_ids, size = min(min_in, len(src_ids)), replace = False) for s in sample: if (s, t) not in seen: rows.append(make_row(s, t)) # Fill remaining routes randomly ... Cypher 25 Syntax If you are running Neo4j 2025.06 or later, the CALL { WITH n ... } subquery syntax used in batch deletes is deprecated. Use the new variable scope syntax instead: Cypher MATCH (n) CALL (n) { DETACH DELETE n } IN TRANSACTIONS OF 10000 ROWS Summary We've built a supply chain resilience analysis pipeline that models a supply chain as a graph, identifies its most critical nodes using betweenness centrality, simulates a real-world disruption, and finds alternative routes using native Cypher. Each tool did what it does best: Spark handled bulk data loading, Neo4j stored the graph and answered path queries, NetworkX computed the graph algorithm, and Plotly produced interactive visualizations at every stage. The gotchas section above contains several useful engineering lessons, which should save you time and effort on your projects. The full source code is available on GitHub.

By Akmal Chaudhri CORE

Every performance guide starts the same way. "Add an index." And yes, indexes matter. But I've spent years fixing production databases, and here's the truth: indexing is the easy 20%. The hard 80% is everything nobody writes blog posts about. I once spent three days chasing a query that had a perfect index. The index wasn't the problem. The problem was that the database's own statistics were lying to it. This article is about that other 80%. Why This Problem Keeps Coming Back Most teams treat database performance as a one-time task. Add indexes during launch week. Move on. But databases are not static. Data grows. Traffic patterns shift. Your "small lookup table" from six months ago now has four million rows. The query that ran in 2ms during testing can quietly become a 4-second query in production. Nobody notices until users complain. Here's the uncomfortable part: indexing advice assumes your query planner always makes good decisions. It doesn't. Query planners are guessing machines. They guess based on statistics, and statistics go stale. Why Developers Struggle With This Most backend engineers learn SQL as a language, not as an execution engine. You write SELECT * FROM orders WHERE customer_id = 123, it returns rows, and that feels like magic. But behind that query is a planner making dozens of decisions: Should it use an index or scan the whole table?Should it join tables in this order or that order?Should it use a hash join or a nested loop? Developers rarely see this decision-making. So when performance drops, the first (and often only) fix is "add an index." Sometimes that helps. Often it doesn't touch the real issue. The Real Problem: Stale Statistics Most relational databases (Postgres, MySQL, SQL Server) use cost-based optimizers. These optimizers don't know your data. They estimate it using statistics — sampled snapshots of your table's shape. If those statistics are outdated, the optimizer makes bad guesses. It might think a column has 10 distinct values when it actually has 10 million. Here's a real example from a Postgres system I worked on: SQL -- Table: events (48 million rows) EXPLAIN ANALYZE SELECT * FROM events WHERE event_type = 'checkout_completed' AND created_at > NOW() - INTERVAL '7 days'; The plan showed a sequential scan, even though we had an index on event_type. Why? The table statistics thought checkout_completed made up 40% of rows. In reality, it was 0.3%. The fix wasn't a new index. It was this: SQL ANALYZE events; One command. Query time dropped from 6.2 seconds to 90 milliseconds. Lesson: An index is only useful if the planner trusts it's worth using. Common Mistakes Developers Make Let's go through the mistakes I see over and over, across different companies and different stacks. 1. Trusting SELECT * Pulling every column, even ones you don't need, forces the database to read more data pages than necessary. On wide tables, this alone can double query time. 2. Ignoring the N+1 Query Pattern This one is everywhere in ORM-heavy codebases. Python # Bad: 1 query for orders + N queries for customers orders = Order.objects.all() for order in orders: print(order.customer.name) # triggers a new query each time Python # Good: 1 query total orders = Order.objects.select_related("customer").all() for order in orders: print(order.customer.name) If you have 500 orders, the bad version runs 501 queries. The good version runs 1. 3. Deep Pagination With OFFSET SQL -- Gets slower as the offset grows SELECT * FROM products ORDER BY id LIMIT 20 OFFSET 100000; The database still has to scan and discard 100,000 rows before returning your 20. On a page 5,000 request, this crawls. Better approach — keyset pagination: SQL SELECT * FROM products WHERE id > 100000 ORDER BY id LIMIT 20; This uses the index directly. No wasted scanning. Pagination MethodPerformance at Page 10Performance at Page 5000ComplexityOFFSET/LIMITFastVery slowLowKeyset (cursor-based)FastFastMediumPrecomputed pagesFastFastHigh (needs caching) 4. Doing Math on Indexed Columns SQL -- Index on created_at is useless here SELECT * FROM orders WHERE DATE(created_at) = '2026-07-20'; Wrapping a column in a function usually breaks the database's ability to use its index. SQL -- This keeps the index usable SELECT * FROM orders WHERE created_at >= '2026-07-20' AND created_at < '2026-07-21'; Small rewrite. Big difference. How Modern Systems Actually Solve This Real production systems don't rely on a single trick. They layer several defenses. Plain Text Client Request │ ▼ API Layer │ ▼ Query Cache (Redis) ├── cache hit? return here ▼ Connection Pool (PgBouncer) │ ▼ Read Replica (for reads) ──── Primary DB (for writes) │ ▼ Query Planner + Statistics │ ▼ Storage Engine Each layer exists to reduce pressure on the layer below it. Miss the cache, and you hit the pool. Miss the primary's write load, and reads go to a replica. Connection Pooling Matters More Than People Think Opening a raw database connection is expensive. It involves a TCP handshake, authentication, and memory allocation on the database side. Without pooling, a burst of traffic can create hundreds of connections in seconds. Postgres, for example, starts choking well before 500 connections. Plain Text # pgbouncer.ini [databases] mydb = host=127.0.0.1 port=5432 dbname=mydb [pgbouncer] pool_mode = transaction max_client_conn = 1000 default_pool_size = 25 With transaction pooling mode, PgBouncer hands out a real database connection only for the duration of a transaction, then returns it to the pool. This lets 1,000 app connections share just 25 real ones. Lock Contention: The Silent Killer This is the bottleneck that almost nobody talks about, because it doesn't show up in slow query logs the same obvious way. Here's what happened to us. A "quick" query started timing out during peak hours: SQL UPDATE inventory SET stock = stock - 1 WHERE product_id = 42; Individually, this query was fast. But during a flash sale, hundreds of these updates hit the same row at the same time. Each transaction had to wait for the previous one to release its row lock. The queries weren't slow. They were queued. Plain Text Time Transaction A Transaction B Transaction C 0ms LOCK row 42 waiting... waiting... 5ms UPDATE + COMMIT LOCK row 42 waiting... 6ms UPDATE + COMMIT LOCK row 42 7ms UPDATE + COMMIT How we fixed it: Moved to an eventual-consistency model for stock counts (queue-based decrement)Used SELECT ... FOR UPDATE SKIP LOCKED for job-queue-style tablesBatched decrements instead of doing them one row at a time SQL -- Instead of 100 individual UPDATE statements UPDATE inventory SET stock = stock - sub.qty FROM ( VALUES (42, 3), (43, 1), (44, 7) ) AS sub(product_id, qty) WHERE inventory.product_id = sub.product_id; One batched statement instead of a hundred lock acquisitions. Isolation Levels: A Trade-off, Not a Setting You Ignore Most engineers leave the isolation level at whatever the database defaults to. That's usually fine — until it isn't. Isolation LevelPreventsPerformance CostCommon Use CaseRead UncommittedNothing muchLowestRarely used, riskyRead CommittedDirty readsLowDefault in Postgres, most web appsRepeatable ReadNon-repeatable readsMediumFinancial reports, reconciliationSerializablePhantom readsHighestBanking transactions, inventory locks Higher isolation means more correctness guarantees. It also means more locking, more retries, and lower throughput. Don't default to Serializable "to be safe." You'll pay for it in throughput, and most apps don't need it. Query Plan Reading: A Skill Most Engineers Skip If you only remember one thing from this article, remember this: learn to read EXPLAIN ANALYZE output. It tells you the truth. Everything else is a guess. SQL EXPLAIN ANALYZE SELECT o.id, c.name FROM orders o JOIN customers c ON o.customer_id = c.id WHERE o.status = 'pending'; Sample output to watch for: SQL Hash Join (cost=120.50..3400.22 rows=850 width=64) (actual time=12.100..340.556 rows=42000 loops=1) Hash Cond: (o.customer_id = c.id) -> Seq Scan on orders o (cost=0.00..2900.00 rows=850) (actual time=0.020..300.100 rows=42000 loops=1) Notice the gap: the planner estimated 850 rows. The actual count was 42,000. That's a 49x miss. When estimated and actual rows differ by a wide margin, that's your signal. Stale statistics, bad indexes, or a query shape the planner can't reason about well. Denormalization: Sometimes the Right Move Normalization is taught as the "correct" way to design schemas. In practice, strict normalization can hurt performance on read-heavy systems. We had a dashboard query joining six tables to compute one number: total revenue per region. SQL SELECT r.name, SUM(o.total) FROM orders o JOIN customers c ON o.customer_id = c.id JOIN regions r ON c.region_id = r.id JOIN order_items oi ON oi.order_id = o.id JOIN products p ON oi.product_id = p.id JOIN categories cat ON p.category_id = cat.id GROUP BY r.name; This ran in 4 seconds. Dashboard needed sub-second response. We added a summary table, updated by a nightly job: SQL CREATE TABLE revenue_by_region ( region_name TEXT PRIMARY KEY, total_revenue NUMERIC, updated_at TIMESTAMP ); Dashboard query became: SQL SELECT region_name, total_revenue FROM revenue_by_region; From 4 seconds to 8 milliseconds. The trade-off: data is now up to 24 hours stale. This only works if your business can tolerate staleness. For real-time fraud detection, this approach would be wrong. Know your consistency requirements before you denormalize. Performance Considerations Checklist Before shipping a query to production, run through this: ✔ Did you check EXPLAIN ANALYZE, not just EXPLAIN? ✔ Are your table statistics current (ANALYZE run recently)? ✔ Does the query avoid functions wrapped around indexed columns? ✔ Are you selecting only the columns you need? ✔ Is pagination using keyset instead of large OFFSET values? ✔ Are batch writes used instead of row-by-row loops? ✔ Is the isolation level appropriate for the use case, not just the default? ✔ Have you tested this query against production-sized data, not a dev sample? Security Considerations Performance work sometimes creates security gaps. Watch for these: Dynamic query building for "flexible filters" often leads to string concatenation, which opens SQL injection risk. Use parameterized queries even for performance-tuned raw SQL.Read replicas used for reporting sometimes get looser access controls because "it's just a read replica." That's still your data.Caching layers (Redis, Memcached) can leak sensitive data if you cache full row objects without checking what's in them. Scaling Challenges As systems grow, new problems appear that indexing can't fix: Plain Text Single DB Instance │ ▼ Growing write load │ ▼ Read Replicas (helps reads, not writes) │ ▼ Still hitting write limits │ ▼ Sharding (splits writes across nodes) │ ▼ Cross-shard joins become painful Sharding solves write throughput but creates a new problem: joins across shards don't work the way they used to. You end up doing joins in application code, which is slower and more error-prone than letting the database do it. This is why teams delay sharding as long as possible. It's a last resort, not a first optimization. What We Learned A few honest lessons from years of doing this: Statistics decay silently. Schedule ANALYZE (or your database's equivalent) as a routine job, not an afterthought.The slowest part of a query is often not the query itself. It's lock waiting, connection exhaustion, or network round trips.ORMs hide problems well. They also hide the N+1 pattern extremely well. Turn on query logging in staging and actually read it.Caching isn't free. Cache invalidation bugs have cost us more debugging time than the queries we were trying to avoid.Nobody reads execution plans until something breaks. Read them earlier. It's a habit, not a rescue tool. When Not to Use These Techniques Not every optimization belongs in every system. Don't denormalize a table that changes every second the sync job will never catch up.Don't add read replicas if your write load, not read load, is the actual bottleneck.Don't reach for sharding if a bigger instance and better indexing would solve it for the next two years.Don't tune isolation levels down for "performance" on a system handling money movement. Optimization without a clear bottleneck measurement is just guessing with extra steps. Final Thoughts Indexing is the first lesson in database performance, not the last one. The real bottlenecks stale statistics, lock contention, bad pagination, and isolation level mismatches don't show up in a "10 SQL Tips" listicle. They show up at 2 AM, during a traffic spike, when your on-call phone rings. The next challenge for most teams isn't learning these techniques. It's building the habit of checking for them before a query becomes a production incident. That habit reading EXPLAIN ANALYZE, tracking replication lag, watching lock wait times matters more than any single trick in this article.

The Problem: Our p99 Was 3-5 Seconds Our PyFlink pipeline was missing its latency SLO by seconds. The pipeline itself was straightforward: consume events from Kafka, transform them, serialize them as Protobuf, and write the results to downstream systems. Yet under production load, p99 end-to-end latency was consistently in the 3-5 second range. Profiling pointed us to an unexpected bottleneck: we were deserializing Protobuf messages in Python, even though the Flink runtime processing our stream was JVM-based. Every record that entered the Python path had to cross the JVM-to-Python process boundary, get parsed by a Python UDF, and then cross back. The business logic wasn't the problem. The doorway was. We moved Protobuf deserialization to Flink's JVM-side Protobuf format and kept Python for orchestration and SQL. In our environment, p99 dropped to approximately 500 milliseconds, with less code and a pipeline that is easier to reason about. Verified on AWS Managed Service for Apache Flink (formerly Kinesis Data Analytics). Why Python-Side Deserialization Is So Expensive The naive PyFlink architecture looks like this: A Kafka source table declared with a generic format (raw, json, or a SimpleStringSchema), so every record arrives as opaque bytes or a string.A Python map() or UDF that imports generated _pb2.py classes and calls ParseFromString() on every message.Downstream transforms and sinks. Two costs hide in step 2, and they compound at high throughput. The process boundary. PyFlink is not Python running inside Flink; it is a JVM runtime coordinating with a separate Python execution environment. Every record that enters the Python execution path incurs overhead associated with moving data between the JVM and Python, and depending on the operator and execution mode, that can involve serialization and inter-process communication in both directions. For a per-record deserialization UDF on a latency-sensitive pipeline, that overhead is paid before the actual business transformation begins. Per-record parse cost. Even when Python's Protobuf implementation uses its native backend, parsing in a Python UDF still requires the record to enter the Python execution path. When the workload is latency-sensitive and high-throughput, the combination of serialization, inter-process communication, Python execution, and parsing overhead can become significant. In our case, profiling showed that this path was a major contributor to our latency. In our pipeline, these two costs together accounted for the bulk of the gap between a 3–5 second p99 and the ~500ms target we needed, before the enrichment logic even began executing. The Key Realization: PyFlink Already Runs on the JVM Here's the insight that changes the architecture: if Protobuf is declared at the table DDL level, Flink's Kafka connector deserializes it with its native, optimized JVM-based Protobuf format before any data reaches the Python side. Your columns simply arrive typed and ready. Python's role shrinks to what it's genuinely good at in this stack: orchestration and SQL. No rewrite to Java. No change to how jobs are deployed. Just a different declaration of intent. The trade is that Flink's native Protobuf format needs compiled Java message classes on the classpath; it does not consume .proto files or Python _pb2 modules directly. That means adding a small build step to your workflow, which we'll cover below. Implementation The pipeline splits into two declarative jobs. Job 1: JSON In, Protobuf Out The source table reads the raw JSON topic; the sink table declares format = 'protobuf' and points at the compiled Java class. The JVM handles typed-row-to-Protobuf encoding. SQL -- SOURCE: raw JSON payload as STRING plus Kafka record timestamp CREATE TABLE source_events_json ( event_data STRING, kafka_timestamp TIMESTAMP(3) METADATA FROM 'timestamp' ) WITH ( 'connector' = 'kafka', 'topic' = '${INPUT_JSON_TOPIC}', 'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}', 'scan.startup.mode' = 'latest-offset', 'format' = 'raw' ); -- SINK: Protobuf out to Kafka (JVM handles typed row to Protobuf) CREATE TABLE sink_events_pb ( id STRING, organization_id STRING, event_ts ROW<`seconds` BIGINT, `nanos` INT>, is_active BOOLEAN, event_type STRING ) WITH ( 'connector' = 'kafka', 'topic' = 'acme.events.pb.v1', 'properties.bootstrap.servers' = 'kafka:9092', 'format' = 'protobuf', 'protobuf.message-class-name' = 'com.acme.events.v1.EventOuterClass$EnrichedEvent' ); -- TRANSFORM: pure SQL, no Python UDFs INSERT INTO sink_events_pb SELECT JSON_VALUE(event_data, '$.id') AS id, JSON_VALUE(event_data, '$.organization_id') AS organization_id, ROW( UNIX_TIMESTAMP(NULLIF(JSON_VALUE(event_data, '$.after.event_ts'), '')), CAST(EXTRACT(NANOSECOND FROM CAST(NULLIF(JSON_VALUE(event_data, '$.after.event_ts'), '') AS TIMESTAMP_LTZ(9))) AS INT) ) AS event_ts, CAST(JSON_VALUE(event_data, '$.is_active') AS BOOLEAN) AS is_active, JSON_VALUE(event_data, '$.event_type') AS event_type FROM source_events_json; Note what's absent: no ParseFromString(), no _pb2.py imports, no Python deserialization loop. The Python program registers DDL and runs SQL. Job 2: Protobuf In, OpenSearch Out Downstream, the sanitized Protobuf topic becomes a typed source, using the same protobuf.message-class-name property, plus ignore-parse-errors so a malformed record can't poison the pipeline. SQL -- SOURCE: Protobuf from the sanitized Kafka topic CREATE TABLE kafka_source_pb ( id STRING, organization_id STRING, event_ts ROW<`seconds` BIGINT, `nanos` INT>, is_active BOOLEAN, event_type STRING, kafka_timestamp TIMESTAMP(3) METADATA FROM 'timestamp' ) WITH ( 'connector' = 'kafka', 'topic' = 'acme.events.pb.v1', 'properties.bootstrap.servers' = 'kafka:9092', 'scan.startup.mode' = 'latest-offset', 'format' = 'protobuf', 'protobuf.message-class-name' = 'com.acme.events.v1.EventOuterClass$EnrichedEvent', 'protobuf.ignore-parse-errors' = 'true' ); -- SINK: OpenSearch (JSON) CREATE TABLE opensearch_sink ( id STRING, organization_id STRING, event_ts TIMESTAMP_LTZ(3), is_active BOOLEAN, event_type STRING, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'opensearch-2', 'hosts' = '${OPENSEARCH_ENDPOINT}:443', 'index' = 'acme-events-v1', 'format' = 'json' ); INSERT INTO opensearch_sink SELECT id, organization_id, TO_TIMESTAMP_LTZ(event_ts.seconds * 1000, 3), is_active, event_type FROM kafka_source_pb; The Build Step: Getting Java Classes Onto Flink's Classpath The one genuinely new piece of workflow is compiling your .proto definitions to Java and packaging them into the job's fat JAR. The essential Maven pieces: XML <dependencies> <dependency> <groupId>com.google.protobuf</groupId> <artifactId>protobuf-java</artifactId> <version>3.25.5</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-protobuf</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-kafka</artifactId> <version>${flink.connector.kafka.version}</version> </dependency> <!-- plus your sink connectors, e.g. flink-connector-opensearch2 --> </dependencies> Two practices that made this maintainable for us: Version-control the generated Java sources (or generate them in CI from a single canonical .proto repo) and pull them in with build-helper-maven-plugin's add-source, rather than compiling .proto files in every consuming project. One schema source of truth, many consumers.Shade everything into one JAR with maven-shade-plugin, excluding signature files (META-INF/*.SF, *.DSA, *.RSA). On AWS Managed Flink, pass it via the job's JAR configuration; on self-managed Flink, drop it in lib/ or use --classpath. The full workflow: define the .proto, compile it to Java with protoc, package the fat JAR, put it on Flink's classpath, author the PyFlink job with the DDL above, then deploy and watch end-to-end p99. How We Measured the Improvement We measured end-to-end p99 latency as the time from a record landing on the source Kafka topic to the corresponding OpenSearch write being acknowledged MetricBeforeAfterp99 latency3-5s~500msSustained throughput~5,000 events/sec~5,000 events/secFlink parallelism128Python UDF parsingYesNoJVM/Python boundary on hot pathYesNoProtobuf decodingPythonJVM Results End-to-end p99 latency around 500 milliseconds in our environment at production load, down from a 3-5 second baseline, by eliminating per-record JVM-to-Python crossings and Python-side parsing on the hot pathLess code. The deserialization UDFs, the _pb2 imports, and their error handling all disappeared. What remains is DDL plus SQLSimpler and easier to operate. The pipeline now relies on Flink's Kafka connector and Protobuf format for serialization and parsing, with built-in parse-error handling, instead of hand-rolled Python parsing When This Optimization Won't Help Moving Protobuf decoding to the JVM won't automatically solve every latency problem. If your pipeline's critical path is dominated by sink backpressure, network latency, external API calls, state access, or checkpointing overhead rather than deserialization, changing the serialization path may have little effect on end-to-end latency. This optimization is most valuable when profiling specifically shows that Python execution and JVM/Python data movement are significant contributors to the critical path, which is why we'd recommend profiling first rather than applying this as a default change. When You Should Still Use Python UDFs This pattern is not "never write Python UDFs." It's "keep them off the per-record deserialization path." Python remains the right tool when: The transformation genuinely needs Python libraries (ML feature computation, model inference, specialized parsing that has no SQL equivalent).Throughput is modest and developer velocity matters more than the last hundred milliseconds.You're prototyping. Even then, declare the format natively from day one anyway; it costs nothing and you won't have to migrate later. If a UDF is unavoidable on a hot path, at least let the JVM do the deserialization first so the UDF receives typed columns rather than raw bytes. Gotchas Worth Knowing Before You Ship Property syntax varies by Flink version. Some versions use format = 'protobuf'; newer key/value descriptors prefer value.format = 'protobuf'. Check your version's docs.Enums: surface them as STRING if you need ergonomic SQL manipulation, or keep them numeric with a lookup table.Schema evolution: favor backward-compatible, additive changes with defaults. Because the compiled Java classes are baked into the JAR, a schema change means a rebuild and redeploy, so make that a deliberate, versioned step in CI rather than an afterthought. ignore-parse-errors is your safety net during rollout windows, but monitor the drop counter so it doesn't silently eat data.Benchmark end-to-end, not just the UDF: source lag, operator latency, and sink acknowledgments under production load patterns. Deserialization wins can be masked, or dwarfed, by sink backpressure.Security: lock down OpenSearch credentials and TLS; pin Kafka client versions compatible with your Flink release. Closing Thoughts We didn't rewrite the pipeline in Java. We removed an unnecessary per-record JVM-to-Python boundary from the hot path and let Flink's JVM-native Protobuf format do the work it was designed to do. If your PyFlink job parses Protobuf messages in Python today, check whether Flink's native format support can move that work into the JVM-side execution path. For latency-sensitive pipelines, eliminating unnecessary Python boundaries may be one of the highest-leverage optimizations to investigate, especially when profiling shows that serialization and Python execution are on the critical path.

Cloud migration projects almost always treat security as a downstream concern something to bolt on after workloads have already moved, once the “real” migration work is done. Across dozens of enterprise migrations spanning finance, healthcare, and manufacturing workloads, that ordering is consistently the source of the costliest rework: reopened firewall rules, retrofitted identity models, and access reviews that should have happened before a single virtual machine was provisioned. The pattern holds regardless of which cloud provider is on the receiving end. What follows is a framework provider-agnostic by design for embedding zero-trust principles into the migration process itself, rather than applying them after the fact. Why Bolt-On Security Fails Traditional migration playbooks are organized around workload movement: discover, assess, re-platform, cut over, optimize. Security tasks are usually inserted late, as a checklist item before go-live. Three consequences follow reliably: Implicit trust survives the move. Implicit trust survives the move. On-premises networks often rely on perimeter trust: anything inside the firewall is assumed safe. When that assumption is lifted-and-shifted into the cloud without redesign, the perimeter simply becomes larger and harder to defend.Identity sprawl compounds. Identity sprawl compounds. Migrations frequently multiply service accounts, temporary roles, and cross-environment credentials used to bridge on-prem and cloud during cutover. Few of these get cleaned up.Retrofitting is expensive. Retrofitting is expensive. Segmenting a network or re-scoping IAM roles after hundreds of workloads are already live requires downtime windows and change approvals that could have been avoided by designing correctly the first time. The Framework: 4 Pillars, Applied in Migration Order The framework below organizes zero-trust adoption into four pillars, sequenced to match the natural phases of a migration rather than treated as a parallel workstream. 1. Identity as the New Perimeter Before any workload assessment begins, establish the identity model the migrated environment will use, not the one the source environment happens to have. Define role-based access aligned to job function, not to legacy group membership inherited from the source directory.Require multi-factor authentication for every administrative path into the target environment before migration tooling is granted access, not after.Treat every migration-tooling service account as temporary by default, with an explicit expiration and re-certification date. 2. Segment Before You Migrate, Not After Network segmentation decisions made during the assessment phase are cheap. The same decisions made post-migration require change windows and stakeholder sign-off. Group workloads into trust tiers during discovery (e.g., internet-facing, internal-only, regulated-data) rather than assuming a flat network topology will be corrected later.Design micro-segmentation boundaries around workload tiers before the first server moves, so that day-one network policy already reflects least-privilege communication paths.Validate east-west traffic rules against actual application dependency maps, not assumed ones; dependency mapping tools exist for this precisely because assumptions are usually wrong. 3. Encrypt and Verify at Every Hop, Not Just at Rest Most cloud providers make encryption at rest close to a default setting. The gap is almost always in transit and in verification. Require mutual TLS or equivalent between service-to-service calls introduced during migration, especially temporary bridging connections between source and target environments.Treat data classification as a migration input, not a post-migration audit finding. Classify before you move, so encryption and access policy can be applied by tier from day one.Build verification checkpoints into the cutover plan itself: an environment isn't “migrated” until its access logs confirm no implicit-trust paths remain from the legacy network. 4. Assume Breach, Instrument Accordingly The final pillar is operational rather than architectural: build the assumption of compromise into monitoring from the start of the migration, not after an incident. Instrument logging and alerting for the target environment before cutover, so that abnormal access patterns are visible from hour one rather than backfilled weeks later.Run tabletop exercises against the migrated architecture; specifically, lessons from the legacy environment's incident response plan rarely transfer cleanly.Track a small set of leading indicators (privileged session anomalies, unexpected cross-tier traffic, credential reuse across environments) rather than waiting for a full SIEM rollout to catch up. Lessons From Enterprise Deployments A few patterns show up consistently across large, regulated deployments: Sequencing beats scope. Organizations that tried to implement all four pillars simultaneously across an entire estate stalled. The deployments that succeeded phased identity and segmentation first, then layered encryption verification and monitoring in as workloads landed.Legacy exceptions need sunset dates. Legacy exceptions need sunset dates. Every migration produces temporary trust exceptions to keep the business running during cutover. Without a hard expiration date attached at creation, these exceptions become permanent attack surface.Cross-functional ownership matters more than tooling. Cross-functional ownership matters more than tooling. The deployments with the fewest post-migration security incidents were the ones where network, identity, and application teams jointly signed off on the trust model before migration started, not the ones with the most sophisticated tooling. Common Pitfalls Treating zero trust as a product purchase rather than an architectural discipline applied throughout the migration lifecycle.Migrating identity and network configuration as-is with the intention to “harden it later” rarely comes without an incident forcing it.Measuring migration success purely on workload count and timeline, with security posture reviewed only at the end. Closing Thought Zero trust and cloud migration are often treated as separate initiatives running on separate timelines. The organizations that get the best outcomes fewer post-migration incidents and faster time-to-secure-operations are the ones that treat zero trust as a design constraint on the migration itself, sequenced into discovery, assessment, and cutover rather than appended afterward. The framework above is intentionally provider-agnostic because the discipline it describes identity first, segmentation before movement, verification at every hop, and instrumentation from day one holds regardless of which cloud the workloads land on.

Many engineering teams currently face a knowledge challenge. Information does exist; however, the information is distributed across various documentation formats such as design documents, runbooks, architectural notes, deployment guides, and incident reports. In general, a developer is aware of which services depend on each other (the Checkout Service depends upon the Payment API), the database or technology stack being used by the dependent services (the Payment API utilizes PostgreSQL), and who owns/operates the dependent service (Platform Team owns and operates the Payment API), however, these pieces of information typically reside in separate locations. A traditional search capability can locate documents with references to those terms. Using a retrieval-augmented generation (RAG) solution allows retrieval of relevant fragments/chunks based on contextually relevant keywords provided to the RAG model, which can then be passed along to a large language model (LLM). That strategy is effective for answering most types of questions. However, there are certain types of questions that are not simply about identifying the content of one document; they are about relating concepts. For instance: Plain Text Which team is owns the service that 'Checkout' relies upon? Relating all applicable data points is necessary when answering this type of question. To aid in the process of relating all applicable data points, a knowledge graph can become helpful. This article describes building a very basic knowledge graph from unformatted text using Python. As described above, keeping the example as simple as possible, but again, this is essentially how you would implement your own GraphRAG system: Identify entities from text, determine how those entities relate to each other, represent those relations as edges within a graph structure, and query that graph for relevant data prior to generating an answer. What We Are Building We will start with a few short engineering notes: Plain Text Checkout Service depends on Payment API. Payment API uses PostgreSQL. Platform Team owns Payment API. Recommendation Service calls Catalog API. Catalog API uses Elasticsearch. Search Team owns Catalog API. From those notes, we want to build a graph like this: Plain Text Checkout Service --DEPENDS_ON--> Payment API Payment API --USES--> PostgreSQL Platform Team --OWNS--> Payment API Recommendation Service --CALLS--> Catalog API Catalog API --USES--> Elasticsearch Search Team --OWNS--> Catalog API Once we have that structure, we can answer questions by traversing the graph instead of scanning raw text. Project Setup Create a new folder: Shell mkdir python-knowledge-graph cd python-knowledge-graph Create a requirements.txt file: Plain Text networkx==3.3 spacy==3.7.5 Install the dependencies: Shell pip install -r requirements.txt python -m spacy download en_core_web_sm We will use: spaCy for basic Natural Language Processing (NLP)NetworkX for building and querying the graph For this first example, we will not use a database. Keeping everything in memory makes the workflow easier to understand. Step 1: Identify The Input Documents Create a new file called build_graph.py. Python documents = ["Checkout Service depends on Payment API.", "Payment API uses PostgreSQL.", "Platform Team owns Payment API.", "Recommendation Service calls Catalog API.", "Catalog API uses Elasticsearch.", "Search Team owns Catalog API.", ] In a real-world deployment, the input document could have originated from a variety of sources (e.g., Markdown files, Confluence pages, GitHub repositories, service catalogs, incident reports). In this case, a couple of lines of example text should be sufficient to illustrate the concept. Step 2: Determine Relationship Triples Knowledge graphs typically store information in triple format (the subject, its relationship with another entity, and that other entity): Plain Text subject ---Relationship---> object An example would be: Plain Text Checkout Service --DEPENDS_ON--> Payment API In general, relationship detection in a large-scale application is often performed by a trained machine learning model. For demonstration purposes in this post, we'll utilize a simple rule-based detector to keep things straightforward. Add the following to build_graph.py: Python import re RELATION_PATTERNS = [ (r"(.+?) depends on (.+?)\.", "DEPENDS_ON"), (r"(.+?) uses (.+?)\.", "USES"), (r"(.+?) owns (.+?)\.", "OWNS"), (r"(.+?) calls (.+?)\.", "CALLS"), ] def extract_triples(text): triples = [] for pattern, relation in RELATION_PATTERNS: match = re.match(pattern, text, re.IGNORECASE) if match: subject = normalize_entity(match.group(1)) object_ = normalize_entity(match.group(2)) triples.append((subject, relation, object_)) return triples def normalize_entity(value): return value.strip() This function is intentionally simple. It looks for a small set of verbs and converts each sentence into a graph-friendly structure. Try it: Python for doc in documents: print(extract_triples(doc)) Expected output: Plain Text [('Checkout Service', 'DEPENDS_ON', 'Payment API')] [('Payment API', 'USES', 'PostgreSQL')] [('Platform Team', 'OWNS', 'Payment API')] [('Recommendation Service', 'CALLS', 'Catalog API')] [('Catalog API', 'USES', 'Elasticsearch')] [('Search Team', 'OWNS', 'Catalog API')] This is the first useful step. We have converted unstructured text into structured facts. Step 3: Create a Graph With NetworkX We can now add those triplets into a directed graph. Python import networkx as nx def build_knowledge_graph(documents): graph = nx.DiGraph() for doc in documents: triples = extract_triples(doc) for subject, relation, object_ in triples: graph.add_node(subject) graph.add_node(object_) graph.add_edge(subject, object_, relation=relation, source_text=doc) return graph The use of a directed graph makes sense when dealing with relations that are directional. This: Plain Text Checkout Service --DEPENDS_ON--> Payment API does not mean the same thing as this: Python Payment API --DEPENDS_ON--> Checkout Service Direction matters for dependency analysis, ownership lookup, impact analysis, and retrieval. Step 4: Print the Graph Add a helper function: Python def print_graph(graph): for source, target, data in graph.edges(data=True): relation = data["relation"] print(f"{source} --{relation}--> {target}") Now run the full flow: Python if __name__ == "__main__": graph = build_knowledge_graph(documents) print_graph(graph) Output: Plain Text Checkout Service --DEPENDS_ON--> Payment API Payment API --USES--> PostgreSQL Platform Team --OWNS--> Payment API Recommendation Service --CALLS--> Catalog API Catalog API --USES--> Elasticsearch Search Team --OWNS--> Catalog API At this point, we have a working knowledge graph. It is small, but it already gives us something normal keyword search does not: explicit relationships. Step 5: Query the Graph Let’s answer a practical question: Plain Text Who owns the API that Checkout Service depends on? That requires two hops: Plain Text Checkout Service -> Payment API -> Platform Team The first hop finds the dependency. The second hop finds the owner. Add this function: Python def find_owner_of_dependency(graph, service_name): results = [] for dependency in graph.successors(service_name): edge_data = graph.get_edge_data(service_name, dependency) if edge_data["relation"] != "DEPENDS_ON": continue for possible_owner in graph.predecessors(dependency): owner_edge = graph.get_edge_data(possible_owner, dependency) if owner_edge["relation"] == "OWNS": results.append( { "service": service_name, "dependency": dependency, "owner": possible_owner, } ) return results Call it: Python owners = find_owner_of_dependency(graph, "Checkout Service") for item in owners: print( f"{item['owner']} owns {item['dependency']}, " f"which is used by {item['service']}." ) Output: Plain Text Platform Team owns Payment API, which is used by Checkout Service. This is a simple example, but it shows the main value of graph-based retrieval. We did not search for similar text. We followed relationships. Step 6. Save the Graph When building a small prototype, you can often save the graph as JSON. Here’s how to do that. Python import json def export_graph(graph, output_path): data = { "nodes": list(graph.nodes()), "edges": [ { "source": source, "target": target, "relation": edge_data["relation"], "source_text": edge_data["source_text"], } for source, target, edge_data in graph.edges(data=True) ], } with open(output_path, "w", encoding="utf-8") as file: json.dump(data, file, indent=2) export_graph(graph, "graph.json") The output looks like this: JSON { "nodes": [ "Checkout Service", "Payment API", "PostgreSQL", "Platform Team", "Recommendation Service", "Catalog API", "Elasticsearch", "Search Team" ], "edges": [ { "source": "Checkout Service", "target": "Payment API", "relation": "DEPENDS_ON", "source_text": "Checkout Service depends on Payment API." } ] } Keeping the original source_text when saving the graph is important. This is because if you were to plug this graph into either a RAG or GraphRAG pipeline, there needs to be some way for the LLM to see not just the relationship within the graph but also have access to the actual supporting text for each one. Where spaCy Fits In Regular expressions were used in the previous example because of how simple your examples were (you know exactly what will appear). However, most real-world documentation is not as straightforward. This is why we use spaCy. It can find all types of named entities; e.g., organization, product, person, etc. Below is a small sample: Python import spacy nlp = spacy.load("en_core_web_sm") text = "Platform Team owns Payment API, which uses PostgreSQL." doc = nlp(text) for entity in doc.ents: print(entity.text, entity.label_) As with many things with regard to NLP, depending upon your specific model and the input data you provide, spaCy may automatically find some entities. However, generally speaking, you will need to either develop custom rules or train your own model to extract certain types of entities that pertain to specific domains of interest (e.g., engineering-related terms such as internal services, APIs, teams, etc.). In practice, one common method to take advantage of this is to use a combination of approaches: Plain Text Use spaCy for general entity extraction. Use rule-based patterns for known engineering relationships. Use an LLM only when the relationship cannot be extracted reliably with simpler methods. Using a combination of these methods provides a way to keep costs and complexity at reasonable levels. How This Connects to GraphRAG Pipeline GraphRAG (graph retrieval-augmented generation) uses a graph during the retrieval phase prior to generating an answer using an LLM. Markdown Documents | Entity and relationship extraction | Knowledge graph | Graph traversal | Supporting text | Large Language Model | Answer This article implements the graph construction and traversal steps only; LLM integration is outside the scope of this example. The graph is used instead of traditional RAG in many cases. However, it is used most often when the question relates to entities that have some form of relationship to one another. Good fit: Plain Text Which services are indirectly affected by a Payment API outage? Usually not worth the extra complexity: Plain Text What is the timeout value for Payment API? For simple fact lookup, vector search may be enough. For dependency, ownership, impact analysis, and multi-hop questions, graph retrieval can add real value. Complete Example Here is the complete script: Python import json import re import networkx as nx documents = [ "Checkout Service depends on Payment API.", "Payment API uses PostgreSQL.", "Platform Team owns Payment API.", "Recommendation Service calls Catalog API.", "Catalog API uses Elasticsearch.", "Search Team owns Catalog API.", ] RELATION_PATTERNS = [ (r"(.+?) depends on (.+?)\.", "DEPENDS_ON"), (r"(.+?) uses (.+?)\.", "USES"), (r"(.+?) owns (.+?)\.", "OWNS"), (r"(.+?) calls (.+?)\.", "CALLS"), ] def normalize_entity(value): return value.strip() def extract_triples(text): triples = [] for pattern, relation in RELATION_PATTERNS: match = re.match(pattern, text, re.IGNORECASE) if match: subject = normalize_entity(match.group(1)) object_ = normalize_entity(match.group(2)) triples.append((subject, relation, object_)) return triples def build_knowledge_graph(documents): graph = nx.DiGraph() for doc in documents: triples = extract_triples(doc) for subject, relation, object_ in triples: graph.add_node(subject) graph.add_node(object_) graph.add_edge(subject, object_, relation=relation, source_text=doc) return graph def find_owner_of_dependency(graph, service_name): results = [] for dependency in graph.successors(service_name): edge_data = graph.get_edge_data(service_name, dependency) if edge_data["relation"] != "DEPENDS_ON": continue for possible_owner in graph.predecessors(dependency): owner_edge = graph.get_edge_data(possible_owner, dependency) if owner_edge["relation"] == "OWNS": results.append( { "service": service_name, "dependency": dependency, "owner": possible_owner, } ) return results def export_graph(graph, output_path): data = { "nodes": list(graph.nodes()), "edges": [ { "source": source, "target": target, "relation": edge_data["relation"], "source_text": edge_data["source_text"], } for source, target, edge_data in graph.edges(data=True) ], } with open(output_path, "w", encoding="utf-8") as file: json.dump(data, file, indent=2) if __name__ == "__main__": graph = build_knowledge_graph(documents) for source, target, data in graph.edges(data=True): print(f"{source} --{data['relation']}--> {target}") owners = find_owner_of_dependency(graph, "Checkout Service") for item in owners: print( f"{item['owner']} owns {item['dependency']}, " f"which is used by {item['service']}." ) export_graph(graph, "graph.json") Run it: Shell python build_graph.py Production Considerations This example is intentionally small. In a real system, the hardest part is not creating the graph. It is keeping the graph clean. A few things matter quickly: Entity normalization: Payment API, payment-api, and Payments API may all refer to the same system.Relationship quality: Bad relationships are worse than missing relationships because they lead retrieval in the wrong direction.Source tracking: Every edge should preserve where it came from. This helps with debugging, trust, and answer citation.Incremental updates: Rebuilding the entire graph every time a document changes is usually wasteful.Storage choice: NetworkX is excellent for local prototypes. For larger graphs, use a graph database such as Neo4j or another persistent graph store. Key Takeaways A knowledge graph is a practical way to represent relationships hidden inside documents. You do not need a complex architecture to get started. A small Python script can extract triples, build a graph, and answer multi-hop questions. Graph-based retrieval is most useful when the answer depends on connections between entities. It is less useful for simple lookup questions where traditional search already works well. The foundation of a good GraphRAG system is not the LLM prompt. It is the quality of the entities, relationships, and supporting evidence in the graph. Try It Yourself Add these two documents: Plain Text Checkout Service runs on Kubernetes. Platform Team manages Kubernetes. Then add a new relationship type called RUNS_ON. Update the query function to answer: Plain Text Who manages the platform that Checkout Service runs on? This small exercise will help you see why graph traversal becomes useful as relationships grow.

The Production Story Several years ago, my team made a decision that felt obviously correct: If a downstream call fails, retry it. More retries, more resilience. We set three retries on every integration touching our order-fulfillment platform, shipped it on a Thursday, and went home feeling good about our reliability posture. Six weeks later, retries were the single largest source of traffic in the platform. It surfaced during a routine inventory-sync slowdown. The inventory service got a little sluggish. Nothing dramatic. p99 crept from 200ms to maybe 1.2s. Our order API, sitting one layer up, started timing out and retrying. Three times each. The MuleSoft layer feeding the order API also had retries configured, so it retried the retries. By the time traffic reached the already-struggling inventory service, a single user click had turned into somewhere between nine and twenty-seven backend calls. The inventory service didn't recover. It got buried. We took a partial outage caused entirely by our own retry logic trying to save us. Figure 1: Retry amplification across API layers. One client request fans out to as many as 27 backend calls, while a retry budget keeps the downstream bounded. Why It Happened This part isn't obvious until you've watched it happen. Each retry decision was reasonable on its own, but together they were tearing the platform apart. Every team had configured retries looking only at their own layer. Three retries seemed fine in isolation. The trouble is that retries multiply across layers, and nobody owned the end-to-end number. Two retries here, three there, and suddenly one click is nine calls. The math was sitting in plain sight, and none of us had done it. What really bit us was how retries behave during a partial outage. When a downstream is healthy, retries are cheap, because failures are rare. When it's degraded, which is exactly when you're retrying the most, those retries pile more load onto a service that's already on its knees. So the system gets most aggressive at the worst possible moment. That's a feedback loop, and feedback loops like this end in outages. I've written before about retry storms and bounded reliability, and about how AI-generated DataWeave can fail quietly in production. This is the same family of problem. No single component is broken here. What's missing is a limit that spans the whole call path. Retries without a budget are really just a slow, polite way to DDoS yourself. The Bad Implementation This is what almost everyone ships first. I've shipped it myself. Java @Retryable( value = { RemoteServiceException.class }, maxAttempts = 3, backoff = @Backoff(delay = 200, multiplier = 2) ) public InventoryResponse checkInventory(String sku) { return inventoryClient.get(sku); } At first glance, this looks reasonable. Exponential backoff, a sane attempt count, a typed exception. Code review passes in thirty seconds. The problem is there's no awareness of anything beyond this one method. It retries during a full downstream outage just as eagerly as during a one-off network blip. It has no idea that the caller above it is also retrying. Worse, it happily retries errors that will never succeed. A 400, a validation failure, a duplicate-order rejection. You're burning retries on requests that were dead on arrival. DimensionBad (Naive Retry)Good (Budgeted Retry)Retry triggerAny failureOnly retryable failuresLimitPer-call attempt countFraction of total trafficBehavior under outageAmplifies loadSheds retries, stays boundedCross-layer awarenessNoneBudget shared end-to-endFailure modeRetry stormGraceful degradation The Good Implementation A retry budget flips the control. Instead of asking "how many times should this one call retry," you ask "what fraction of my total traffic is allowed to be retries?" The rule of thumb that's served me well: retries should never exceed 10% of your real request volume. If more than one in ten requests is a retry, something is genuinely broken, and hammering it harder won't fix it. It'll only dig the hole deeper. Here's a token-bucket budget that enforces this. Successful calls slowly refill the budget; each retry spends from it. When the budget is empty, you stop retrying and fail fast. Java public class RetryBudget { private final double retryRatio; // e.g. 0.10 = 10% private final AtomicLong tokens = new AtomicLong(); private final long maxTokens; public RetryBudget(double retryRatio, long maxTokens) { this.retryRatio = retryRatio; this.maxTokens = maxTokens; } // Every real request deposits a little budget back. public void onRequest() { tokens.updateAndGet(t -> Math.min(maxTokens, t + (long)(retryRatio * 100))); } // A retry is only allowed if the budget can pay for it. public boolean tryRetry() { return tokens.updateAndGet(t -> t >= 100 ? t - 100 : t) >= 0 && tokens.get() >= 0 && spend(); } private boolean spend() { return tokens.getAndUpdate(t -> Math.max(0, t - 100)) >= 100; } } The key behavior: under normal load, the budget stays full, and retries work as expected. Under a real outage, failures outpace successes, the budget drains, retries stop, and you protect the downstream instead of finishing it off. Wiring it into a Spring Boot client looks like this. Notice the two gates before any retry happens. The error has to be retryable, and the budget has to allow it. Java public InventoryResponse checkInventory(String sku) { budget.onRequest(); try { return inventoryClient.get(sku); } catch (RemoteServiceException ex) { if (isRetryable(ex) && budget.tryRetry()) { return inventoryClient.get(sku); // single budgeted retry } throw ex; // fail fast, don't amplify } } private boolean isRetryable(RemoteServiceException ex) { int code = ex.statusCode(); return code == 502 || code == 503 || code == 504 || code == 429; } Not every error deserves a retry. This distinction matters more than the budget math, because retrying a non-retryable error is pure waste. ErrorRetryable?Why503 Service UnavailableYesTransient, likely to clear504 Gateway TimeoutYesDownstream slow, may recover429 Too Many RequestsYes, with backoffHonor Retry-After, slow down502 Bad GatewayYesUsually transient routing issue400 Bad RequestNoRequest is malformed, will always fail401 / 403NoAuth won't fix itself on retry409 Conflict (duplicate order)NoRetrying creates a real data problem422 Validation ErrorNoDeterministic rejection The Architecture Pattern In MuleSoft, the same idea applies, and it's where I see the most damage because retries get configured at multiple layers without anyone counting. Keep the platform-level retry shallow and let your error type drive the decision. XML <until-successful maxRetries="1" millisBetweenRetries="500" doc:name="Budgeted Retry"> <http:request method="GET" config-ref="Inventory_HTTP" path="/inventory/{sku}"/> </until-successful> Then classify errors in DataWeave so the flow only retries what's worth retrying, and so a budget breach degrades cleanly rather than throwing: Shell %dw 2.0 output application/json var retryable = [502, 503, 504, 429] --- { shouldRetry: retryable contains payload.statusCode, action: if (retryable contains payload.statusCode) "RETRY_IF_BUDGET" else "FAIL_FAST" } The surprising part, when we rolled this out, was how rarely the budget actually engaged. Under healthy conditions, you'd never know it's there. It only shows its value during the bad fifteen minutes that used to turn into a bad three hours. Figure 2: The retry budget as a token bucket. Successful requests refill it, retries drain it, and once it falls below the 10% line, retries are disabled, and calls fail fast. A Real Production Example Picture a payment-processing flow calling an external gateway, with order-fulfillment and a Salesforce sync downstream. The gateway has a rough afternoon and starts returning intermittent 503s. Without a budget: every failed charge retries three times, order-fulfillment retries the payment call, and the Salesforce sync retries too. The gateway, already wobbling, gets three-to-nine times its normal load and falls over completely. A partial degradation becomes a full payment outage during peak hours. With a 10% budget: the first wave of retries is absorbed normally. As 503s climb, the budget drains within seconds. Retries stop, failed charges fail fast with a clear error, and customers see a retry-later message instead of a spinner. The gateway gets breathing room and comes back on its own. You take a small, honest failure now instead of a much bigger one you caused yourself. Metrics That Matter A budget you can't see is a budget you won't trust. These are the four numbers I put on a dashboard before I ship any retry change to production. MetricWhat it tells youHealthy rangeRetry ratio (retries / total requests)Whether retries are amplifying load< 10%Budget exhaustion eventsHow often the brake engagesRare, spikes during incidentsRetry success rateWhether retries actually help> 50%; if low, stop retryingDownstream p99 during retriesWhether you're worsening the outageShould not climb with retries If your retry success rate is low, that's the tell. You're retrying things that were never going to succeed, and the budget is doing you a favor by cutting them off. Monday-Morning Checklist Count your real end-to-end retry multiplier across every layer, not per service.Set a retry budget at roughly 10% of traffic and enforce it with a token bucket.Classify every downstream error as retryable or not, and never retry 4xx except 429.Cap retries to a single attempt at most layers; let the budget, not the attempt count, be your safety limit.Honor Retry-After on 429s instead of guessing backoff.Put retry ratio and budget-exhaustion metrics on a dashboard before you ship.Test it: degrade a downstream in staging and confirm retries actually stop. Final Thoughts Retries feel like reliability. Really, they're a loan against your downstream's capacity, and like any loan, they're cheap when you don't need them and expensive at the worst possible time. What makes the retry budget useful is that it ties retrying to the one number that should govern it: how much real traffic you're actually serving. I've made this mistake myself, and I've watched sharp teams make it too, because every individual decision looked correct in review. The fix isn't fancier backoff or smarter jitter. It's a ceiling. Decide up front how much of your traffic you'll allow to be retries, then hold that line even when every instinct is screaming at you to push harder. Especially then.

As a data engineer, I’ve noticed business teams submitting intake forms, compliance documents, and project proposals that a tech team then manually validates against a set of predefined business rules stored in a database that gets updated quarterly. The time it takes to validate a single form is typically in the hours, and by the time you’ve validated the form, the submitter has moved on to other work. When I needed to validate project intake forms against 60+ business rules of financial, compliance, and other types of business rules and guidelines (some of them to be used in a deterministic way and others to be used in a more nuanced manner), I knew that a simple if-else logic-based manual review process would not scale. This article walks through how I developed an async, AI-powered validation API with AWS Bedrock Agents and Serverless Architecture to process and validate intake forms within 60 seconds without blocking the user. The architecture also manages cross-account authentication to get access to the AI-powered engine and shows failure recovery gracefully. Why Async? The Problem With Synchronous AI APIs Integrating AI into an API synchronously means users send a request, the server processes it, and returns results in one HTTP response, but many systems that use AI-powered validation take more than 30 seconds. The AI agent I built was taking anywhere from 30 seconds to 1 minute to evaluate all of the form fields for all the applicable rules and conditions. But the hard limit for the API Gateway is 29 seconds (HTTP timeout). One approach to make this API request work is to transform the synchronous request and response into an async request with a subsequent background processing step and poll the results from a separate endpoint. This can be implemented as follows: Client submits the form via POST, receives a request_id immediately (under 2 seconds)Validation runs asynchronously in the background (30–60 seconds)Client polls a GET endpoint with the request_id until results are ready By making the form submission step separate from the AI validation of that form in the background, users can continue working on other tasks instead of being stuck staring at a page waiting 30 to 60 seconds for the form to be validated. Architecture Overview As a data engineer, I was required to tackle three main challenges to create a production AI validation API: 1) the frontend application is deployed in a different AWS account, 2) AI agent-based form validation is extremely computationally expensive to run, and 3) business rules for this type of validation are likely to change from time to time without API code deployment. The architecture consists of five components: API Gateway (REST API): With Cognito Authorizer for cross-account JWT authenticationAsync Handler Lambda: It’s an entry point for the API. An Async Handler Lambda function is invoked by a POST request. It will store the form payload on S3, then trigger the Validation Lambda function and store an initial "processing" status in S3. The function immediately returns a request_id to the frontend client within 2 seconds.Validation Lambda: This function loads up all the rules for a given request from S3. It then builds up all the prompts for the Bedrock Agent and runs the Agent. The results of the Agent are then saved off in S3 for the Polling API.Polling Lambda: Handles GET requests and checks S3 for completed resultsRules Sync Lambda: Separate independent process to read validation rules from the data warehouse using EventBridge scheduler and sync to S3 for validation with AI model. Implementation: The Async Handler The async handler is the entry point. Its task is quite straightforward. It accepts the payload, stores it, triggers the Validation Lambda function, stores an initial "processing" status in S3, and returns the “processing” status with the request ID to the client. The function does all of this within a couple of seconds. Here is the core implementation: Python import json, boto3, uuid from datetime import datetime s3 = boto3. client(' s3') Lambda_client = boto3. client('Lambda') S3_BUCKET = 'my-validation-bucket' VALIDATION_LAMBDA = 'ai-validation-function' def lambda_handler(event, context): payload = json. loads (event. get ('body', "{}')) request_id = str(uuid.uuid4()) # Store initial processing status s3.put_object( Bucket=S3_BUCKET, Key=f'validation-output/(request_id)/status.json', Body=json.dumps({ 'request_id': request_id, 'status': 'processing', 'submitted_at': datetime. ttenew() .isoformat() }) ) # Fire-and-forget: invoke validation async pay Load ['_request_id'] = request_id lambda_client.invoke( FunctionName=VALIDATION_LAMBDA, InvocationType='Event', # Async invocation Payload=json. dumps (payload) ) return { 'statusCode': 202, 'body': json. dumps ({ 'request_id': request_id, 'status': 'processing' }) } In the above code snippet, I specifically invoke the validation lambda from the async handler by setting the InvocationType='Event'. This allows the async handler to return immediately to the frontend with the request_id for the submitted request. The Validation Lambda will then complete asynchronously and store the results in S3. Implementation: The Polling Handler The Polling Handler Lambda function manages the GET endpoint; it polls S3 for the updated status file and returns the current status of Validation Lambda processing: completed or failed. Here is the core implementation: Python def lambda_handler(event, context): request_id = event['pathParameters']['request_id'] try: status_obj = s3.get_object( Bucket=S3_BUCKET, Key=f'validation-output/{request_id}/status.json' ) status = json.loads(status_obj['Body'].read()) if status['status'] == 'processing': return {'statusCode': 200, 'body': json.dumps(status)} # Completed - return full results results_obj = s3.get_object( Bucket=S3_BUCKET, Key=f'validation-output/{request_id}/results.json' ) results = json.loads(results_obj['Body'].read()) return {'statusCode': 200, 'body': json.dumps(results)} except s3.exceptions.NoSuchKey: return {'statusCode': 404, 'body': 'Request not found'} S3 Decoupling: Using S3 as an intermediary between the validation Lambda and the polling handler allows for natural decoupling. The validation Lambda writes the results of the validation to S3, and the polling handler reads from S3 to return the latest status to the frontend. There is no shared state between the validation handler and the polling handler; there are no database connections, and there are no race conditions. Integrating the Bedrock Agent for Intelligent Validation An intelligent validation function would need more than just a set of rules to check for requirements and best practices. There are a lot of judgment calls that a human would make based on examples of how a policy or guideline would be applied in real life. To achieve that, the more effective way is to integrate with an existing AI function that is designed to handle a wide variety of scenarios and functions The Bedrock Agent architecture solved this by combining: Knowledge base: Containing policy documents, guidelines, and past examples of work for the intelligent validation to reference during the evaluation process.Dynamic prompts: The prompts for the AI model are built dynamically from the current validation rules. These are loaded from S3 as a JSON file and then injected with the current values for the specific field being evaluated.Structured output: Parse the assessment’s pass/fail status, confidence in the assessment, and a set of detailed recommendations made by the agent. The prompt for the AI agent is generated at runtime by the validation function. The rules are loaded from S3 earlier in the function's execution. Here is an example prompt: “Evaluate field [Project Justification] with value [user input] against rule: The justification must clearly describe the business problem being solved and include quantified impact. Reference the knowledge base for examples of approved justifications.” The AI returns a structured assessment of whether or not the field has passed validation, the confidence that the AI has in the assessment, and recommendations. Dynamic Rules Management: Keeping Rules in Sync Without Code Deploys Rules typically change on a monthly or quarterly basis by the business teams. To keep up with the current policy, the rules must be separate from the rest of the application code. To achieve that, I used Rules Sync Lambda, triggered daily by EventBridge: EventBridge fires at 6 AM daily.The Rules Sync Lambda queries the Data Warehouse (Redshift) for the current validation rules for the application.It also takes a copy of the most current version of the rules in S3 for purposes of rollback.It transforms and then uploads the new rules file to S3 as a new copy of the Validation_Rules.json file.Upon failure to update the rules in S3, a CloudWatch Alarm is triggered, which in turn triggers an SNS notification to the appropriate engineering team. The rules are managed as a database of rules (as opposed to being stored within the application code), which allows business analysts to easily update the rules on a quarterly basis without requiring any code changes or deployments. Cross-Account Authentication With Cognito In this case, the frontend application and the AI backend were set up in two different AWS accounts. When deployed within different accounts (as within an enterprise), cross-account authentication is required. Since the frontend application was already authenticated against a company’s SSO (Single Sign On) using Cognito, it was only a matter of how to reuse these tokens within another account without involving the Frontend team for changes. The solution was to create a Cognito Authorizer and attach it to a REST API created in the API Gateway. This API can then be set up to trust the User Pool from the frontend account. Below is a simplified representation of this configuration: API Gateway REST API with a Cognito Authorizer pointing to the frontend account’s Cognito User Pool ARN.CORS (Cross-Origin Resource Sharing) configuration for only that frontend domain.The frontend application is already authenticated with CognitoThe backend application accepts the tokens that the frontend application is using for authenticationThe frontend application simply sends the existing Cognito tokens that the frontend application already has created in the authentication process From the frontend team’s perspective, this was a simple implementation that required them to send the existing Cognito token with the request and to implement a polling loop for the GET endpoint. Results and Lessons Learned After deploying to production: Validation time: reduced from 2 -3 hours (manual) to less than a minute (automated)API response time for form submission: less than 2 seconds for GET API using an async pattern, meaning the frontend never has to wait for the backend60+ validation rules: per form, including both deterministic and AI-judgement rules Zero code deploys: for changes to the rules, which are stored in the database, sync daily Key lessons as a developer building this: Design for async from the start: Retrofitting a synchronous API to be async is very hard. If your AI inference takes more than 5 seconds, which is generally the case, then design your API to be async from day one.Use S3 as your state machine: S3 is the simplest, cheapest, and most reliable way to pass results between decoupled Lambdas. No databases, no queues, no DynamoDB for this pattern.Separate dynamic rules from code: Separate process for managing rules which are dynamic and change often to avoid deployment bottleneck Bedrock Agents are good for making judgment calls. If you have a deterministic check (is a field empty), then you can code that. But for a judgment call (does a justification make sense), then use an AI agent to make the call. Conclusion There is an entirely new way to approach the request lifecycle for APIs in this AI-powered validation API development. The asynchronous API with polling for validation is better than simply trying to work around the timeout limits of APIs. Bedrock Agents, along with S3 to manage the state of the workflow and EventBridge to synchronize rules on a daily basis from a database created by business users via a simple UI created by frontend team, while backend team does not need to write any code for new rules, all integrated together to form complex data validation system powered by AI-powered judgment calls while maintaining simple to deploy and scalable system. As a data engineer, there’s nothing quite like watching hours of manual work by a reviewer get compressed down into 60 seconds or less of automated work while maintaining the high level of evaluation that a business stakeholder expects.

In a lot of organizations, the real integration platform is a person. Someone exports orders from the ERP every morning and pastes them into the planning tool. Someone else re-types customer updates from the CRM into the invoicing system. It works until that person is on holiday or makes a typo in a price field or the volume doubles. Replacing that manual work with a synchronization service sounds like a junior-level task: read from system A, write to system B, schedule it, done. In practice, sync services are where many integration projects quietly fail. They fail not because moving data is hard, but because the edge cases are partial failures, retries that duplicate records, two systems that both think they own a field, and errors that nobody notices for three weeks. This article walks through the design decisions that separate a sync layer you can trust from one you learn to fear. The examples use Python and pseudo-SQL, but every pattern here is language-agnostic. Decision 1: One Source of Truth Per Entity The single most important design decision in any sync architecture is not technical. It is organizational: for every entity, exactly one system is allowed to win. Orders live in the ERP. The webshop may create them, but once created, the ERP's version is the truth, and the webshop displays what the ERP says. Customer contact details live in the CRM. The ERP receives updates from the CRM and never edits them locally. The moment two systems can both modify the same entity and both push their version, you have built a conflict generator. Last-write-wins will silently destroy data. Merge logic will grow into an unmaintainable swamp of special cases. The fix is almost never smarter conflict resolution. It is removing the conflict by assigning ownership. Write this down as a table before writing any code: EntityOwnerMay createMay updateOrderERPWebshop, ERPERP onlyCustomer contactCRMCRMCRM onlyProduct/pricingERPERPERP onlyStock levelERPERPERP only If you cannot fill in this table, you are not ready to build the sync. Any cell where two systems appear in the "may update" column is a design problem to resolve with the business first, not a technical challenge to code around. Decision 2: Idempotency, or Retries Will Hurt You Your sync will fail mid-run. The network will drop after 4,000 of 5,000 records. The target API will return a 500 halfway through. The scheduler will fire twice. None of these are exceptional; they are Tuesday. The only sane response to failure is retry, and retry is only safe when every operation is idempotent: running it twice produces the same result as running it once. The classic mistake looks like this: Python # Dangerous: creates a duplicate on every retry def sync_order(order): target_api.create_order( customer=order.customer_id, lines=order.lines, total=order.total, ) If this call succeeds on the target but the response is lost (a timeout, a crashed worker), the retry creates a second order. Someone ships it. The fix is to make every write carry a stable, deterministic key derived from the source record, and make the target treat that key as unique: Python # Safe: the natural key makes the operation idempotent def sync_order(order): target_api.upsert_order( external_id=f"erp-{order.erp_id}", # stable key from the source customer=order.customer_id, lines=order.lines, total=order.total, ) If the target system has no upsert endpoint, simulate one: look up by external_id first, then create or update. Wrap that lookup-and-write in one function and forbid every other code path from writing directly. The same rule applies to your own bookkeeping. Store sync state keyed by the same external ID, so a re-run of yesterday's batch is harmless by construction. Decision 3: Pull Changes, Don't Diff Worlds The naive sync reads all records from both sides and compares them. This works in the demo and collapses in production, where "all records" means 400,000 rows over a SOAP API that pages 100 at a time. You need change detection, and there are three workable tiers, in order of preference: The source has reliable updated_at timestamps or a change log. Store a high-water mark after each successful run and query only what changed since. This is the happy path; verify that the timestamp actually updates on every mutation, including the ones done by nightly batch jobs inside the legacy system. Legacy systems lie about this more often than you would expect.The source has no usable timestamps, but you can read all records cheaply. Compute a hash per record and compare against the hash you stored last run. Only records with changed hashes get pushed downstream: Python import hashlib, json def record_hash(record: dict) -> str: canonical = json.dumps(record, sort_keys=True, default=str) return hashlib.sha256(canonical.encode()).hexdigest() def detect_changes(records, stored_hashes): for r in records: h = record_hash(r) if stored_hashes.get(r["id"]) != h: yield r, h Neither is possible. You are down to full comparisons on a schedule. Constrain the entity scope aggressively and be honest with stakeholders about latency. Whichever tier you land on, keep the change detector separate from the writer. A queue between them, even a simple database table with pending / done / failed states, gives you retry, rate limiting, and an audit trail almost for free: SQL CREATE TABLE sync_queue ( id BIGSERIAL PRIMARY KEY, entity_type TEXT NOT NULL, external_id TEXT NOT NULL, payload JSONB NOT NULL, status TEXT NOT NULL DEFAULT 'pending', attempts INT NOT NULL DEFAULT 0, last_error TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), processed_at TIMESTAMPTZ, UNIQUE (entity_type, external_id, status) ); That UNIQUE constraint is doing real work: it prevents the same pending change from being enqueued twice, which keeps the queue idempotent too. Decision 4: A Silent Sync Is Worse Than No Sync Here is the paradox of a working sync layer: the better it works, the more people trust it, and the more damage it does on the day it silently stops. A sync that visibly fails gets fixed the same morning. A sync that dies quietly keeps its consumers confidently reading stale data. Sales quotes yesterday's* stock levels. Finance invoices from last week's prices. By the time someone notices, you are reconstructing three weeks of drift. Minimum viable observability for a sync service is four things: A heartbeat. Every run writes a completion record. An external check alerts when the most recent successful run is older than the expected interval. Do not rely on the sync alerting about itself; a crashed process sends no alerts.Drift metrics. Periodically count records on both sides and compare. The counts will never match perfectly in a live system, so alert on trend, not on exact equality.A dead-letter state. After N failed attempts, a queue item moves to failed and a human is notified with the payload and the last error. Infinite retry loops on a permanently broken record will otherwise clog the queue and mask new failures behind old ones.Readable logs per record. When finance asks why invoice 4482 shows the old address, you want to answer with one query, not a debugging session. None of this is sophisticated. All of it is regularly skipped, because on the day the sync ships, it works, and observability feels like polish. It is not polish. It is the feature that determines whether you find out about failure from a dashboard or from an angry customer. The Shape of the Whole Thing Put together, a trustworthy sync layer is small and boring: Plain Text [ Source system ] → change detection → sync_queue → idempotent writer → [ Target system ] ↓ heartbeat, drift checks, dead letters Two processes, one queue table, a handful of metrics. The value is not in the volume of code; most implementations of this design fit in a few hundred lines. The value is in the decisions encoded in it: one owner per entity, stable keys on every write, changes flowing through an inspectable queue, and failure treated as a normal input rather than an exception. Build it this way, and the sync becomes infrastructure nobody thinks about, which is the highest compliment integration code can receive. Build it as a quick script, and you have not removed the human integration layer at all. You have just changed whose Friday afternoon gets ruined.

Last year, I was working on deploying an agentic AI system to help manage cloud infrastructure at scale. The idea was straightforward: give the agent access to AWS APIs, let it observe infrastructure state, and allow it to take remediation actions autonomously. Scale a deployment here, restart a service there, update a configuration when metrics cross a threshold. What I did not fully appreciate at the time was how differently an AI agent fails compared to a traditional automation script. When a shell script goes wrong, it fails in a bounded, diagnosable way. You get an error code. You trace it. You fix it. When an agentic AI system fails, it can fail in ways you never anticipated, hallucinating resource states, misinterpreting instruction scope, or acting on adversarial inputs buried in a monitoring alert. These failures do not produce clean stack traces. They produce production damage. That realization sent me down a path of building a guardrail system. What I eventually learned, and this took four calibration cycles to prove empirically, is that no single guardrail layer can solve this problem. You need multiple complementary layers, and you need to design them to compensate for each other's blind spots. Here is what I built, what broke along the way, and what I would do differently from the start. Why the Obvious Solutions Did Not Work My first instinct was to use AWS Bedrock Guardrails. Configure a topic denial policy for destructive operations, set the content filters to HIGH, block PII like access keys. Simple, managed, done. I ran it against 100 representative agent prompts, a mix of read operations, staging changes, risky production changes, destructive operations, and adversarial jailbreak variants. The result stopped me cold. Tuned for zero false negatives, meaning I wanted to catch every genuinely dangerous action, the guardrail produced a 40% false positive rate. It was blocking list operations. It was blocking staging scale-outs. It was blocking service configuration updates that had nothing to do with deletion or destruction. That is not a deployable guardrail. That is a system that would make the AI agent useless within a day. The second problem was structural, not tuning-related. A Bedrock guardrail intercepts the model's text output. But an agent does not only produce text; it invokes tool calls. An agent can generate a perfectly compliant response like "I will scale the deployment safely" and then immediately invoke a delete API as a tool call. The guardrail never sees the tool call. It evaluated the wrong boundary. The third issue came when I looked at policy-as-code frameworks. OPA with Gatekeeper is excellent at Kubernetes admission time, evaluating manifests before they are deployed. But a DevOps agent is not deploying manifests. It is generating action proposals at runtime against live infrastructure that changes by the hour. A static policy that denies writes to "production resources" is useless unless it knows, at this exact moment, which resources are tagged as production. That information is not in a manifest. It is in live EC2 tags pulled from the AWS API. These were not flaws in the tools. There were boundary mismatches. Each tool was designed for a different problem. None of them was designed for the problem of governing an autonomous agent at the tool-call execution boundary. The Architecture I Landed On After a lot of iteration, I settled on a five-layer pipeline that intercepts at the tool-call boundary the moment the agent transitions from thinking to acting. Any layer can terminate the pipeline. The default is blocked. Blast-Radius Scoring Before any layer fires, every proposed action gets a blast-radius score between 0 and 1. Read-only operations (list, get, describe, monitor) score 0.1. Reversible mutations (restart, scale, update, patch) score 0.4. Potentially irreversible operations (delete, destroy, terminate, purge, wipe) score 0.9. This score drives routing. Low-risk actions auto-approve without touching the full pipeline. High-risk actions require human approval. Everything in between goes through policy evaluation. This is what keeps the system from adding 8 seconds of latency to every "list all EC2 instances" call. Layer 1: Bedrock Guardrail With a Bypass I kept Bedrock Guardrails as the first layer but added something critical: a selective bypass for low-risk and staging operations. Before the guardrail fires, the pipeline checks the blast-radius score and environment context. If the action is read-only or explicitly targeting a staging environment, the guardrail is skipped entirely. This one change took the false positive rate from 40% down to 18%. That occurred not by tuning the guardrail, but by changing the architecture around it. Layer 2: OPA Against Live State The second layer runs Open Policy Agent, but not against a static manifest. It pulls live AWS context via boto3 immediately before each evaluation: EC2 inventory with environment tags, S3 buckets, IAM roles. That live context becomes part of the input document that OPA evaluates. Now the policy can answer the question that actually matters: "Is this specific resource, right now, a production resource?" A rule that reads is_production(resource)` checks the live tag, not a manifest field. This is what catches the actions that Layer 1 misses, like "purge all messages from the SQS queue," that use vocabulary outside the guardrail's topic examples but clearly target production infrastructure. In my evaluation, Layer 2 was the sole blocking layer for 43% of correctly blocked prompts. Without it, those would have been false negatives. Layer 3: Confidence Delta Scorer This layer does something a little unconventional. It invokes the model twice, once with the guardrail attached and once without, and measures how much the guardrail changed the model's response. It looks at hedging vocabulary frequency: words like "should," "might," "could," "consider," "careful." A big divergence between the guardrail and unguarded responses means the guardrail substantially changed the model's behavior. That is a signal; the action is near the policy boundary and worth extra scrutiny. It is a probabilistic safety margin for borderline cases. Layer 4: HITL Gate With DynamoDB TTL For high-risk actions, a Lambda function writes a DynamoDB record with a 5-minute TTL and sends an SNS email with approve and reject URLs. The design principle that matters here: **silence equals rejection**. If the TTL expires without a response, DynamoDB automatically deletes the record, and the pipeline denies the action. An unavailable approver never produces an implicit approval. I built this using DynamoDB's native TTL feature, which means there is no polling loop, no Lambda checking every 30 seconds whether the token expired. It just works. Layer 5: Audit and Rollback Every pipeline execution writes a structured JSON record to CloudWatch: the prompt, risk classification, blast-radius score, per-layer decisions, latency, cost, and outcome. A CloudWatch alarm fires if the block rate drops below 70% in any 5-minute window, which would indicate something has degraded in the guardrail configuration. Before every approved action, boto3 captures a snapshot of the relevant resource state to S3. If health metrics degrade within 5 minutes after execution, automated rollback restores the pre-action state from that snapshot. What the Numbers Actually Showed I ran the complete pipeline against 100 prompts on live AWS infrastructure in us-east-1. Here is what came back: CategoryAccuracyFP RateFN RateAvg LatencyRead operations (20)95%5%0%910msSafe staging changes (20)100%0%0%7,920msRisky production changes (20)95%0%5%13,949msDestructive operations (20)100%0%0%9,590msAdversarial jailbreaks (20)90%0%10%8,125msTotal (100)96%1%3%8,099ms Three things stood out. Zero false negatives on destructive operations: Not one delete, destroy, terminate, or purge operation made it through. That was the most important number for me.22x latency reduction for blocked actions: Intercepted actions averaged 347ms. Allowed actions averaged 8,099ms. Because Layer 1 fires before the full LLM invocation, blocking a harmful action is actually faster than allowing a safe one. The safety mechanism is also a performance optimization for the adversarial case.Total cost: $0.0017 USD for 100 prompts: At 10,000 agent actions per day, that is roughly $0.17 daily. The cost of not having this is measured in incidents. The Calibration Study Nobody Talks About The finding I keep coming back to is the calibration progression: VersionAccuracyFP RateWhat Changedv1: Single Bedrock guardrail60%40%Baselinev2: Added low-risk bypass79%18%Architectural changev3: Added staging context in OPA89%8%Live state integrationv4: Expanded service config keywords96%1%Allow-list expansion What strikes me is that each improvement required a fundamentally different mechanism. The bypass addressed a structural mismatch. The staging context detection required live infrastructure data that no static guardrail can access. The keyword expansion fixed a vocabulary coverage gap. None of these is achievable by turning a dial on a single layer. This is the empirical case for layered defense-in-depth. Not as a philosophical preference. As a measurable engineering necessity. Practical Takeaways If you are building agentic DevOps tooling, here is what I would tell myself from a year ago: Intercept at the execution boundary: Your safety mechanism must fire when the agent calls a tool, not when it generates text.Pull live state before every policy evaluation: A policy that cannot see which resources are actually in production right now is not protecting production.Make your HITL gate fail closed: Design it so an unresponsive approver produces a denial, not a permit. DynamoDB TTL handles this elegantly without polling.Run your calibration study before going live: Measure FP and FN rates separately. They trade off against each other in ways that are not obvious until you measure them.Snapshot before every approved action: Automated rollback is not glamorous, but it is the safety net you will want when something approved turns out to be harmful. The Code Everything described here is open source: https://github.com/ManvithaP-hub/agentic-devops-guardrails That includes the Lambda functions, OPA Rego policies, boto3 state fetching, DynamoDB approval gate, CloudWatch audit, and a Terraform deployment module. You can run the full evaluation on your own AWS account for under a dollar.

In Part 1, we solved one direction of the multi-cloud connectivity problem: a workload running in Google Cloud interacting with an AWS cloud resource. A GKE pod read a Google-issued OIDC token from the metadata server, handed it to AWS STS via AssumeRoleWithWebIdentity, and received short-lived AWS credentials, with no static access keys stored anywhere. MultiCloudJ wrapped the token dance behind a portable client so the application code never touched a provider SDK directly. This article covers the return trip: a workload running in AWS calling into Google Cloud — specifically, an Amazon EKS pod reading and writing a Google Cloud Storage (GCS) bucket — again with zero long-lived credentials. The zero-trust principle is identical. The mechanism is a bit different. And that asymmetry is the single most important thing to understand before you build it. Authentication Flow from AWS to GCP 1. Build a SigV4-signed GetCallerIdentity request (signed with STS credentials): It's assumed that the EKS pod already holds temporary AWS credentials. 2. Call sts.googleapis.com for token exchange: The pod sends that signed request to Google Cloud as the input to an OAuth 2.0 token exchange. It is asking Google, "Here is proof of who I am on AWS - please give me a Google token to access cloud resources." 3. Replay the GetCallerIdentity signed request: Google does not trust the request blindly. It runs the signed request against AWS STS on the caller's behalf. 4. Response with ARN: AWS checks the signature and replies with the caller's ARN (the AWS role identity) as part of the GetCallerIdentity response. Now Google knows exactly which AWS identity is asking - proven by the signature, with no shared secret. 5. Validate the ARN with the pool: Google checks that ARN against the Workload Identity Pool rules - which AWS account and which role are allowed in, and how the ARN maps to a Google identity. 6. Access token: Once the ARN passes, Google returns a short-lived access token to the EKS pod. 7. Access the resource with the access token: The pod uses that token to read and write Cloud Storage. When the token expires (usually within an hour), the flow repeats. Nothing long-lived is ever stored. Summary: AWS proves the pod's identity by answering Google's replayed request, and Google issues a short-lived token based on that proof. No access keys, no service-account key files - just a signed request and a temporary token crossing the trust boundary. Please note that this authentication flow can be used for any cloud service and is not specifically for cloud storage. Direct Pool Access vs. Service Account Impersonation Once Google has verified the caller's AWS identity through the signed request, it still has to map that AWS identity to something that actually holds permissions on the bucket. There are two ways to do this mapping, and you should pick one before you grant any IAM role. Option 1: Direct Pool Access You grant the Cloud Storage role straight to the federated identity. In IAM, the member looks like this: principalSet://iam.googleapis.com/projects/123456789/locations/global/workloadIdentityPools/aws-pool/* The permission sits on this pool principal, not on the AWS role. The AWS role never holds any GCP permission. Its only job is to prove identity: it answers Google's replayed GetCallerIdentity request so Google knows which AWS identity is asking. Google then checks that identity against the pool rules and, if it is allowed in, treats the caller as this pool principal. The bucket role, such as roles/storage.objectAdmin, is bound to that principal, so that is where the actual access comes from. No service account sits in the middle. In your code, the value you pass is the pool provider resource name (the audience), and Google issues a token that represents the pool identity directly. Option 2: Service Account Impersonation You create a GCP service account, grant that service account the bucket role, and then let the federated identity impersonate it. The federated identity needs roles/iam.serviceAccountTokenCreator on that service account, and the exchange gets a second hop: first a pool token, then an impersonated service-account token. In your code, the value you pass is the service account email. Which to Choose For a straight AWS EKS to GCS case like this one, direct pool access is the better default: Fewer moving parts. No service account to create, no token-creator grant to manage, and no second token hop.Tighter blast radius. The bucket permission is tied to identities coming through this specific pool, not to a service account that other workloads might also be able to impersonate. You can narrow it further to a single AWS role with an attribute condition on the principal.Less to audit. One IAM binding on the bucket tells the whole story. Reach for impersonation only when you actually need what a service account gives you: You must reuse an existing service account that already carries permissions across many GCP resources.A downstream Google API or tool only understands service-account identities and cannot evaluate a principalSet:// member.Your organization standardizes on service accounts as the single unit of access, to stay consistent with other human and machine grants. In short, direct pool access is simpler and safer, so use it unless a concrete requirement forces impersonation. Set Up Workload Identity Pool on GCP Before any code runs, you configure the trust relationship on Google Cloud once. Three things: a pool, an AWS provider inside it, and an IAM grant on the bucket. Create the Workload Identity Pool: The pool is the identity container that your AWS workloads will be represented as.gcloud iam workload-identity-pools create aws-pool --location="global" --display-name="AWS workloads"Create the AWS provider inside the pool: The provider is the entry gate. It tells Google to trust GetCallerIdentity results from a specific AWS account, how to map the caller's ARN into a Google attribute, and which callers are allowed in.Two important parts here: The attribute mapping turns the caller's raw ARN into a stable attribute.aws_role value with the session name stripped, so grants survive session rotation.The attribute condition is the first gate: only callers from your AWS account are admitted, before any IAM binding is even checked. Shell gcloud iam workload-identity-pools providers create-aws aws-provider \ --location="global" \ --workload-identity-pool="aws-pool" \ --account-id="123456789012" \ --attribute-mapping="google.subject=assertion.arn,attribute.aws_role=assertion.arn.contains('assumed-role') ? assertion.arn.extract('{account_arn}assumed-role/') + 'assumed-role/' + assertion.arn.extract('assumed-role/{role_name}/') : assertion.arn,attribute.account=assertion.account" \ --attribute-condition="assertion.account == '123456789012'" Grant the bucket role to the pool principal: This is the direct pool access model. The permission binds to the AWS role (via the mapped attribute), not to a service account. Shell gcloud storage buckets add-iam-policy-binding gs://my-archive-bucket \ --role="roles/storage.objectAdmin" \ --member="principalSet://iam.googleapis.com/projects/PROJECT_NUMBER/locations/global/workloadIdentityPools/aws-pool/attribute.aws_role/arn:aws:sts::123456789012:assumed-role/my-eks-role" After this, the EKS pod's role can federate into the pool and read/write the bucket, and the application code in the next section never touches any of this setup again. Implementation With MultiCloudJ MultiCloudJ exposes the same BucketClient abstraction you saw in Part 1; you build it for the "gcp" provider and attach a CredentialsOverrider that carries the federated identity. The library handles the SigV4 signing, the STS token exchange, and (on the impersonation path) the generateAccessToken call internally; your code just does blob operations (full example). Java private static final String REGION = "us-west-2"; // The audience is the full Workload Identity Pool provider resource name. // We grant the bucket role directly to this pool principal (direct pool // access), so no service account sits in the middle. private static final String AUDIENCE = "//iam.googleapis.com/projects/123456789/locations/global/workloadIdentityPools/aws-pool/providers/aws-provider"; // The supplier runs on every GCP token refresh. Each time it signs a fresh // GetCallerIdentity request with the pod's AWS role (IRSA, picked up from the // ambient AWS credential chain) and returns the subject token GCP expects. Supplier<String> webIdentityTokenSupplier = GcsFromAws::buildSubjectToken; CredentialsOverrider overrider = new CredentialsOverrider.Builder(CredentialsType.ASSUME_ROLE_WEB_IDENTITY) .withRole(AUDIENCE) .withWebIdentityTokenSupplier(webIdentityTokenSupplier) .build(); // Portable client: same API as the AWS side in Part 1, // only the provider string changes. BucketClient bucketClient = BucketClient.builder("gcp") .withBucket("my-archive-bucket") .withCredentialsOverrider(overrider) .build(); ListBlobsPageResponse page = bucketClient.listPage(ListBlobsPageRequest.builder().withMaxResults(10).build()); page.getBlobs().forEach(b -> System.out.println(b.getName())); // Signs a GetCallerIdentity request with the pod's AWS role, then shapes the // signed request into the URL-encoded JSON envelope that Google STS expects // as an AWS4 subject token. private static String buildSubjectToken() { // Google requires the audience to travel inside the signed headers, so it is // bound to the signature and the request cannot be replayed against any other // target. SignOptions options = SignOptions.builder() .withCustomHeader("x-goog-cloud-target-resource", AUDIENCE) .build(); StsUtilities stsUtil = StsUtilities.builder("aws").withRegion(REGION).build(); // Passing null means "just sign a GetCallerIdentity request, there is no // service payload to hash." The library fills in Action=GetCallerIdentity. SignedAuthRequest signed = stsUtil.newCloudNativeAuthSignedRequest(null, options); JsonObject envelope = .. // construct json object from signed request uri return URLEncoder.encode(envelope.toString(), StandardCharsets.UTF_8); } Conclusion Part 1 showed GCP calling AWS, and Part 2 completes the picture with AWS calling GCP. Both use the same idea: federation, no static keys, and only short-lived credentials. They differ only in how identity is proven. GCP to AWS presents a Google OAuth identity token, while AWS to GCP sends a signed request that GCP verifies with AWS. This is exactly where MultiCloudJ earns its place. All of these provider-specific differences, such as the bearer token here, the signed request and replay there, the STS token exchange, the service-account impersonation, and the token refresh, are abstracted away inside the library. You build one portable client, attach a credentials overrider, and call the API. Your application code never learns which cloud it is talking to or which way the call is going, so it stays clean, portable, and free of long-lived secrets in both directions.