### Create Pushdown Example Table

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/data_source_v1.md

Creates a keyspace and a table named 'pushdownexample' for demonstrating pushdown filters. Ensure the keyspace and table do not already exist or handle their creation appropriately.

```sql
CREATE KEYSPACE IF NOT EXISTS pushdowns WITH replication = { 'class' : 'SimpleStrategy', 'replication_factor' : 3 };
USE pushdowns;

CREATE TABLE pushdownexample (
    partitionkey1 BIGINT,
    partitionkey2 BIGINT,
    partitionkey3 BIGINT,
    clusterkey1   BIGINT,
    clusterkey2   BIGINT,
    clusterkey3   BIGINT,
    regularcolumn BIGINT,
    PRIMARY KEY ((partitionkey1, partitionkey2, partitionkey3), clusterkey1, clusterkey2, clusterkey3)
);
```

--------------------------------

### Create Cassandra Keyspace and Table

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/2_loading.md

Example CQL statements to create a keyspace and a table in Cassandra for testing purposes.

```sql
CREATE KEYSPACE test WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1 };
CREATE TABLE test.words (word text PRIMARY KEY, count int);
```

--------------------------------

### Create Cassandra UDT and Table

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/2_loading.md

Example of creating a Cassandra User Defined Type (UDT) and a table using it.

```sql
CREATE TYPE test.address (city text, street text, number int);
CREATE TABLE test.companies (name text PRIMARY KEY, address FROZEN<address>);
```

--------------------------------

### Run Named Cassandra Container and Get IP

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/13_spark_shell.md

Start a Cassandra container with a specific name and then retrieve its IP address using 'docker inspect'.

```bash
$ docker run --name cassie -d tobert/cassandra
$ docker ps
CONTAINER ID        IMAGE               COMMAND                  CREATED             STATUS              PORTS                                               NAMES
3e9a39418f6a        tobert/cassandra    "/bin/cassandra-docke"   2 seconds ago       Up 2 seconds        7000/tcp, 7199/tcp, 9042/tcp, 9160/tcp, 61621/tcp   cassie
$ CASSANDRA_CONTAINER_IP=$(docker inspect -f '{{ .NetworkSettings.IPAddress }}' cassie)
```

--------------------------------

### Start Spark Shell with Connector

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/13_spark_shell.md

Launch the Spark Shell, specifying master address, connector JARs, packages, and Cassandra connection host. Omit --master for local mode.

```bash
cd spark/install/dir
#Include the --master if you want to run against a spark cluster and not local mode
./bin/spark-shell [--master sparkMasterAddress] --jars yourAssemblyJar --packages com.datastax.spark:spark-cassandra-connector_2.12:3.5.1 --conf spark.cassandra.connection.host=yourCassandraClusterIp
```

--------------------------------

### Start PySpark Shell with Cassandra Connector

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/15_python.md

Include the Spark Cassandra Connector Maven artifact when starting the PySpark shell. This is the preferred method for enabling Cassandra access.

```bash
./bin/pyspark \
  --packages com.datastax.spark:spark-cassandra-connector_2.12:3.5.1 \
  --conf spark.sql.extensions=com.datastax.spark.connector.CassandraSparkExtensions
```

--------------------------------

### Start Spark in Standalone Mode

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/13_1_setup_spark_shell.md

Navigate to the Spark directory and run this script to start Spark master and worker processes in standalone mode.

```bash
cd spark*
./sbin/start-all.sh
```

--------------------------------

### Start Spark Shell with Dockerized Cassandra IP

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/13_spark_shell.md

Launch the Spark Shell, connecting to a Cassandra instance running in a Docker container by using its obtained IP address.

```bash
cd spark/install/dir
#Include the --master if you want to run against a spark cluster and not local mode
./bin/spark-shell [--master sparkMasterAddress] --jars yourAssemblyJar --conf spark.cassandra.connection.host=$CASSANDRA_CONTAINER_IP
```

--------------------------------

### Create Cassandra Tables for Examples

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/16_partitioning.md

SQL schema definitions for the 'purchases' and 'users' tables in the 'doc_example' keyspace. The 'purchases' table has a composite partition key (userid, purchaseid, objectid), while the 'users' table has a single partition key (userid).

```sql
CREATE KEYSPACE doc_example WITH replication = 
    {'class': 'SimpleStrategy', 'replication_factor': '1'}  AND durable_writes = true;

CREATE TABLE doc_example.purchases (
    userid int,
    purchaseid int,
    objectid text,
    amount int,
    PRIMARY KEY (userid, purchaseid, objectid) // Partition Key "userid"
);

CREATE TABLE doc_example.users (
    userid int PRIMARY KEY, // Partition key "userid"
    name text,
    zipcode int
);
```

--------------------------------

### Run Unit and Integration Tests

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/README.md

Execute unit and integration tests for the Spark Cassandra Connector using SBT. Integration tests require CCM to be installed.

```bash
./sbt/sbt test
./sbt/sbt it:test
```

