Multi-Model Data Federation: Ingesting PostgreSQL Tables and Delta Lake Batch Files into Neo4j via PySpark

Resolving JVM URI Hostname Incompatibilities, Managing Distributed MERGE Lock Collisions, and Evaluating Single-Node Graph Storage Trade-offs in a Containerized Sandbox

Introduction

In Phase 3 of this series, we engineered a multi-tiered real-time Delta Lakehouse pipeline to offload high-velocity clickstream telemetry payloads from our transactional database core. This operational boundary protected our relational PostgreSQL buffer cache from disk I/O degradation. However, executing relational joins or analytical scans over complex network paths generates high infrastructure costs. Tracing recursive metadata pathways or entity overlaps across deep SQL tables or flat Parquet structures forces continuous self-joins. At scale, this access pattern compromises database throughput.

To handle interconnected entities efficiently, we introduced Neo4j 5 to our isolated staging topology subcontext-d-graph (the GitHub repository subfolder for this phase). The primary objective of this phase is not predictive modeling or machine learning; rather, it is a pure Data Engineering Proof of Concept (PoC) designed to validate multi-model data pipeline federation. This artifact documents the physical challenges of loading data concurrently from PostgreSQL and Delta Lake using PySpark, mitigating distributed transactional deadlocks, and examining real-world production alternatives for connected network storage layers.

Chapter 1: System Blueprint and Schema Design

The graph tier functions as a separate deployment unit inside our container workspace, connecting to the analytical network via a dedicated external link. The platform layers are organized directly across three storage planes:

<pre><code>
┌────────────────────────────────────────┐
│  OPERATIONAL TIER (PostgreSQL 16)     │
│  - User Metadata Tables via JDBC      │
└──────────────────┬─────────────────────┘
                   │
                   ▼
┌────────────────────────────────────────┐    (Bolt Protocol)    ┌────────────────────────────────────────┐
│  DISTRIBUTED COMPUTE (PySpark v3.5)    │ ───────────────────▶  │  GRAPH Persistence PLANE (Neo4j 5)     │
│  - Joint Ingestion & Data Federation   │                       │ - Topological Edge Intersections        │
└──────────────────▲─────────────────────┘                       └────────────────────────────────────────┘
                   │
                   ▼
┌────────────────────────────────────────┐
│  ANALYTICAL STORAGE (MinIO S3 Core)    │
│  - Refined Delta Lake Silver Files     │
└────────────────────────────────────────┘
</code></pre>

📐 The Target Schema Model

To avoid computation overheads during execution, our target schema extracts specific string markers into distinct graph vertices:

Nodes (Labels)

• (:User) — Represents user accounts pulled from the operational PostgreSQL table. Properties: user_id (INT), first_name (STR), email (STR).
• (:IPAddress) — Represents unique routing strings scanned from the Delta Lake Silver directory. Properties: ip_string (STR).

Relationships (Edges)

• (:User)-[:CLICKED_FROM {device_type, ingested_at}]->(:IPAddress): Connects individual users to network locations.

ℹ️ DESIGN CAVEAT: 
Relying entirely on simple graph links to identify fraud rings creates immediate false-positive anomalies. 
Legitimate users inside the same household (e.g., Ali and Ayşe), corporate office, or public network space regularly share a single public IP node through network address translation (NAT). 
For realistic profiling, data pipelines must ingest operational metadata—such as hardware variables and short time windows—to avoid misidentifying clean household structures as malicious activities.

Chapter 2: Overcoming JVM Environment Failures and Distributed Write Contentions

Orchestrating raw data across JDBC, S3A, and Bolt interfaces introduces runtime environment errors that require specific database adjustments.

Bypassing Host Environment Dependencies via Containerized Kernels

Our initial implementation plan attempted to execute the synchronization script natively via the host shell. This failed immediately because the local host environment lacked a native Java JDK installation, throwing a fatal [JAVA_GATEWAY_EXITED] runtime error. PySpark requires a valid active Java gateway process to bind internal Python data tasks to JVM classes. Rather than installing local JDK instances and polluting our host system environment variables, we bypassed this issue by migrating our entire processing logic into a containerized Jupyter notebook utilizing an embedded PySpark Kernel (v3.5.0). This keeps our host OS clean.

Handling JVM URI Hostname Restraints and Underscore Discrepancies

When executing our ingestion engine using early standalone spark-submit runs, the underlying Java driver rejected our network paths, throwing an address configuration exception:

java.lang.IllegalArgumentException: Illegal character in scheme name at index 2: lh_neo4j:7687

