Production Graph System: Capstone Project Build

Graph System Requirements: defining project scope

Complex software systems obscure their internal dependencies, making changes risky and impact analysis difficult. Microservice architectures, in particular, often lack a centralized, up-to-date view of service-to-service communication or library usage. This opacity hinders development velocity, complicates incident response, and slows security patching efforts.

The capstone project will build a production-grade graph system to model and query these relationships. Its primary function is to provide a single source of truth for dependencies across a distributed software ecosystem. This system will enable engineers to visualize connections and answer critical questions about system structure.

User stories define the system’s core interactions:

• As a developer, I need to see all direct upstream and downstream services for service-payment-gateway to understand its immediate impact radius.
• As a release engineer, I need to identify all services transitively dependent on library-crypto-v1.2 to assess the scope of a security vulnerability and plan a coordinated upgrade.
• As an SRE, I need to trace the full dependency chain for api-user-login to pinpoint potential bottlenecks or single points of failure.

Functional requirements include:

• Data Ingestion: The system must accept dependency data from various sources, such as build system outputs (e.g., pom.xml, package.json), configuration files, or runtime introspection agents. Input will be JSON or Protobuf messages defining nodes (services, libraries) and directed edges (depends-on, uses).
• Graph Storage: Store nodes and edges, supporting attributes on both (e.g., service owner, library version, dependency type). The graph must support efficient traversal.
• Query API: Provide an HTTP/gRPC API for querying. This includes direct neighbor queries (1-hop), pathfinding up to N hops, and filtering based on node/edge attributes.
• Update Mechanism: Support incremental updates to the graph, reflecting new dependencies or removals without full rebuilds.

Non-functional requirements dictate the system’s operational characteristics:

• Scalability: The system must handle a graph of up to 10,000 unique services and 100,000 directed dependencies. It should scale horizontally to accommodate growth.
• Latency: Direct neighbor queries must complete within 100 milliseconds. Transitive queries for paths up to 5 hops must complete within 500 milliseconds.
• Availability: The system must maintain 99.9% availability.
• Data Freshness: Dependency updates ingested into the system must be queryable within five minutes of receipt.

Architecture Design: how components integrate

The production graph system requires a resilient architecture to handle high-throughput ingestion, real-time processing, and low-latency query access. Data enters the system via Apache Kafka 3.x, acting as a durable message queue. This decouples data producers from the graph system, providing backpressure handling and enabling message replay for recovery or reprocessing.

Data streams from Kafka are processed by Apache Spark Streaming 3.x. This layer performs schema validation, denormalization, and transforms raw events into graph mutations—additions or updates of nodes and edges. Spark’s micro-batching capabilities ensure near real-time updates while handling large data volumes efficiently.

The processed graph mutations are written to JanusGraph 0.6, which uses ScyllaDB 5.x as its storage backend. JanusGraph provides a standardized Gremlin graph API, abstracting the underlying storage. ScyllaDB, a Cassandra-compatible NoSQL database, offers high throughput and low-latency reads and writes, scaling horizontally across commodity hardware. This combination provides a distributed, highly available graph store. The tradeoff is increased operational complexity managing both JanusGraph and ScyllaDB clusters compared to a single-node graph database.

Client applications interact with the graph system through a FastAPI service. This Python-based API layer exposes specific graph queries and mutations, ensuring controlled access and optimized query execution. For example, a common endpoint retrieves all neighbors of a given node:

# app/api/v1/endpoints/graph.py
from fastapi import APIRouter

router = APIRouter()

@router.get("/nodes/{node_id}/neighbors")
async def get_node_neighbors(node_id: str, relationship_type: str = None):
    """
    Retrieves neighbors for a given node, optionally filtered by relationship type.
    """
    # Placeholder for actual Gremlin query execution
    gremlin_query = f"g.V('{node_id}').outE('{relationship_type}').inV().valueMap()"
    if not relationship_type:
        gremlin_query = f"g.V('{node_id}').out().valueMap()"
    
    # Execute query against JanusGraph
    # results = await execute_gremlin(gremlin_query)
    # return results
    return {"node_id": node_id, "neighbors": []} # Mock response