--------------------------------

### Create Cassandra Collection Table

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/2_loading.md

Example of creating a Cassandra table with a SET<text> column using cqlsh.

```sql
CREATE TABLE test.users (username text PRIMARY KEY, emails SET<text>);
INSERT INTO test.users (username, emails) 
     VALUES ('someone', {'someone@email.com', 's@email.com'});
```

--------------------------------

### Start Spark Streaming Computation

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/8_streaming.md

Initiate the Spark Streaming computation after defining the streams and transformations.

```scala
ssc.start()
```

--------------------------------

### Direct Join Example with Explain

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/14_data_frames.md

Demonstrates a direct join between a Spark DataFrame and a Cassandra table, showing how the join is optimized. The join condition must fully restrict the partition key, and key types must match.

```sql
CREATE TABLE ks.kv (
    k int PRIMARY KEY,
    v int)

```

```scala
val range = spark.range(1, 1000).selectExpr("cast(id as int) key") // note the key type must match PRIMARY KEY type
val joinTarget = spark.read.table("myCatalog.ks.kv")

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

```

```text
== Physical Plan ==
*(2) Project [key#2, k#8, v#9]
+- Cassandra Direct Join [k = key#2] ks.kv - Reading (k, v) Pushed {}
   +- *(1) Project [cast(id#0L as int) AS key#2]
      +- *(1) Range (1, 1000, step=1, splits=12)

```

--------------------------------

### Create Cassandra Vector Table

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/2_loading.md

Example of creating a Cassandra table with a vector column using cqlsh.

```sql
CREATE TABLE test.things (id int PRIMARY KEY, name text, features vector<float, 3>);
INSERT INTO test.things (id, name, features) VALUES (1, 'a', [1.0, 2.0, 3.0]);
INSERT INTO test.things (id, name, features) VALUES (2, 'b', [2.2, 2.1, 2.0]);
INSERT INTO test.things (id, name, features) VALUES (3, 'c', [1.0, 1.5, 4.0]);
```

--------------------------------

### Read TTL and WRITETIME from Cassandra

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/14_data_frames.md

Example of selecting TTL and WRITETIME for a column in Cassandra using Spark SQL extensions. Ensure the table and keyspace are correctly specified.

```sql
SELECT TTL(v) as t, WRITETIME(t) FROM casscatalog.ksname.testTable WHERE k = 1

```

--------------------------------

### Pushdown Clustering Keys (Mixed Predicates)

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/data_source_v1.md

Filters a DataFrame by including equality and range predicates on clustering keys. This example shows that filters can be pushed down when the last predicate is a range and others are equality.

```scala
df.filter("clusterkey1 = 1 AND clusterkey2 > 0 AND clusterkey2 < 10").show()
```

--------------------------------

### Join RDD with Cassandra RDD on Partition Key (No Shuffle)

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/16_partitioning.md

When joining a non-Cassandra RDD with a Cassandra RDD, if the Cassandra RDD is keyed on its partition key, the join can occur without shuffling the Cassandra RDD. This example demonstrates the debug string and count time for such an optimized join.

```scala
import com.datastax.spark.connector

val ks = "doc_example"
val rdd = { sc.cassandraTable[(String, Int)](ks, "users")
  .select("name" as "_1", "zipcode" as "_2", "userid")
  .keyBy[Tuple1[Int]]("userid")
}

val joinrdd = sc.parallelize(1 to 10000).map(x => (Tuple1(x),x) ).join(rdd)
joinrdd.toDebugString
//(1) MapPartitionsRDD[26] at join at <console>:35 []
// |  MapPartitionsRDD[25] at join at <console>:35 []
// |  CoGroupedRDD[24] at join at <console>:35 []
// +-(8) MapPartitionsRDD[23] at map at <console>:35 []
// |  |  ParallelCollectionRDD[22] at parallelize at <console>:35 []
// |  CassandraTableScanRDD[16] at RDD at CassandraRDD.scala:15 []
joinrdd.count
//16/04/08 17:25:47 INFO DAGScheduler: Job 14 finished: count at <console>:37, took 5.339490 s
```

--------------------------------

### Read Cassandra Table into JavaPairRDD with Column Mapping

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/7_java_api.md

Obtain a CassandraJavaPairRDD by specifying column readers for key and value. This example maps the first column to Integer key and the second to String value.

```java
CassandraJavaPairRDD<Integer, String> rdd1 = javaFunctions(sc)
    .cassandraTable("ks", "people", mapColumnTo(Integer.class), mapColumnTo(String.class))
    .select("id", "name");
```

--------------------------------

### Create Keyspace, Table, and Insert Data (CQL)

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/7_java_api.md

Sample CQL commands to set up a keyspace, table, and insert data for testing the Java API.

