Getting Started with Spark Streaming: Building Real-Time Data Pipelines

The Power of Real-Time Processing

In today’s fast-paced world, businesses need to act on data as it arrives. Real-time processing allows you to analyze and respond to events instantly, whether it’s detecting fraud, updating dashboards, or triggering alerts. Spark Streaming extends Spark’s batch processing model to handle continuous data, offering:

• Scalability: Processes streams across a cluster, handling high-throughput data.
• Fault Tolerance: Ensures data is processed reliably with checkpointing and recovery.
• Unified API: Combines streaming with batch and SQL operations for seamless pipelines.
• Integration: Connects to sources like Kafka, Flume, or TCP sockets.

Unlike traditional streaming systems that process events one-by-one, Spark Streaming uses a micro-batch approach, balancing latency and throughput. For a foundational look at Spark, see Spark how it works.

What is Spark Streaming?

Spark Streaming is a Spark module that processes continuous data streams by dividing them into small, fixed-size batches called micro-batches. Each micro-batch is treated as a Resilient Distributed Dataset (RDD) or DataFrame, allowing you to use Spark’s familiar APIs for transformations and actions. This approach leverages Spark’s distributed architecture, enabling fault tolerance, scalability, and integration with batch processing.

Key concepts include:

• DStream: A sequence of RDDs representing the stream, used in the legacy DStream API.
• Structured Streaming: A higher-level API (introduced in Spark 2.0) built on DataFrames, offering better optimization and ease of use.
• Micro-Batches: Data collected over a short interval (e.g., 1 second) processed as a batch.
• Sources: Input streams like Kafka, files, or sockets PySpark streaming input sources.
• Sinks: Output destinations like files, databases, or consoles PySpark streaming output sinks.

Spark Streaming supports two APIs:

1. DStream API: Legacy API, RDD-based, still used in some scenarios (PySpark legacy DStream API).
2. Structured Streaming: Modern API, DataFrame-based, recommended for new applications (PySpark structured streaming overview).

This guide focuses on Structured Streaming, as it’s the preferred approach, but we’ll touch on the DStream API for context.

How Spark Streaming Works

Spark Streaming processes data in a continuous loop:

1. Data Ingestion: Reads data from a source (e.g., Kafka topic) in real-time.
2. Micro-Batch Creation: Groups incoming data into micro-batches based on a time interval (e.g., every 2 seconds).
3. Processing: Applies transformations (e.g., filter, groupBy) and actions (e.g., write to sink) to each micro-batch.
4. Output: Writes results to a sink, such as a file, database, or console.
5. Checkpointing: Saves state for fault tolerance, ensuring no data loss on failure (PySpark streaming checkpointing).

Structured Streaming builds on Spark SQL’s Catalyst Optimizer and Tungsten engine, enabling optimizations like predicate pushdown and whole-stage code generation (Spark Catalyst Optimizer).

Setting Up Spark Streaming

Before building a streaming application, ensure your environment is ready. We’ll use Spark with a Kafka source, a common choice for streaming data.

Prerequisites

1. Spark Installation:
   
   • Install Spark 3.5.x or later PySpark installation.
   • Verify with:

   • spark-shell

1. Kafka Setup:
   
   • Install Apache Kafka (e.g., version 3.7.x) from kafka.apache.org.
   • Start ZooKeeper and Kafka server:

   • bin/zookeeper-server-start.sh config/zookeeper.properties
          bin/kafka-server-start.sh config/server.properties

   • Create a topic:

   • bin/kafka-topics.sh --create --topic sales --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

1. Dependencies:
   
   • Include the Kafka connector for Spark:
   • Maven (Scala): org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0.
   • PySpark: Ensure the JAR is available (spark.jars.packages).
   • Example for PySpark:

   • spark = SparkSession.builder \
              .appName("StreamingSetup") \
              .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0") \
              .getOrCreate()

1. Data Source:
   
   • Simulate streaming data by sending messages to the Kafka topic:

   • bin/kafka-console-producer.sh --topic sales --bootstrap-server localhost:9092

Input sample JSON:

{"order_id": 1, "amount": 100, "region": "North"}
     {"order_id": 2, "amount": 200, "region": "South"}

Building Your First Structured Streaming Application

Let’s create a simple streaming application that reads sales data from Kafka, processes it, and writes results to the console. We’ll cover both PySpark and Scala examples, focusing on Structured Streaming.

Structured Streaming with PySpark

This example reads JSON messages from a Kafka topic, filters high-value orders, and aggregates totals by region.

Code

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as sum_
from pyspark.sql.types import StructType, StructField, StringType, IntegerType

# Initialize Spark session
spark = SparkSession.builder \
    .appName("StructuredStreamingKafka") \
    .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0") \
    .config("spark.sql.streaming.checkpointLocation", "/tmp/checkpoint") \
    .getOrCreate()

# Define schema for JSON data
schema = StructType([
    StructField("order_id", IntegerType()),
    StructField("amount", IntegerType()),
    StructField("region", StringType())
])