The data flow progresses from Kafka ingestion, through Spark for transformation, to JanusGraph with ScyllaDB for persistent storage, and finally to the FastAPI service for client access. This architecture supports both analytical and transactional graph workloads at scale.

Ship Graph Data: modeling and ingestion pipelines

The core of the production graph system is its data model, which represents global maritime traffic. We model ships, ports, and voyages as nodes. Relationships define their interactions: DOCKED_AT, TRAVELED_ROUTE, CARRIES. Each node and relationship carries relevant properties for analysis and querying.

Ship nodes (:Ship) are identified by their IMO number, a unique identifier for vessels. Properties include imoId: String, name: String, type: String (e.g., “Cargo”, “Tanker”), and flag: String. Port nodes (:Port) use UN/LOCODE as their primary key, with properties like locode: String, name: String, country: String, and coordinates: Point. Voyages (:Voyage) link a ship, its origin, and destination, including startTime: DateTime and endTime: DateTime.

Relationships connect these entities. A ship DOCKED_AT a port has arrival: DateTime and departure: DateTime properties. A ship TRAVELED_ROUTE between two ports can include distanceKm: Float and durationHours: Float. This structure allows queries to trace ship movements, port calls, and cargo flows across the network.

Source data for this model originates from multiple streams. AIS (Automatic Identification System) transponder data provides real-time ship positions, which we aggregate to infer port calls and voyage segments. Static ship registries supply vessel details, while port authority databases offer port metadata. These disparate sources require standardization before ingestion.

The ingestion pipeline uses a two-phase approach: initial batch load for historical data, followed by continuous streaming updates. The batch process transforms CSV or JSON files from historical AIS archives and registry dumps into Cypher MERGE statements. This ensures idempotency; re-running the ingestion for existing entities updates properties without creating duplicates.

For example, ingesting a ship and its port call:

MERGE (s:Ship {imoId: $shipImoId})
ON CREATE SET s.name = $shipName, s.type = $shipType
ON MATCH SET s.name = $shipName, s.type = $shipType
MERGE (p:Port {locode: $portLocode})
ON CREATE SET p.name = $portName, p.country = $portCountry
ON MATCH SET p.name = $portName, p.country = $portCountry
MERGE (s)-[d:DOCKED_AT {arrival: $arrivalTime}]->(p)
ON CREATE SET d.departure = $departureTime
ON MATCH SET d.departure = $departureTime
RETURN s, p, d

Streaming updates process near real-time AIS messages via a Kafka topic. A Python service consumes these messages, performs lightweight transformation, and generates Cypher transactions. These transactions are batched for efficiency and sent to the graph database. This approach provides low-latency updates for current ship positions and inferred events. The system ensures data consistency by using transaction boundaries and retries for transient database connection issues.

The ingestion pipeline’s architecture prioritizes fault tolerance. Each processing stage (data extraction, transformation, loading) is decoupled, often running as separate microservices or Airflow tasks. Failed batches can be reprocessed from source, and individual streaming message failures are logged for manual review or automated dead-letter queue handling. This design maintains data integrity even under high load or intermittent source system outages.

Graph Application: core logic implementation

The application layer translates business requirements into graph queries and exposes these operations via API endpoints. This layer ensures data integrity and provides structured access to the underlying graph database, which was configured in the previous chapter.

Consider a system recommending products. A basic query might fetch a product by its ID. A more complex request, such as “recommend products liked by users that a specific user follows, excluding products the primary user already likes,” requires multi-hop traversals and property filtering within the graph.

The get_recommended_products function demonstrates this logic. It constructs a Cypher query to traverse FOLLOWS and LIKES relationships, then filters out already-liked products. The function uses a graph_client object, assumed to be initialized and capable of executing read operations against the graph database.

# app/services/recommendations.py
from typing import List, Dict

# Assume graph_client is an initialized connection to the graph database
# from a previous chapter (e.g., Neo4j driver instance)
from app.database import graph_client