```cql
create keyspace if not exists ks with replication = {'class':'SimpleStrategy', 'replication_factor':1};

create table if not exists ks.people (
  id int primary key,
  name text,
  birth_date timestamp
);

create index on ks.people (name);

insert into ks.people (id, name, birth_date) values (10, 'Catherine', '1987-12-02');
insert into ks.people (id, name, birth_date) values (11, 'Isadora', '2004-09-08');
insert into ks.people (id, name, birth_date) values (12, 'Anna', '1970-10-02');
```

--------------------------------

### Load Spark Shell with Connector and Configuration

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/0_quick_start.md

Launch spark-shell including the connector, its dependencies, and custom Spark configurations for Cassandra connection host and SQL extensions.

```bash
$SPARK_HOME/bin/spark-shell --conf spark.cassandra.connection.host=127.0.0.1 \
                                --packages com.datastax.spark:spark-cassandra-connector_2.12:3.5.1 \
                                --conf spark.sql.extensions=com.datastax.spark.connector.CassandraSparkExtensions
```

--------------------------------

### Insert Data into Cassandra Table

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/2_loading.md

Example CQL statements to insert data into the 'words' table.

```sql
INSERT INTO test.words (word, count) VALUES ('foo', 20);
INSERT INTO test.words (word, count) VALUES ('bar', 20);
```

--------------------------------

### Initialize Cassandra and Create Schema

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/13_spark_shell.md

Initialize a CassandraConnector and execute CQL commands to create a keyspace and a table within the Spark Shell.

```scala
val c = CassandraConnector(sc.getConf)
c.withSessionDo ( session => session.execute("CREATE KEYSPACE test WITH replication={'class':'SimpleStrategy', 'replication_factor':1}"))
c.withSessionDo ( session => session.execute("CREATE TABLE test.fun (k int PRIMARY KEY, v int)"))
```

--------------------------------

### Using Cassandra Connector with Session

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/1_connecting.md

Demonstrates how to obtain a Cassandra session using the connector and execute CQL statements. This is useful for administrative tasks like creating keyspaces or tables.

```scala
import com.datastax.spark.connector.cql.CassandraConnector

CassandraConnector(conf).withSessionDo {
  session =>
  session.execute("CREATE KEYSPACE test2 WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1 }")
  session.execute("CREATE TABLE test2.words (word text PRIMARY KEY, count int)")
}
```

--------------------------------

### Build Main Artifacts

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/12_building_and_artifacts.md

Run the `package` command in sbt to generate the main library package jars.

```bash
sbt/sbt package
```

--------------------------------

### Launch Spark Shell with Connector (spark-shell)

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/0_quick_start.md

Use spark-shell with the --packages argument to include the connector and its dependencies. This command places the connector on the Spark Driver and Executor paths.

```bash
$SPARK_HOME/bin/spark-shell --packages com.datastax.spark:spark-cassandra-connector_2.12:3.5.1
```

--------------------------------

### Get Cassandra Container IP Address

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/13_spark_shell.md

Retrieve the IP address assigned by Docker to the running Cassandra container.

```bash
CASSANDRA_CONTAINER_IP=$(docker inspect -f '{{ .NetworkSettings.IPAddress }}' $CASSANDRA_CONTAINER_ID)
```

--------------------------------

### Print Spark Streaming Word Counts

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/8_streaming.md

Print the computed word counts to the console and start the Spark Streaming computation.

```scala
wordCounts.print()
    ssc.start()
    ssc.awaitTermination() // Wait for the computation to terminate
```

--------------------------------

### Server-Side Column Pruning with Select

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/7_java_api.md

Perform server-side column pruning by using the select method on CassandraJavaRDD to fetch only specified columns. The example then processes the result.

```java
JavaRDD<String> rdd = javaFunctions(sc).cassandraTable("ks", "people")
        .select("id").map(new Function<CassandraRow, String>() {
            @Override
            public String call(CassandraRow cassandraRow) throws Exception {
                return cassandraRow.toString();
            }
        });
System.out.println("Data with only 'id' column fetched: \n" + StringUtils.join(rdd.toArray(), "\n"));
```

--------------------------------

### Build Assembly Jar

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/12_building_and_artifacts.md

Run the `assembly` command in sbt to generate a fat jar containing the Spark Cassandra Connector and its dependencies.

```bash
sbt/sbt assembly
```

--------------------------------

### Create Cassandra Keyspace with SimpleStrategy

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/14_data_frames.md

Create a Cassandra keyspace using SparkSQL with SimpleStrategy. Ensure replication options are valid.

```sql
CREATE DATABASE IF NOT EXISTS casscatalog.ksname 
  WITH DBPROPERTIES (class='SimpleStrategy', replication_factor='5')
```

--------------------------------

### Import CassandraJavaUtil Statically (Java)

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/7_java_api.md

Import all methods from CassandraJavaUtil statically for convenient access to the Connector's Java API.

```java
import static com.datastax.spark.connector.japi.CassandraJavaUtil.*;
```

--------------------------------

### Creating Datasets using Read Commands

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/data_source_v1.md

Programmatically create a Dataset by invoking a `read` command on the SparkSession, specifying the Cassandra format and providing options as a map. Data loading is lazy and occurs only when an action is called.