This crash stems from RFC network naming regulations enforced by the native java.net.URI library within the Neo4j connector. Standard URI schemas prohibit the use of underscores (_) inside hostnames. Since our default Phase 3 container name contains underscores, the Java parser aborted the connection. We resolved this portability issue by defining a clean network alias (neo4jhost) inside the Docker networks layer, allowing our Spark tasks to write safely over a standardized bolt://neo4jhost:7687 path across any deployment environment.

ℹ️ S3A CONNECTOR DISCREPANCY NOTE: 
White java.net.URI strictly rejects underscores within the Neo4j Bolt initialization step, 
the endpoint variable targeting http://lh_minio:9000 does not fail. 
This occurs because the Apache Hadoop S3A filesystem provider delegates its lower-level HTTP parsing loops to independent client libraries (AWS Java SDK Bundle) which employ alternative string normalization mechanisms that bypass standard strict RFC hostname validations.

Deconstructing Deadlocks: Idempotency, MERGE Semantics, and Batching

During bulk data runs, our PySpark engine allocated write partitions across all available local CPU execution threads simultaneously. This parallel pressure caused severe transactional aborts inside the target graph database engine:

org.neo4j.driver.exceptions.TransientException: ForsetiClient[transactionId=156, clientId=9] 
can't acquire ExclusiveLock on NODE_RELATIONSHIP_GROUP_DELETE(16607) because holders of that lock 
are waiting for ForsetiClient[transactionId=97, clientId=4].

To understand this deadlock, we must look at the native Neo4j write patterns, internal batch ingestion, and lock hierarchy metrics:

• MERGE vs CREATE and Idempotency: The Neo4j Spark connector uses the Cypher MERGE statement instead of CREATE to enforce pipeline idempotency. CREATE blindly appends rows, causing massive duplicate node bloat if a pipeline runs twice. MERGE acts as an “upsert”, checking for node existence before writing.
• The Batch Ingestion Lock Collision: Under the hood, the Spark connector processes data frames in sequential chunks with a default “batch.(size: 1000)”. *When executing multiple parallel tasks (local[]), different threads try to MERGE different batches simultaneously.
• The Deadlock Chain: When thread A checks for an IP node inside batch 1, it acquires an Exclusive Lock on that specific Node structure. If thread B simultaneously tries to link a different user to that exact same IP node inside batch 2, it is forced to wait. If these mutual dependencies lock up across different threads, the internal ForsetiClient locks up and aborts the task to protect storage integrity.

To bypass this deadlock in our local prototype sandbox, we applied a harsh throughput bottleneck by adding .coalesce(1) right before our dataframe write blocks. This serialization forces Spark to drop its parallel executors and channel data through a single write tunnel, removing write collisions completely.

⚠️ PRODUCTION SCALING ALTERNATIVES: 
While using .coalesce(1) resolves lock contentions inside a local sandbox,
it kills write throughput and breaks distributed scale-out logic in production clusters. 
In a real-world infrastructure, you must maintain active parallel execution paths. 
Production deadlocks are mitigated by configuring unique schema constraints 
inside the graph catalog, deploying an application-level retry strategy with exponential backoff,
or partitioning data frames by an explicit hash key to ensure overlapping records are processed sequentially.

Chapter 3: Implementation — The Data Federation Pipeline

The complete multi-model pipeline logic is deployed within our containerized workspace to load relational tables and object storage directories concurrently.

🔒 SECURITY ENFORCEMENT NOTICE: 
The plain-text credential values below (de_password123, graph_password123) are configured strictly for local validation purposes inside an isolated sandbox. 
Storing hardcoded secrets in code repositories represents a major security vulnerability. 
Production deployments must manage credential pools dynamically via secure external infrastructure layers such as HashiCorp Vault.

The pipeline logic contains explicit try/except safety fences around the initial dataset ingestion blocks:

import os
import sys
from pyspark.sql import SparkSession
from pyspark.sql.functions import col

# =========================================================================
# INFRASTRUCTURE ROUTING CONFIGURATIONS
# =========================================================================
PG_URL = "jdbc:postgresql://host.docker.internal:5433/ecommerce_db"
PG_PROPERTIES = {
    "user": "data_engineer",
    "password": "de_password123",
    "driver": "org.postgresql.Driver"
}

# Portable endpoint using our configured network alias
NEO4J_URI = "bolt://neo4jhost:7687"
NEO4J_USER = "neo4j"
NEO4J_PASSWORD = "graph_password123"

# =========================================================================
# INITIALIZE SPARK SESSION WITH DELTA AND NEO4J DRIVERS
# =========================================================================
SUBMIT_PACKAGES = (
    "io.delta:delta-spark_2.12:3.0.0,"
    "org.apache.hadoop:hadoop-aws:3.3.4,"
    "org.postgresql:postgresql:42.6.0,"
    "org.neo4j:neo4j-connector-apache-spark_2.12:5.3.1_for_spark_3"
)