# Read from Kafka
df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "sales") \
    .load()

# Parse JSON from Kafka value
parsed_df = df.selectExpr("CAST(value AS STRING)") \
    .select(from_json(col("value"), schema).alias("data")) \
    .select("data.*")

# Process stream: filter and aggregate
filtered_df = parsed_df.filter(col("amount") > 150)
aggregated_df = filtered_df.groupBy("region").agg(sum_("amount").alias("total_amount"))

# Write to console
query = aggregated_df.writeStream \
    .outputMode("complete") \
    .format("console") \
    .option("truncate", "false") \
    .start()

# Wait for termination
query.awaitTermination()

Parameters Explained

1. Spark Session:
   
   • appName: Names the application for UI tracking.
   • spark.jars.packages: Includes the Kafka connector.
   • spark.sql.streaming.checkpointLocation: Sets a directory for checkpointing to ensure fault tolerance.

1. Read Stream:
   
   • format("kafka"): Specifies Kafka as the source.
   • kafka.bootstrap.servers: Kafka server address (e.g., localhost:9092).
   • subscribe: Kafka topic to read (sales).
   • load(): Creates a streaming DataFrame with Kafka metadata (key, value, topic, etc.).

1. Parsing:
   
   • CAST(value AS STRING): Converts Kafka’s binary value to a string.
   • from_json: Parses JSON using the defined schema.
   • select("data.*"): Flattens the struct into columns (order_id, amount, region).

1. Processing:
   
   • filter(col("amount") > 150): Keeps orders with amount > 150.
   • groupBy("region").agg(sum_("amount")): Aggregates total amount per region.

1. Write Stream:
   
   • outputMode("complete"): Outputs the full result table after each micro-batch (suitable for aggregations).
   • format("console"): Writes to the console for debugging.
   • option("truncate", "false"): Prevents truncating long output.
   • start(): Begins streaming.
   • awaitTermination(): Keeps the job running until stopped.

For streaming concepts, see PySpark structured streaming overview.

Structured Streaming with Scala

The same pipeline in Scala:

Code

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, from_json, sum}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

object StructuredStreamingKafka {
  def main(args: Array[String]): Unit = {
    // Initialize Spark session
    val spark = SparkSession.builder()
      .appName("StructuredStreamingKafka")
      .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0")
      .config("spark.sql.streaming.checkpointLocation", "/tmp/checkpoint")
      .getOrCreate()

    // Define schema
    val schema = StructType(Seq(
      StructField("order_id", IntegerType),
      StructField("amount", IntegerType),
      StructField("region", StringType)
    ))

    // Read from Kafka
    val df = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("subscribe", "sales")
      .load()

    // Parse JSON
    val parsedDf = df.selectExpr("CAST(value AS STRING)")
      .select(from_json(col("value"), schema).alias("data"))
      .select("data.*")

    // Process stream
    val filteredDf = parsedDf.filter(col("amount") > 150)
    val aggregatedDf = filteredDf.groupBy("region")
      .agg(sum("amount").alias("total_amount"))

    // Write to console
    val query = aggregatedDf.writeStream
      .outputMode("complete")
      .format("console")
      .option("truncate", "false")
      .start()

    // Wait for termination
    query.awaitTermination()
  }
}

Running the Scala Application

1. Package with SBT or Maven, including the Kafka dependency.
2. Submit the job:

spark-submit --class StructuredStreamingKafka \
       --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0 \
       target/your-app.jar

The Scala version mirrors the PySpark logic, parsing JSON, filtering, aggregating, and outputting to the console.

Output

For input like:

{"order_id": 1, "amount": 100, "region": "North"}
{"order_id": 2, "amount": 200, "region": "South"}
{"order_id": 3, "amount": 300, "region": "North"}

The console shows (after a few micro-batches):

-------------------------------------------
Batch: 1
-------------------------------------------
+------+-------------+
|region|total_amount|
+------+-------------+
|South |200          |
|North |300          |
+------+-------------+

Only orders with amount > 150 appear, aggregated by region.

Alternative Approach: DStream API

The legacy DStream API processes streams as RDDs, offering lower-level control but less optimization than Structured Streaming. It’s still used in some legacy applications.

Example in PySpark

Reading text from a socket:

from pyspark.sql import SparkSession
from pyspark.streaming import StreamingContext

spark = SparkSession.builder.appName("DStreamSocket").getOrCreate()
sc = spark.sparkContext
ssc = StreamingContext(sc, 1) # 1-second batches

# Read from socket
lines = ssc.socketTextStream("localhost", 9999)

# Process stream
words = lines.flatMap(lambda line: line.split())
word_counts = words.map(lambda word: (word, 1)).reduceByKey(lambda a, b: a + b)

# Output to console
word_counts.pprint()

# Start streaming
ssc.start()
ssc.awaitTermination()

Setup

1. Start a netcat server:

nc -lk 9999

1. Input text (e.g., “hello world”).
2. Output shows word counts per batch.