```scala
val df = spark
  .read
  .format("org.apache.spark.sql.cassandra")
  .options(Map( "table" -> "words", "keyspace" -> "test"))
  .load()
```

--------------------------------

### Saving JavaRDD of Tuples to Cassandra with Custom Mapping

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/7_java_api.md

Save a JavaRDD of tuples to a Cassandra table. This example demonstrates using mapTupleToRow for custom mapping and specifying columns with withColumnSelector.

```java
CassandraJavaUtil.javaFunctions(sc.makeRDD(Arrays.asList(tuple)))
        .writerBuilder("cassandra_java_util_spec", "test_table_4", mapTupleToRow(
                String.class,
                Integer.class,
                Double.class
        )).withColumnSelector(someColumns("stringCol", "intCol", "doubleCol"))
        .saveToCassandra()
```

--------------------------------

### Run Cassandra Container with Volume

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/13_spark_shell.md

Instantiate a Cassandra Docker container, mounting a host volume for data persistence. The container ID is stored for later use.

```bash
mkdir /srv/cassandra
CASSANDRA_CONTAINER_ID=`docker run -d -v /srv/cassandra:/data tobert/cassandra`
```

--------------------------------

### Server-Side Filtering with 'where' Clause in Java

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/7_java_api.md

Use the 'where' clause for server-side filtering when reading data from Cassandra. This example filters records where the 'name' column equals 'Anna'.

```java
JavaRDD<String> rdd = javaFunctions(sc).cassandraTable("ks", "people")
        .where("name=?", "Anna").map(new Function<CassandraRow, String>() {
            @Override
            public String call(CassandraRow cassandraRow) throws Exception {
                return cassandraRow.toString();
            }
        });
System.out.println("Data filtered by the where clause (name='Anna'): \n" + StringUtils.join(rdd.toArray(), "\n"));
```

--------------------------------

### PySpark DataFrame Integration with Cassandra

Source: https://context7.com/apache/cassandra-spark-connector/llms.txt

Integrate Cassandra data with PySpark DataFrames using the Catalog API or legacy format string. Ensure the connector is added as a package when starting PySpark.

```python
# Start PySpark with the connector
# ./bin/pyspark \
#   --packages com.datastax.spark:spark-cassandra-connector_2.12:3.5.1 \
#   --conf spark.sql.extensions=com.datastax.spark.connector.CassandraSparkExtensions

# Register a Catalog
spark.conf.set("spark.sql.catalog.myCatalog", "com.datastax.spark.connector.datasource.CassandraCatalog")

# Read via Catalog
spark.read.table("myCatalog.myKs.myTab").show()

# Write via Catalog
spark.range(1, 10)
    .selectExpr("id as k")
    .write
    .mode("append")
    .saveAsTable("myCatalog.myKs.myTab")

# Read without Catalog (legacy format)
spark.read
    .format("org.apache.spark.sql.cassandra")
    .options(table="kv", keyspace="test")
    .load().show()

# Write without Catalog
df.write
    .format("org.apache.spark.sql.cassandra")
    .mode("append")
    .options(table="kv", keyspace="test")
    .save()

# Pass dotted keys via a dictionary (Python workaround for period in key names)
load_options = {
    "table": "kv",
    "keyspace": "test",
    "spark.cassandra.input.split.size_in_mb": "10"
}
spark.read.format("org.apache.spark.sql.cassandra").options(**load_options).load().show()
```

--------------------------------

### Create Cassandra Keyspace with NetworkTopologyStrategy

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/14_data_frames.md

Create a Cassandra keyspace using SparkSQL with NetworkTopologyStrategy. Specify the Cassandra replication option.

```sql
CREATE DATABASE IF NOT EXISTS casscatalog.ntsname 
  WITH DBPROPERTIES (class='NetworkTopologyStrategy', cassandra='1')
```

--------------------------------

### Read Cassandra Vector Type with Spark SQL

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/14_data_frames.md

Query data including Cassandra vector types using Spark SQL. This example demonstrates specifying a predicate on a vector column.

```sql
SELECT name, features FROM casscatalog.test.things WHERE features = array(float(1), float(1.5), float(4))
```

--------------------------------

### Show Cassandra Namespaces and Tables

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/0_quick_start.md

List all available namespaces (keyspaces) in the 'mycatalog' and tables within the 'mycatalog.testks' keyspace using Spark SQL.

```scala
spark.sql("SHOW NAMESPACES FROM mycatalog").show
spark.sql("SHOW TABLES FROM mycatalog.testks").show
```

--------------------------------

### Define Case Class for Per-Row TTL Example

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/5_saving.md

This case class defines the structure of RDD elements, including a `ttl` field, which will be used to set the TTL for individual rows when saving to Cassandra.

```scala
case class KeyValueWithTTL(key: Int, group: Long, value: String, ttl: Int)

val rdd = sc.makeRDD(Seq(
  KeyValueWithTTL(1, 1L, "value1", 100), 
  KeyValueWithTTL(2, 2L, "value2", 200), 
  KeyValueWithTTL(3, 3L, "value3", 300)))
```

