Cassandra with Spark

Apache Spark is a distributed processing engine for large-scale data analytics. Cassandra stores operational data at massive scale. The DataStax Spark Cassandra Connector bridges these two systems, letting Spark read Cassandra tables as DataFrames and write Spark results back to Cassandra — enabling batch analytics, machine learning, and ETL on live operational data without any data movement to a separate warehouse.

Why Use Spark with Cassandra

Cassandra Strength               Spark Strength
──────────────────────────────────────────────────────────────
Operational reads/writes         Analytical queries (aggregations,
                                 joins, ML)
Key-based lookups (fast)         Full-table scans (distributed)
Always-on, low latency           Batch and streaming processing
Limited aggregation support      SQL, DataFrames, MLlib

Adding the Connector

build.sbt (Scala/Spark)

libraryDependencies += "com.datastax.spark" %% "spark-cassandra-connector"
  % "3.5.0"

Maven (Java)

<dependency>
  <groupId>com.datastax.spark</groupId>
  <artifactId>spark-cassandra-connector_2.12</artifactId>
  <version>3.5.0</version>
</dependency>

PySpark (Python)

# Submit Spark job with the connector JAR:
spark-submit \
  --packages com.datastax.spark:spark-cassandra-connector_2.12:3.5.0 \
  my_spark_job.py

Creating a SparkSession Connected to Cassandra

from pyspark.sql import SparkSession

spark = SparkSession.builder \
  .appName("CassandraAnalytics") \
  .config("spark.cassandra.connection.host", "10.0.0.1,10.0.0.2") \
  .config("spark.cassandra.connection.port", "9042") \
  .config("spark.cassandra.auth.username", "alice") \
  .config("spark.cassandra.auth.password", "Al1ce$ecure!") \
  .config("spark.sql.extensions",
          "com.datastax.spark.connector.CassandraSparkExtensions") \
  .getOrCreate()

Reading a Cassandra Table into a DataFrame

# Read entire table as a Spark DataFrame:
orders_df = spark.read \
  .format("org.apache.spark.sql.cassandra") \
  .options(table="orders_by_customer", keyspace="ecommerce") \
  .load()

orders_df.printSchema()
orders_df.show(10)

# Output:
# +--------------------+--------------------+-------+------+
# |         customer_id|            order_id| status| total|
# +--------------------+--------------------+-------+------+
# |3a2f1b00-4e7d-11ee  |a1b2c3d4-5e6f-7890  |pending| 99.99|
# ...

Spark SQL on Cassandra Tables

# Register as a temporary SQL view:
orders_df.createOrReplaceTempView("orders")
products_df = spark.read \
  .format("org.apache.spark.sql.cassandra") \
  .options(table="products", keyspace="ecommerce") \
  .load()
products_df.createOrReplaceTempView("products")

# Run SQL aggregations:
result = spark.sql("""
  SELECT
    status,
    COUNT(*)        AS order_count,
    SUM(total)      AS revenue,
    AVG(total)      AS avg_order_value,
    MAX(total)      AS max_order
  FROM orders
  GROUP BY status
  ORDER BY revenue DESC
""")
result.show()

# Output:
# +---------+-----------+----------+---------------+---------+
# |   status|order_count|   revenue|avg_order_value|max_order|
# +---------+-----------+----------+---------------+---------+
# | shipped |    124500 | 9234500.0|          74.15| 2499.99 |
# | pending |     18200 | 1360540.0|          74.75|  499.99 |
# +---------+-----------+----------+---------------+---------+

Filtering Pushed Down to Cassandra

The connector pushes compatible WHERE clause filters down to Cassandra, so only matching rows travel over the network. This avoids full-table scans when a partition key filter is available.

# Filter by partition key — pushed down to Cassandra:
customer_orders = spark.read \
  .format("org.apache.spark.sql.cassandra") \
  .options(table="orders_by_customer", keyspace="ecommerce") \
  .load() \
  .filter("customer_id = '11111111-1111-1111-1111-111111111111'")

# Cassandra handles the partition lookup; Spark processes the result.

Writing Spark Results Back to Cassandra

from pyspark.sql.functions import col, current_timestamp
from uuid import uuid4

# Compute a summary DataFrame:
summary_df = orders_df.groupBy("customer_id") \
  .agg(
    {"total": "sum", "order_id": "count"}
  ) \
  .withColumnRenamed("sum(total)",    "lifetime_value") \
  .withColumnRenamed("count(order_id)", "total_orders")

# Write back to a Cassandra summary table:
summary_df.write \
  .format("org.apache.spark.sql.cassandra") \
  .options(table="customer_lifetime_value", keyspace="ecommerce") \
  .mode("append") \
  .save()

Spark Streaming + Cassandra (Structured Streaming)

# Read from Kafka topic (real-time events):
stream_df = spark.readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "kafka-broker:9092") \
  .option("subscribe", "sensor-readings") \
  .load()

from pyspark.sql.functions import from_json, col
from pyspark.sql.types import StructType, StringType, DoubleType

schema = StructType() \
  .add("sensor_id", StringType()) \
  .add("value", DoubleType()) \
  .add("day", StringType())

parsed = stream_df.select(
  from_json(col("value").cast("string"), schema).alias("data")
).select("data.*")

# Write each micro-batch to Cassandra:
def write_to_cassandra(batch_df, batch_id):
  batch_df.write \
    .format("org.apache.spark.sql.cassandra") \
    .options(table="sensor_readings", keyspace="iot") \
    .mode("append") \
    .save()

query = parsed.writeStream \
  .foreachBatch(write_to_cassandra) \
  .outputMode("append") \
  .trigger(processingTime="10 seconds") \
  .start()

query.awaitTermination()

Token-Range Partitioning for Performance

The connector splits Cassandra reads across Spark partitions by token range, aligning Spark parallelism with Cassandra's ring topology. Each Spark partition reads from a subset of Cassandra nodes.

Cassandra ring: 4 nodes, each with 256 token ranges

Spark parallelism = 8 partitions:
  Partition 0 → Token range slice 0–1023 → reads from Node A
  Partition 1 → Token range slice 1024–2047 → reads from Node B
  ...
  Partition 7 → Token range slice 7168–8191 → reads from Node D

Each Spark executor reads from the nearest Cassandra node → low latency

# Tune the number of Spark partitions for Cassandra reads:
spark.conf.set("spark.cassandra.input.split.size_in_mb", "64")

Spark + Cassandra Architecture Summary

Cassandra (operational store)
      │ DataStax Spark Connector
      ▼
Spark (analytical engine)
  ├── Batch jobs (daily reports, ML training, ETL)
  ├── SQL queries (ad-hoc analytics via Spark SQL)
  └── Structured Streaming (real-time aggregations)
      │ results written back
      ▼
Cassandra (serving layer for dashboards and APIs)

Summary

The DataStax Spark Cassandra Connector turns Cassandra tables into Spark DataFrames and vice versa. Spark reads Cassandra data in parallel across token ranges, making full-table analytics feasible. Filter pushdown keeps network traffic minimal when partition key filters are available. Write Spark results back to Cassandra with a single DataFrame write call. Use Structured Streaming with foreachBatch to build real-time pipelines that aggregate Kafka event streams and store results in Cassandra for low-latency serving.