DOCSEARCHbyAlgolia
Submit Docs

Apache Cassandra Spark Connector

repository·trunk·Indexed 23 days ago

https://github.com/apache/cassandra-spark-connector

A high-performance connector bridging Apache Spark and Apache Cassandra, allowing developers to treat Cassandra tables as Spark RDDs, Datasets, or DataFrames. It supports server-side filtering via CQL, data locality through replica-aware partitioning, and integration with Spark SQL via the DatasourceV2 API. Recent versions include support for vector types for AI/RAG workflows and compatibility with Spark 3.5, Cassandra 2.1.5+ through 5.0, and Scala 2.12 or 2.13.

Tokens
31.7K
Snippets
81
Records
159
Agent score
84%

What's inside apache-cassandra-spark-connector

  1. Core features of the Spark-Cassandra-Connector

    trunk

    The connector provides high-performance integration between Apache Spark and Apache Cassandra with the following capabilities:

    • Data Abstraction: Exposes Cassandra tables as Spark RDDs and Datasets/DataFrames.
    • Mapping: Maps table rows to CassandraRow objects or tuples, with a customizable object mapper for user-defined classes.
    • Write Operations: Saves RDDs back to Cassandra via implicit saveToCassandra calls and deletes rows/columns via implicit deleteFromCassandra calls.
    • Optimized Joins: Provides joinWithCassandraTable for RDDs and optimized joins for Datasets/DataFrames.
    • Data Locality: Supports partitioning RDDs according to Cassandra replication using repartitionByCassandraReplica.
    • Querying: Supports server-side filtering via CQL WHERE clauses and execution of arbitrary CQL statements.
    • Advanced Types: Supports all Cassandra data types (including collections) and recently added support for vector types (for Cassandra 5.0 and Astra vectors) to support AI/RAG workflows.

  2. Save User Defined Types (UDTs)

    trunk

    To save data into a Cassandra User Defined Type (UDT), you can use either a Scala case class or the com.datastax.spark.connector.UDTValue class.

    1. Using Case Classes: Define a case class that matches the UDT structure. The parent object's property must be of the UDT case class type.
    2. Using UDTValue: Create a UDTValue instance using the UDTValue.fromMap(Map[String, Any]) factory method. This is useful when converting from generic data structures like Maps.

  3. How Cassandra connection management works

    trunk

    The connector manages connections using the following principles:

    • Initial Contact: The spark.cassandra.connection.host can be any node in the cluster. The driver fetches topology and attempts to connect to the closest node in the same data center to ensure data locality.
    • Data Center Isolation: By default, inter-datacenter communication is forbidden. The connector will not retry operations on nodes in a different data center if local nodes are down. This prevents analytics workloads from impacting real-time production workloads.
    • Connection Pooling: Connections are cached. Multiple calls to the connector within the same JVM will share the same logical connection (the underlying Java Driver Cluster and Session objects).
    • Keep Alive: Unused connections are closed after a period controlled by the spark.cassandra.connection.keep_alive_ms system property.

  4. Use Direct Joins for Cassandra Tables

    trunk

    The connector can automatically convert joins involving a Cassandra table into a joinWithCassandraTable style join (Direct Join) if it is more efficient.

    Configuration

    • Default: directJoinSetting=auto.
    • Automatic Conversion Logic: A join is converted if (table size * directJoinSizeRatio) > size of keys.
    • Manual Control: Set directJoinSetting=on to force conversion or directJoinSetting=off to disable it.

    Requirements for Direct Join

    For a join to be eligible for Direct Join conversion:

    1. At least one side of the join must be a CassandraSourceRelation.
    2. The join condition must fully restrict the partition key.
    3. The join keys' types must match the Cassandra table's primary key types.

    Example

    val range = spark.range(1, 1000).selectExpr("cast(id as int) key")
val joinTarget = spark.read.table("myCatalog.ks.kv")