--------------------------------

### Set Multiple Connector Options

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/data_source_v1.md

Shows how to set multiple connector-specific options by chaining `sqlOption` calls and concatenating the resulting maps.

```scala
options(CassandraConnectorConf.ReadTimeoutParam.sqlOption("7000") ++ ReadConf.TaskMetricParam.sqlOption(true))
```

--------------------------------

### Create Dataset using Format Helper Functions

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/data_source_v1.md

Creates a Spark DataFrame from a Cassandra table using the `cassandraFormat` helper function. This method simplifies the read configuration.

```scala
import org.apache.spark.sql.cassandra._

val df = spark
  .read
  .cassandraFormat("words", "test")
  .load()
 
//Loading an Dataset using a format helper and a option helper
val df = spark
  .read
  .cassandraFormat("words", "test")
  .options(ReadConf.SplitSizeInMBParam.option(32))
  .load()
```

--------------------------------

### Write Cassandra Vector Type with Spark SQL

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/14_data_frames.md

Insert data including Cassandra vector types into a table using Spark SQL. This example shows how to construct and insert an array of floats.

```sql
INSERT INTO casscatalog.test.things (id, name, features) VALUES (9, 'x', array(2, 3, 4))
```

--------------------------------

### Generic CassandraRow Value Access

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/2_loading.md

Demonstrates using the generic `get` method on `CassandraRow` to retrieve column values by specifying the desired return type directly. This allows for flexible type querying.

```scala
firstRow.get[Int]("count")                   // 20       
firstRow.get[Long]("count")                  // 20L
firstRow.get[BigInt]("count")                // BigInt(20)
firstRow.get[java.math.BigInteger]("count")  // BigInteger(20)
```

--------------------------------

### Load Cassandra Table using Catalog Reference

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/15_python.md

Configure a Spark SQL catalog for Cassandra and then load a table using the catalog reference. Requires Spark SQL extensions to be enabled.

```python
spark.conf.set("spark.sql.catalog.myCatalog", "com.datastax.spark.connector.datasource.CassandraCatalog")
spark.read.table("myCatalog.myKs.myTab").show()
```

--------------------------------

### Handle Nullable Cassandra Data

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/2_loading.md

Illustrates how to safely access potentially null Cassandra column values using `Option` types in Scala to prevent `NullPointerException`. Use `getIntOption` or the generic `get[Option[T]]`.

```scala
firstRow.getIntOption("count")        // Some(20)
firstRow.get[Option[Int]]("count")    // Some(20)
```

--------------------------------

### Submit Spark Application with Connector (spark-submit)

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/0_quick_start.md

Use spark-submit with the --packages argument to include the connector and its dependencies. This command places the connector on the Spark Driver and Executor paths.

```bash
$SPARK_HOME/bin/spark-submit --packages com.datastax.spark:spark-cassandra-connector_2.12:3.5.1
```

--------------------------------

### Obtain Cassandra Table as RDD

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/2_loading.md

Use `cassandraTable` on the SparkContext to get an RDD representing a Cassandra table. This RDD will contain `CassandraRow` objects by default. Ensure the `com.datastax.spark.connector._` import is present for implicit functions.

```scala
import com.datastax.spark.connector._ //Loads implicit functions
sc.cassandraTable("keyspace name", "table name")
```

--------------------------------

### Implicit Read Configuration for Cassandra Table

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/2_loading.md

Shows how to configure read parameters for a Cassandra table using implicit `ReadConf` values. This avoids repetitive configuration.

```scala
object ReadConfigurationOne {
  implicit val readConf = ReadConf(100,100)
}
import ReadConfigurationOne._
val rdd = sc.cassandraTable("write_test","collections")
rdd.readConf
//com.datastax.spark.connector.rdd.ReadConf = ReadConf(100,100,LOCAL_ONE,true)
```

```scala
implicit val anotherConf = ReadConf(200,200)
val rddWithADifferentConf = sc.cassandraTable("write_test","collections")
rddWithADifferentConf.readConf
//com.datastax.spark.connector.rdd.ReadConf = ReadConf(200,200,LOCAL_ONE,true)
```

--------------------------------

### Build Connector for Scala 2.11

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/12_building_and_artifacts.md

Use the `++2.11.12` switch with sbt to build artifacts for Scala 2.11.

```bash
sbt/sbt ++2.11.12 package
```

--------------------------------

### Group by Key Without Cassandra Partitioner (Shuffle)

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/16_partitioning.md

Performing a groupByKey operation on an RDD that does not have a partitioner (e.g., after an empty map operation) will result in a shuffle, increasing processing time. This example shows the debug string and count time for such a scenario.