Parameters

• StreamingContext(sc, 1): Creates a context with 1-second batches.
• socketTextStream(host, port): Reads text from a TCP socket.
• flatMap, map, reduceByKey: Transform the DStream.
• pprint(): Prints results to the console.

For DStream details, see PySpark legacy DStream API.

Why Choose Structured Streaming?

• Optimization: Leverages Catalyst and Tungsten Spark Tungsten optimization.
• Ease of Use: DataFrame API is more intuitive than RDDs.
• Features: Supports watermarking, triggers, and stateful operations PySpark streaming watermarking.

Use DStream only for legacy compatibility or specific low-level needs.

Configuring Spark Streaming

Fine-tune your streaming job with these settings:

1. Batch Interval:
   
   • Controls micro-batch size.
   • Structured Streaming: Set via trigger:

   • query = aggregated_df.writeStream \
              .trigger(processingTime="2 seconds") \
              .outputMode("complete") \
              .format("console") \
              .start()

   • For triggers, see PySpark streaming triggers.

1. Checkpointing:
   
   • Enables fault tolerance by saving state:

   • .option("checkpointLocation", "/tmp/checkpoint")

1. Kafka Options:
   
   • startingOffsets: Where to start reading (earliest, latest).

   • .option("startingOffsets", "earliest")

   • maxOffsetsPerTrigger: Limits messages per batch.

   • .option("maxOffsetsPerTrigger", "1000")

1. Output Modes:
   
   • complete: Full result table (used above).
   • append: New rows only.
   • update: Updated rows only.
   • Example:

   • .outputMode("append")

For output modes, see PySpark structured streaming overview.

Debugging and Monitoring

Streaming jobs require continuous monitoring:

• Spark UI: Check the Streaming tab for batch progress, latency, and throughput.
• Logs: Enable event logging:

• spark.conf.set("spark.eventLog.enabled", "true")
    spark.conf.set("spark.eventLog.dir", "hdfs://namenode:8021/spark-logs")

• Execution Plans: Use explain() to verify optimizations.
• Metrics: Monitor input rate vs. processing rate to avoid backlog.

For debugging, see Spark how to debug Spark applications.

Best Practices

Build robust streaming applications with these tips:

• Use Structured Streaming: Prefer it over DStream for optimization and simplicity.
• Set Checkpointing: Ensure fault tolerance with a reliable checkpoint directory.
• Tune Batch Size: Balance latency and throughput with appropriate triggers.
• Monitor Performance: Watch the Spark UI for backlog or latency issues.
• Handle Skew: Use salting for uneven data PySpark handling skewed data.
• Validate Data: Check schemas and sample inputs PySpark printSchema.

Common Pitfalls

Avoid these mistakes:

• No Checkpointing: Risks data loss. Solution: Always set checkpointLocation.
• Large Batches: Causes latency. Solution: Adjust trigger interval.
• Unoptimized Queries: Slows processing. Solution: Use filters and pruning PySpark filter.
• Ignoring Backlog: Overwhelms the job. Solution: Monitor rates and scale resources.

Practical Example: Enhanced Streaming Pipeline

Let’s extend the pipeline to write to a file sink and handle late data with watermarking:

spark = SparkSession.builder \
    .appName("EnhancedStreaming") \
    .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0") \
    .config("spark.sql.streaming.checkpointLocation", "/tmp/checkpoint") \
    .getOrCreate()

schema = StructType([
    StructField("order_id", IntegerType()),
    StructField("amount", IntegerType()),
    StructField("region", StringType()),
    StructField("timestamp", StringType())
])

df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "sales") \
    .option("startingOffsets", "earliest") \
    .load()

parsed_df = df.selectExpr("CAST(value AS STRING)") \
    .select(from_json(col("value"), schema).alias("data")) \
    .select("data.*")

# Add watermark for late data
watermarked_df = parsed_df.withWatermark("timestamp", "10 minutes")
filtered_df = watermarked_df.filter(col("amount") > 150)
aggregated_df = filtered_df.groupBy("region").agg(sum_("amount").alias("total_amount"))

# Write to Parquet
query = aggregated_df.writeStream \
    .outputMode("append") \
    .format("parquet") \
    .option("path", "s3://bucket/output") \
    .partitionBy("region") \
    .trigger(processingTime="5 seconds") \
    .start()

query.awaitTermination()

This pipeline:

• Adds a timestamp field and watermarking to handle late data PySpark streaming watermarking.
• Writes to partitioned Parquet files for efficient storage PySpark write Parquet.
• Uses a 5-second trigger for lower latency.

Next Steps

Continue exploring Spark Streaming with:

• Stateful streaming Spark stateful vs stateless streaming.
• Kafka integration Spark how to build streaming app using Kafka.
• Performance optimization Spark how to optimize jobs for max performance.

Try the Databricks Community Edition for hands-on practice.

By mastering Spark Streaming, you’ll build real-time pipelines that scale seamlessly, delivering insights as data flows in.