os.environ["PYSPARK_SUBMIT_ARGS"] = f"--packages {SUBMIT_PACKAGES} pyspark-shell"
os.environ["AWS_ACCESS_KEY_ID"] = "admin"
os.environ["AWS_SECRET_ACCESS_KEY"] = "minio_password123"

try:
    spark = SparkSession.builder \
        .appName("Ecommerce_Graph_Topology_Sync_Engine") \
        .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
        .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
        .config("spark.hadoop.fs.s3a.endpoint", "http://lh_minio:9000") \
        .config("spark.hadoop.fs.s3a.path.style.access", "true") \
        .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem") \
        .config("spark.hadoop.fs.s3a.connection.ssl.enabled", "false") \
        .getOrCreate()

    print("🏁 Containerized PySpark Session initialized.")
except Exception as e:
    print(f"❌ Failed to initialize distributed Spark environment: {str(e)}")
    sys.exit(1)

# =========================================================================
# PIPELINE EXECUTION PASSTHROUGH
# =========================================================================
try:
    print("\n📥 Staging users from PostgreSQL via JDBC...")
    pg_users_df = spark.read.jdbc(url=PG_URL, table="users", properties=PG_PROPERTIES)

    print("📥 Staging access log records from Delta Lake Silver tier...")
    delta_silver_df = spark.read.format("delta").load("file:///home/jovyan/work/notebooks/data/silver/clickstream")
except Exception as e:
    print(f"❌ Pipeline Stage 2 Extraction aborted due to connection error: {str(e)}")
    sys.exit(1)

print("\n🚀 Commencing graph Ingestion passes...")

# Pass A: Write (:User) Nodes via single-thread pipeline serialization
pg_users_df.coalesce(1).write \
    .format("org.neo4j.spark.DataSource") \
    .option("url", NEO4J_URI) \
    .option("authentication.type", "basic") \
    .option("authentication.basic.username", NEO4J_USER) \
    .option("authentication.basic.password", NEO4J_PASSWORD) \
    .option("labels", "User") \
    .option("node.properties", "user_id,first_name,email") \
    .option("node.keys", "user_id") \
    .mode("Append") \
    .save()

print("✅ (:User) Nodes written successfully.")

# Pass B: Extract and write unique (:IPAddress) Nodes
unique_ips_df = delta_silver_df.select("ip_address").distinct().withColumnRenamed("ip_address", "ip_string")

unique_ips_df.coalesce(1).write \
    .format("org.neo4j.spark.DataSource") \
    .option("url", NEO4J_URI) \
    .option("authentication.type", "basic") \
    .option("authentication.basic.username", NEO4J_USER) \
    .option("authentication.basic.password", NEO4J_PASSWORD) \
    .option("labels", "IPAddress") \
    .option("node.properties", "ip_string") \
    .option("node.keys", "ip_string") \
    .mode("Append") \
    .save()

print("✅ (:IPAddress) Nodes written successfully.")

# Pass C: Inject structural edges via native Cypher merge options
delta_silver_df.coalesce(1).write \
    .format("org.neo4j.spark.DataSource") \
    .option("url", NEO4J_URI) \
    .option("authentication.type", "basic") \
    .option("authentication.basic.username", NEO4J_USER) \
    .option("authentication.basic.password", NEO4J_PASSWORD) \
    .option("relationship", "CLICKED_FROM") \
    .option("relationship.save.strategy", "keys") \
    .option("source.labels", "User") \
    .option("source.keys", "user_id") \
    .option("source.properties", "user_id") \
    .option("target.labels", "IPAddress") \
    .option("target.keys", "ip_address:ip_string") \
    .option("target.properties", "ip_address") \
    .option("relationship.properties", "device_type,ingested_at") \
    .mode("Append") \
    .save()

print("✅ [:CLICKED_FROM] edges successfully locked to storage blocks.")

Chapter 4: Data Evaluation and Architectural Trade-offs

To check the state of our graph storage, we execute queries directly inside the native Neo4j browser console.

Multi-Entity Network Intersections

To find distinct user accounts sharing a single network access location, we execute a structural Cypher Query Language match statement:

MATCH (u1:User)-[r1:CLICKED_FROM]->(ip:IPAddress)<-[r2:CLICKED_FROM]-(u2:User)
WHERE u1.user_id < u2.user_id
RETURN u1, r1, ip, r2, u2
LIMIT 25;

This returns a clear node connection view inside the graphical console engine:

[ User: Wendy ] ───[:CLICKED_FROM {device: tablet}]───┐
                                                       ▼