```scala
val ks = "doc_example"

//Empty map will remove the partitioner
val rdd = { sc.cassandraTable[(Int, Int)](ks, "purchases")
  .select("purchaseid" as "_1", "amount" as "_2", "userid", "objectid")
  .keyBy[(Int, String)]("userid","objectid")
}.map( x => x)

rdd.partitioner
//res2: Option[org.apache.spark.Partitioner] = None

rdd.groupByKey.toDebugString
//res3: String =
//(4) ShuffledRDD[11] at groupByKey at <console>:35 []
// +-(4) MapPartitionsRDD[10] at map at <console>:35 []
//    |  CassandraTableScanRDD[9] at RDD at CassandraRDD.scala:15 []

rdd.groupByKey.count
//16/04/08 16:28:23 INFO DAGScheduler: Job 1 finished: count at <console>:35, took 46.550789 s
```

--------------------------------

### Generate Reference Document

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/README.md

Generate the Reference Document for the Spark Cassandra Connector. The output location defaults to doc/reference.md.

```bash
./sbt/sbt spark-cassandra-connector-unshaded/run (outputLocation)
```

--------------------------------

### Join RDD Without Cassandra Partitioner (Shuffle)

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/16_partitioning.md

If the Cassandra RDD is not keyed by its partition key (e.g., after an empty map operation), joining it with another RDD will require a shuffle of the Cassandra RDD, leading to increased processing time. This example illustrates the debug string and count time for a join operation that incurs a shuffle.

```scala
//Use an empty map to drop the partitioner
val rddnopart = { sc.cassandraTable[(String, Int)](ks, "users")
                  .select("name" as "_1", "zipcode" as "_2", "userid")
                  .keyBy[Tuple1[Int]]("userid").map( x => x)
}

val joinnopart = sc.parallelize(1 to 10000).map(x => (Tuple1(x),x) ).join(rddnopart)
joinnopart.toDebugString
//res18: String =
//(8) MapPartitionsRDD[73] at join at <console>:36 []
// |  MapPartitionsRDD[72] at join at <console>:36 []
// |  CoGroupedRDD[71] at join at <console>:36 []
// +-(8) MapPartitionsRDD[70] at map at <console>:36 []
// |  |  ParallelCollectionRDD[69] at parallelize at <console>:36 []
// +-(1) MapPartitionsRDD[68] at map at <console>:34 []
//    |  CassandraTableScanRDD[61] at RDD at CassandraRDD.scala:15 []

joinnopart.count
//16/04/08 17:30:04 INFO DAGScheduler: Job 19 finished: count at <console>:37, took 6.308165 s
```

--------------------------------

### Submit a Fat Jar Containing the Connector

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/17_submitting.md

Build a fat (uber) jar that includes the Spark Cassandra Connector main artifact and all its dependencies. This allows the Spark application to be submitted without any extra `spark-submit` options, as all necessary classes are bundled within the application jar. This approach is not well-suited for `spark-shell`.

--------------------------------

### Configure Cassandra Connector Metrics Source

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/11_metrics.md

Add this configuration to your `metrics.properties` file to enable Codahale metrics for the Cassandra Connector in both executor and driver.

```properties
executor.source.cassandra-connector.class=org.apache.spark.metrics.CassandraConnectorSource
driver.source.cassandra-connector.class=org.apache.spark.metrics.CassandraConnectorSource
```

--------------------------------

### Configure SparkContext with SparkConf

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/FAQ.md

For applications not using `spark-submit`, set configuration options directly in the `SparkConf` object when creating your `SparkContext`. This is the recommended approach for manual configuration.

```scala
val conf = SparkConf()
  .set("Option","Value")
  ...
  
val sc = SparkContext(conf)
```

--------------------------------

### Changing Cluster/Keyspace Level Properties and Loading DataFrames

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/data_source_v1.md

Demonstrates how to set cluster and keyspace level properties using `setCassandraConf` and then load a DataFrame with these configurations. Overrides are applied based on specificity, with table-level options taking precedence.

```scala
spark.setCassandraConf("ClusterOne", "ks1", ReadConf.SplitSizeInMBParam.option(32))
spark.setCassandraConf("default", "test", ReadConf.SplitSizeInMBParam.option(128))

val df = spark
  .read
  .format("org.apache.spark.sql.cassandra")
  .options(Map( "table" -> "words", "keyspace" -> "test"))
  .load() // This Dataset will use a spark.cassandra.input.size of 128

val otherdf =  spark
  .read
  .format("org.apache.spark.sql.cassandra")
  .options(Map( "table" -> "words", "keyspace" -> "test" , "cluster" -> "ClusterOne"))
  .load() // This Dataset will use a spark.cassandra.input.size of 32

val lastdf = spark
  .read
  .format("org.apache.spark.sql.cassandra")
  .options(Map(
    "table" -> "words",
    "keyspace" -> "test" ,
    "cluster" -> "ClusterOne",
    "spark.cassandra.input.split.size_in_mb" -> 48
    )
  ).load() // This Dataset will use a spark.cassandra.input.split.size of 48
```

--------------------------------

### Create Cassandra Keyspace and Table

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/0_quick_start.md