def get_recommended_products(user_id: str) -> List[Dict]:
    """
    Retrieves product recommendations for a given user.
    Recommendations are products liked by users the target user follows,
    excluding products the target user has already liked.
    """
    query = """
    MATCH (u:User {id: $user_id})-[:FOLLOWS]->(f:User)-[:LIKES]->(p:Product)
    WHERE NOT (u)-[:LIKES]->(p)
    RETURN DISTINCT p.id AS product_id, p.name AS product_name
    LIMIT 10
    """
    params = {"user_id": user_id}
    
    # Execute the query and map results to a list of dictionaries
    result_records = graph_client.execute_read(query, params)
    return [{"id": r["product_id"], "name": r["product_name"]} for r in result_records]

This core logic is then exposed through an API endpoint. Using a web framework like Flask or FastAPI, a route is defined to accept requests for user recommendations. The endpoint calls the get_recommended_products function and formats its output as a JSON response.

# app/api/recommendations.py
from flask import Flask, jsonify
from app.services.recommendations import get_recommended_products

app = Flask(__name__) # Simplified for example, assume proper app setup

@app.route("/users/<string:user_id>/recommendations", methods=["GET"])
def user_recommendations_endpoint(user_id: str):
    """
    API endpoint to fetch product recommendations for a specific user.
    """
    if not user_id:
        return jsonify({"error": "User ID is required"}), 400
    
    try:
        recommendations = get_recommended_products(user_id)
        return jsonify(recommendations), 200
    except Exception as e:
        # Log the exception for debugging purposes
        print(f"Error fetching recommendations for user {user_id}: {e}")
        return jsonify({"error": "Failed to retrieve recommendations"}), 500

Direct graph queries offer real-time data but can incur higher latency for complex traversals. An alternative involves pre-computing certain recommendations offline and storing them in a cache or a simpler database, trading real-time accuracy for faster response times. The choice depends on the specific latency and data freshness requirements of the application.

Deploy & Operate: production graph systems

Deploying the graph system to a production environment begins with packaging each service into containers. Docker provides a consistent runtime across development and production, encapsulating the application and its dependencies. This ensures the environment where the code runs remains identical, eliminating “it works on my machine” issues.

Orchestration with Kubernetes manages these containerized services, providing automated deployment, scaling, and operational management. A Dockerfile for the graph API service defines its build process, specifying the base image, dependencies, and application entry point.

# Dockerfile for graph-api-service
FROM python:3.9-slim-buster
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Kubernetes manifests define the deployments, services, and ingress rules for the graph system. PersistentVolumeClaims secure the graph database’s data, ensuring data integrity and durability across pod restarts or failures. After defining these configurations, deploy them using kubectl.

kubectl apply -f graph-api-deployment.yaml
kubectl apply -f graph-db-statefulset.yaml
kubectl apply -f graph-api-service.yaml

Monitoring is key to understanding system health and performance. Prometheus collects time-series metrics from application endpoints, providing data on request latency, error rates, and resource consumption. Grafana visualizes these metrics through dashboards, offering a real-time view of the system’s operational state.

Centralized logging aggregates application and infrastructure logs, making it possible to diagnose issues quickly. Tools like kubectl logs offer immediate access to container output for a specific pod.

kubectl logs graph-api-service-7c8d9f-abc12 -n graph-system

Alerting configurations define thresholds for critical metrics, triggering notifications when anomalies occur. This proactive approach allows operators to address issues before they impact users. Examples include alerts for high API error rates, low database disk space, or increased query latency.

Operational readiness involves defining clear strategies for data backup and recovery. Implement automated snapshots of graph database volumes, stored in a separate, secure location. Regular testing of the recovery process confirms its effectiveness.

Scaling strategies address varying load conditions. Horizontal Pod Autoscalers (HPAs) automatically adjust the number of API service replicas based on CPU utilization or custom metrics. For the graph database, scaling might involve adding read replicas or vertically scaling resources.

Finally, establish an incident response plan. This includes defining on-call rotations, documenting common issue runbooks, and outlining communication protocols during outages. A well-defined plan reduces resolution times and minimizes service disruption.