[ User: Brian ] ───[:CLICKED_FROM {device: tablet}]───➔ ( IPAddress: 192.168.1.153 )
                                                       ▲
[ User: Emily ] ───[:CLICKED_FROM {device: tablet}]───┘

The Data Generation Uniformity Paradox and Cardinality Explosion Risks

Analyzing our benchmark metrics exposes an important data engineering reality:

• Every single connection weight logs a flat frequency score of exactly.

(User: Payne)  ───[Connection_Weight: 1]───➔ (IPAddress: 192.168.1.153) ➔ (User: Bush)
(User: Payne)  ───[Connection_Weight: 1]───➔ (IPAddress: 192.168.1.153) ➔ (User: Cooper)

This uniform distribution is a side effect of synthetic test generation. In actual software systems, high-risk botnets hit specific network paths repeatedly, creating heavy weighted clusters. However, our standard Python Faker pipeline spreads log streams across a completely flat, homogeneous randomness curve. This data generation noise removes natural density from our edges.

⚠️ CARDINALITY EXPLOSION RISK (SUPER_NODES): 
While our synthetic pipeline maintains flat connection weights, 
scaling real-world telemetry workloads introduces severe Cardinality Explosion risks. 
If millions of unique user nodes connect to a single public IP address vertex (such as a large cellular tower proxy or corporate data center NAT), that vertex transforms into a "Supernode". 
In Neo4j's native pointer-based storage layer, traversing a Supernode forces long label scans across giant relationship groups, severely dragging down transactional query performance. 
Production graphs resolve this by enforcing strict degree limits, indexing specific property keys, or partitioning dense labels into separate structural sub-types.

To filter out this background variance and locate specific footprints inside our local data distribution, we deploy a Virtual Relationship Query via APOC, matching hardware profiles alongside network nodes:

MATCH (u1:User)-[r1:CLICKED_FROM]->(ip:IPAddress)<-[r2:CLICKED_FROM]-(u2:User)
WHERE u1.user_id < u2.user_id 
  AND r1.device_type = r2.device_type  // Composite Filter: Extracting identical hardware footprints
WITH u1, u2, ip, r1.device_type AS shared_device
RETURN u1, u2, apoc.create.vRelationship(u1, 'Niche_Collusion', {Shared_IP: ip.ip_string, Hardware: shared_device}, u2) AS rel
LIMIT 15;

This narrows our browser view down to isolated account structures using matching hardware types over shared network infrastructure simultaneously.

Single-Node Graph Database vs. Local In-Memory Batch Alternatives

The performance of a graph query depends heavily on indices, data cardinality, and hardware sizing. Within our single-node local development sandbox, engineers face a clear trade-off between native operational storage and in-memory execution engines:

Native Graph Tier (Neo4j Community Sandbox):

• Operational Strength: Offers high transactional index speeds, persistent disk-level node pointers, and instant visualization of relationship networks via native Cypher execution loops.
• Local Boundary: Operational threads are entirely constrained by single-instance host CPU and memory limits. Scaling out requires structural sharding layers.

In-Memory Batch Tier (PySpark GraphX / Local Frames):

• Operational Strength: Bypasses data movement costs entirely. Executes heavy graph algorithms (like PageRank or Connected Components) directly inside local RAM executors using existing distributed dataframes.
• Local Boundary: Completely unsuited for ad-hoc transactional lookups or sub-second UI queries. Acts purely as a heavy batch transformation and compute utility.

Key Takeaways

• Federated Pipelines: Combining relational database states and lakehouse files inside a distributed compute layer provides clean ingestion tracking without polluting the source OLTP engine buffers.
• Concurrency Optimization: Distributed write pipelines cause lock contentions on overlapping records. Sandbox scenarios can use task serialization, but production engines require unique indices combined with application retry layers.
• RFC Portability Standards: Native Java runtime components reject underscores inside hostname configurations. Deploying explicit network aliases inside container networks ensures cross-environment compatibility.
• Synthetic Data Limitations: Mock data sets built on uniform randomness flatten behavioral connection metrics. Engineers must apply composite filters to extract structural clarity from synthetic noise.

Conclusion

By decoupling high-velocity network markers from our operational core and loading them into an independent graph storage layer, we completed our hybrid architecture. This PoC confirms that modern database engineering is a multi-model equation where relational safety, lakehouse analytical performance, and graph relationship topologies must be engineered to run in sync.

The project-specific README prepared for this article can be accessed here.

The full project repository is available on GitHub here.

[embed]GitHub - EnigmaDevelop/ecommerce-database-engineering Contribute to EnigmaDevelop/ecommerce-database-engineering development by creating an account on GitHub.github.com