Use Spark SQL to create a new keyspace with specified replication strategy and a table with a partition key. The 'USING cassandra' clause specifies the Cassandra data source.

```scala
spark.sql("CREATE DATABASE IF NOT EXISTS mycatalog.testks WITH DBPROPERTIES (class = \'SimpleStrategy\',replication_factor = \'1\')")
spark.sql("CREATE TABLE mycatalog.testks.testtab (key Int, value STRING) USING cassandra PARTITIONED BY (key)")
```

--------------------------------

### Create Dataset using Read Command

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/data_source_v1.md

Reads data from a Cassandra table into a Spark DataFrame using the `read` command with specified table and keyspace.

```scala
val df = spark
  .read
  .format("org.apache.spark.sql.cassandra")
  .options(Map( "table" -> "words", "keyspace" -> "test" ))
  .load()
  
df.show
```

--------------------------------

### Keying RDD by Partition Key Creates CassandraPartitioner

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/16_partitioning.md

Demonstrates how keying a CassandraTableScanRDD by the table's partition key ('userid' in this case) results in the creation of a CassandraPartitioner. This enables Spark to leverage Cassandra's data distribution.

```scala
import com.datastax.spark.connector._

val ks = "doc_example"
val rdd = { sc.cassandraTable[(String, Int)](ks, "users")
  .select("name" as "_1", "zipcode" as "_2", "userid")
  .keyBy[Tuple1[Int]]("userid")
}

//16/04/08 11:19:52 DEBUG CassandraPartitioner: Building Partitioner with mapping
//Vector((userid,userid))
//for table TableDef(doc_example,users,ArrayBuffer(ColumnDef(userid,PartitionKeyColumn,IntType)),ArrayBuffer(),ArrayBuffer(ColumnDef(name,RegularColumn,VarCharType), ColumnDef(zipcode,RegularColumn,IntType)),List(),false)
//16/04/08 11:19:52 DEBUG CassandraTableScanRDD: Made partitioner Some(com.datastax.spark.connector.rdd.partitioner.CassandraPartitioner@6709eccf) for CassandraTableScanRDD[5] at RDD at CassandraRDD.scala:15
//16/04/08 11:19:52 DEBUG CassandraTableScanRDD: Assigning new Partitioner com.datastax.spark.connector.rdd.partitioner.CassandraPartitioner@6709eccf

rdd.partitioner
//res3: Option[org.apache.spark.Partitioner] = Some(com.datastax.spark.connector.rdd.partitioner.CassandraPartitioner@94515d3e)
```

--------------------------------

### Enable Cassandra Streaming Connector

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/8_streaming.md

Import the necessary Cassandra Spark Connector functions for streaming operations.

```scala
import com.datastax.spark.connector.streaming._
```

--------------------------------

### Feature Branching and Merging Strategy

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/developers.md

Follow this workflow for introducing new features, ensuring compatibility across different connector versions. It involves creating feature branches for each target version and merging sequentially.

```shell
git fetch apache
git checkout -b SPARKC-9999-b2.5 apache/b2.5
# do the work, commit
git push origin SPARKC-9999-b2.5

# Forward merge on the next version:
git checkout -b SPARKC-9999-b3.0 apache/b3.0
git merge SPARKC-9999-b2.5
# Resolve conflict, if any
# Push the new feature branch:
git push origin SPARKC-9999-b3.0

# Forward merge on the next version:
git checkout -b SPARKC-9999-b3.1 apache/b3.1
git merge SPARKC-9999-b3.0
# Resolve conflict, if any
# Push the new feature branch:
git push origin SPARKC-9999-b3.1

# Repeat for b3.2, b3.3, b3.4

# Forward merge on the next version:
git checkout -b SPARKC-9999-trunk apache/trunk
git merge SPARKC-9999-b3.2
# Resolve conflict, if any
# Push the new feature branch:
git push origin SPARKC-9999-trunk

```

--------------------------------

### Pull Cassandra Docker Image

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/13_spark_shell.md

Download the Cassandra Docker image from Docker Hub.

```bash
docker pull tobert/cassandra
```

--------------------------------

### Map Cassandra Columns with Aliases for TTL and Write Time

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/7_java_api.md

Demonstrates using the as() method to alias normal columns, TTL, and write times to specific property names in the mapped object.

```java
javaFunctions(sc).cassandraTable("test", "table", mapRowTo(SomeClass.class)).select(
        column("no_alias"),
        column("simple").as("simpleProp"),
        ttl("simple").as("simplePropTTL"),
        writeTime("simple").as("simpleWriteTime"))
```

--------------------------------

### Partitioned Join Between Cassandra RDDs

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/16_partitioning.md

Demonstrates joining two Cassandra RDDs on a common partition key using `applyPartitionerFrom` to ensure they share the same partitioner, thus avoiding shuffles. The `toDebugString` and `count` methods are shown for verification.