range.join(joinTarget, joinTarget("k") === range("key")).explain()

  5. Use unshaded artifacts for custom shading

    trunk

    If you need to use other libraries that depend on the Cassandra Java Driver, the standard shaded artifacts will cause conflicts.

    To support this, use the spark-cassandra-connector-unshaded artifact. Note that using unshaded artifacts requires:

    1. Manually shading Guava references within your own code.
    2. Launching your application with an "uber-jar".
    3. Note: The --packages method will no longer work with unshaded artifacts.

  6. How the connector calculates Spark partitions

    trunk

    The connector determines the number of Spark partitions by dividing the estimated table size by the spark.cassandra.input.split.size_in_mb value.

    The resulting number of partitions will never be smaller than 1 + 2 * SparkContext.defaultParallelism.

    Note on Size Estimation: The connector uses the internal Cassandra system table system.size_estimates (available in Cassandra $\ge$ 2.1.5). This estimate is not perfectly accurate, especially for smaller tables.

  7. Modify CQL Collections (Append, Prepend, Remove, Overwrite)

    trunk

    By default, the connector overwrites collections (lists, sets, maps) during insertion. You can override this by applying specific behaviors to the ColumnSelector using the as method and a collection operation keyword.

    Supported operations:

    • append (lists, sets, maps)
    • prepend (lists)
    • remove (lists, sets)
    • overwrite (lists, sets, maps) — Default

    Note: remove is not supported for Maps.

    Example syntax: ("column_name" as "rdd_field" append)

  8. Group rows by partition key efficiently using spanBy

    trunk

    Because Cassandra stores data grouped by partition key, you can group data in Spark without a shuffle by using spanBy or spanByKey. These methods iterate through Spark partitions locally and start a new group whenever the key changes.

    Requirements:

    1. Data must be sequentially ordered by the clustering keys.
    2. The grouping keys must follow the natural clustering key order (e.g., if the PK is (year, month, ts), you can span by (year), (year, month), or (year, month, ts), but not by (month)).
    3. You must have enough memory to store the largest single group.

    Methods:

    • spanBy(function): Groups based on a function applied to the row.
    • keyBy(function).spanByKey: Groups based on a key extracted from the row.
    // Example: Grouping events by year and month without shuffling
sc.cassandraTable("test", "events")
  .spanBy(row => (row.getInt("year"), row.getInt("month")))

  9. Understand the project sub-projects

    trunk

    The connector is organized into several modules:

    • connector: Contains the core connector code, the Java API, and related code. This is where new features and tests should be implemented.
    • driver: Contains all code relating to the Java Driver, including connection factories and row transformers. This code is usable by applications even if Spark is not involved.
    • test-support: Contains CCM Wrapper code used for spawning clusters and managing test parallelization.

  11. Optimize Queries with Predicate Pushdown and Count Pushdown

    trunk

    The connector automatically optimizes queries in two ways:

    1. Predicate Pushdown & Column Pruning: Valid predicates (WHERE clauses) are pushed to Cassandra, and only the required columns are selected. You can verify this using .explain().
    2. Count Pushdown: Requests for row counts that do not require column values are converted into Cassandra count operations, preventing unnecessary data transfer to Spark.

    Example of a pushed-down count:

    spark.sql("SELECT Count(*) FROM mycatalog.ks.tab WHERE key = 1").explain

  12. Avoid tombstones using CassandraOption

    trunk

    To avoid creating tombstones when writing data, you can use the com.datastax.spark.connector.types.CassandraOption trait. This allows you to specify whether a column should be treated as a Value, Null (delete the value), or Unset (leave the parameter unbound in the prepared statement).

    Using Unset is particularly useful when copying data between tables, as it prevents missing columns from being treated as deletes.

    When reading, a column loaded as a CassandraOption will represent missing columns as Unset.

    sealed trait CassandraOption[+A] extends Product with Serializable
  
object CassandraOption {
  case class Value[+A](value: A) extends CassandraOption[A]
  case object Unset extends CassandraOption[Nothing]
  case object Null extends CassandraOption[Nothing]
}