🚀 Admissions Open for Winter Training Programs 2026
Datadrix Company
Data EngineeringFull Stack Pipeline14 min read

Build a Real-Time Data Pipeline Using Kafka, PySpark and PostgreSQL

End-to-end tutorial assembling a streaming data pipeline: streaming financial transactions through Apache Kafka, processing with PySpark Structured Streaming, and sinking enriched records into PostgreSQL.

SR
Siddharth RoyVerified Mentor
Principal Distributed Systems Engineer @ Datadrix • Published Aug 02, 2026
#Real Time Pipeline#Kafka to PySpark#PySpark Streaming#PostgreSQL Pipeline#Data Engineering Capstone
Datadrix Technical SpecInteractive Architecture
Domain: Data EngineeringVerified by Datadrix Technical Council
Executive Overview

For technical interviews at top engineering firms, demonstrating an end-to-end distributed pipeline connecting an ingestion broker, a stream compute engine, and an operational database is the ultimate proof of competency. In this complete project tutorial, we connect Apache Kafka, PySpark Structured Streaming, and a PostgreSQL database to ingest, clean, detect fraud, and persist live transactions.

11. Architecture & Pipeline Data Flow

Step 1 (Ingestion): Financial microservice generates simulated payment events published to a Kafka topic named "financial_events".

Step 2 (Stream Compute): PySpark Structured Streaming consumes the stream, unpacks the binary JSON, applies schema validation, and flags suspicious transactions.

Step 3 (Sink & Persistence): Enriched records are continuously inserted into a PostgreSQL database using foreachBatch, ready for live BI dashboards.

22. Complete PySpark Streaming Application

Here is the complete Python streaming application bridging Kafka and PostgreSQL:

Production Code Snippet
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, when, current_timestamp
from pyspark.sql.types import StructType, StructField, StringType, DoubleType

# Initialize Spark with Kafka connector
spark = SparkSession.builder \
    .appName("KafkaPySparkPostgresPipeline") \
    .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0,org.postgresql:postgresql:42.6.0") \
    .getOrCreate()

# Define transaction schema
schema = StructType([
    StructField("txn_id", StringType(), False),
    StructField("account_id", StringType(), False),
    StructField("amount", DoubleType(), False),
    StructField("country", StringType(), True)
])

# 1. Read Kafka Stream
df_kafka = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "financial_events") \
    .load()

# 2. Parse & Apply Fraud Logic
processed_df = df_kafka \
    .select(from_json(col("value").cast("string"), schema).alias("data")) \
    .select("data.*") \
    .withColumn("is_suspicious", when(col("amount") > 50000, True).otherwise(False)) \
    .withColumn("processed_at", current_timestamp())

# 3. Write Micro-Batches to PostgreSQL
def write_to_postgres(batch_df, batch_id):
    batch_df.write \
        .format("jdbc") \
        .option("url", "jdbc:postgresql://localhost:5432/finance_db") \
        .option("dbtable", "live_transactions") \
        .option("user", "postgres_admin") \
        .option("password", "strong_password") \
        .option("driver", "org.postgresql.Driver") \
        .mode("append") \
        .save()
    print(f"Batch {batch_id} committed to PostgreSQL successfully.")

query = processed_df.writeStream \
    .foreachBatch(write_to_postgres) \
    .option("checkpointLocation", "/tmp/spark_checkpoints/postgres_sink") \
    .start()

query.awaitTermination()

33. Production Considerations: Checkpointing & Resilience

Checkpointing: PySpark records Kafka consumer offsets and batch progress to disk. If the Spark worker restarts, it resumes exactly where it left off without duplicating records.

Connection Pooling: In high-scale deployments, opening raw JDBC connections per batch can exhaust database connection pools. Using connection pooling (HikariCP) or pgBouncer ensures database stability.

Key Architectural Takeaways:
  • Use `foreachBatch` to seamlessly write PySpark streaming data into relational stores like PostgreSQL or MySQL.
  • Never run streaming jobs without a persistent checkpoint directory.

Mentor Conclusion & Industry Next Steps

Building this end-to-end streaming project elevates your resume above 95% of applicants. Datadrix’s labs in Delhi guide you through building and containerizing this exact architecture with Docker Compose.

Hands-On Technical Mentorship

Want to Master These Skills in Production?

Join Datadrix’s live cohort with 1:1 mentor code reviews, capstone projects, and 400+ hiring partner referrals.

Book Free Demo Class
Continue Reading

Related Guides & Insights

View All Articles