```scala
val ks = "doc_example"
val rdd1 = {
  sc.cassandraTable[(Int, Int, String)](ks, "purchases")
    .select("purchaseid" as "_1", "amount" as "_2", "objectid" as "_3", "userid")
    .keyBy[Tuple1[Int]]("userid")
}

val rdd2 = {
  sc.cassandraTable[(String, Int)](ks, "users")
    .select("name" as "_1", "zipcode" as "_2", "userid")
    .keyBy[Tuple1[Int]]("userid")
}.applyPartitionerFrom(rdd1) // Assigns the partitioner from the first rdd to this one

val joinRDD = rdd1.join(rdd2)
joinRDD.toDebugString
//res37: String =
//(1) MapPartitionsRDD[123] at join at <console>:36 []
// |  MapPartitionsRDD[122] at join at <console>:36 []
// |  CoGroupedRDD[121] at join at <console>:36 []
// |  CassandraTableScanRDD[115] at RDD at CassandraRDD.scala:15 []
// |  CassandraTableScanRDD[120] at RDD at CassandraRDD.scala:15 []

joinRDD.count
//16/04/08 17:53:45 INFO DAGScheduler: Job 24 finished: count at <console>:39, took 27.583355 s
```

```scala
//Unpartitioned join
val rdd1nopart = {
  sc.cassandraTable[(Int, Int, String)](ks, "purchases")
    .select("purchaseid" as "_1", "amount" as "_2", "objectid" as "_3", "userid")
    .keyBy[Tuple1[Int]]("userid")
}.map(x => x)

val rdd2nopart = {
  sc.cassandraTable[(String, Int)](ks, "users")
    .select("name" as "_1", "zipcode" as "_2", "userid")
    .keyBy[Tuple1[Int]]("userid")
}.map(x => x)

val joinnopart = rdd1nopart.join(rdd2nopart)
joinnopart.toDebugString
//   res41: String =
//   (4) MapPartitionsRDD[136] at join at <console>:36 []
//    |  MapPartitionsRDD[135] at join at <console>:36 []
//    |  CoGroupedRDD[134] at join at <console>:36 []
//    +-(1) MapPartitionsRDD[128] at map at <console>:35 []
//    |  |  CassandraTableScanRDD[127] at RDD at CassandraRDD.scala:15 []
//    +-(4) MapPartitionsRDD[133] at map at <console>:35 []
//       |  CassandraTableScanRDD[132] at RDD at CassandraRDD.scala:15 []

joinnopart.count
//16/04/08 17:54:58 INFO DAGScheduler: Job 25 finished: count at <console>:39, took 40.040341 s
```

--------------------------------

### Pushdown Partition Keys (All Included)

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/data_source_v1.md

Filters a DataFrame by including all partition keys with equality predicates. This configuration allows for effective pushdown of filters to Cassandra.

```scala
df.filter("partitionkey1 = 1 AND partitionkey2 = 1 AND partitionkey3 = 1").show()
```

--------------------------------

### Configure Cassandra Catalog (DataSource V2)

Source: https://context7.com/apache/cassandra-spark-connector/llms.txt

Register a named Cassandra Catalog in your SparkSession to enable SparkSQL over Cassandra. This allows for DDL, DML, and querying using a `catalog.keyspace.table` identifier. Ensure Spark SQL extensions are enabled.

```scala
import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
  .appName("CassandraApp")
  .config("spark.cassandra.connection.host", "127.0.0.1")
  .config("spark.sql.extensions", "com.datastax.spark.connector.CassandraSparkExtensions")
  .config("spark.sql.catalog.casscatalog", "com.datastax.spark.connector.datasource.CassandraCatalog")
  .getOrCreate()

// Create a keyspace and table via SparkSQL
spark.sql("""
  CREATE DATABASE IF NOT EXISTS casscatalog.testks
  WITH DBPROPERTIES (class='SimpleStrategy', replication_factor='1')
""")

spark.sql("""
  CREATE TABLE casscatalog.testks.testtab (key INT, value STRING)
  USING cassandra
  PARTITIONED BY (key)
""")

// List keyspaces and tables
spark.sql("SHOW NAMESPACES FROM casscatalog").show()
spark.sql("SHOW TABLES FROM casscatalog.testks").show()
```

--------------------------------

### Load Cassandra Table into DataFrame

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/data_source_v1.md

Loads data from the 'pushdownexample' table in the 'pushdowns' keyspace into a Spark DataFrame. This is the initial step before applying filters.

```scala
val df = spark
  .read
  .cassandraFormat("pushdownexample", "pushdowns")
  .load()
```

--------------------------------

### Create Spark StreamingContext

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/8_streaming.md

Initialize a Spark StreamingContext with a SparkConf and a batch duration.

```scala
val ssc = new StreamingContext(sparkConf, Seconds(1))
```

--------------------------------

### Configure Profile Path for Java Driver

Source: https://github.com/apache/cassandra-spark-connector/blob/trunk/doc/reference.md

Specify a default Java Driver 4.0 Profile file for the connection. Accepts URLs and references to files distributed via spark.files.

```text
spark.cassandra.connection.config.profile.path
```