<jupyter_start><jupyter_text>What's in this exerciseBasics of how to work with Azure Cosmos DB Cassandra API from Zeppelin in batch.Section 01: Cassandra API connectionSection 02: Keyspace DDLSection 03: Table DDL  **Reference:** https://github.com/datastax/spark-cassandra-connector/blob/master/doc/1_connecting.md  PrerequisitesThe Datastax connector for Cassandra requires the Azure Comsos DB Cassandra API connection details to be initialized as part of the spark context.  When you launch a Jupyter notebook, the spark session and context are already initialized and it is not advisable to stop and reinitialize the Spark context unless with every configuration set as part of the HDInsight default Jupyter notebook start-up.  One workaround is to add the Cassandra instance details to Ambari, Spark2 service configuration directly.  This is a one-time activity that requires a Spark2 service restart.1.  Go to Ambari, Spark2 service and click on configs2.  Then go to custom spark2-defaults and add a new property with the following, and restart Spark2 service:spark.cassandra.connection.host=YOUR_COSMOSDB_ACCOUNT_NAME.cassandra.cosmosdb.azure.comspark.cassandra.connection.port=10350spark.cassandra.connection.ssl.enabled=truespark.cassandra.auth.username=YOUR_COSMOSDB_ACCOUNT_NAMEspark.cassandra.auth.password=<PASSWORD>_KEY --------- 1.0. Cassandra API connection  1.0.1. Configure dependencies<jupyter_code>%%configure -f
{ "conf": {"spark.jars.packages": "com.datastax.spark:spark-cassandra-connector_2.11:2.3.0,com.microsoft.azure.cosmosdb:azure-cosmos-cassandra-spark-helper:1.0.0" }}<jupyter_output><empty_output><jupyter_text>1.0.2. Cassandra API configuration<jupyter_code>import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import spark.implicits._
import org.apache.spark.sql.functions._
import org.apache.spark.sql.Column

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

//datastax Spark connector
import com.datastax.spark.connector._
import com.datastax.spark.connector.cql.CassandraConnector

//CosmosDB library for multiple retry
import com.microsoft.azure.cosmosdb.cassandra

// Specify connection factory for Cassandra
spark.conf.set("spark.cassandra.connection.factory", "com.microsoft.azure.cosmosdb.cassandra.CosmosDbConnectionFactory")

// Parallelism and throughput configs - increase as needed to tune your jobs
spark.conf.set("spark.cassandra.output.batch.size.rows", "1")//Do not modify this
spark.conf.set("spark.cassandra.connection.connections_per_executor_max", "10")
spark.conf.set("spark.cassandra.output.concurrent.writes", "100")
spark.conf.set("spark.cassandra.concurrent.reads", "512")
spark.conf.set("spark.cassandra.output.batch.grouping.buffer.size", "1000")
spark.conf.set("spark.cassandra.connection.keep_alive_ms", "60000000") 
spark.conf.set("spark.cassandra.output.ignoreNulls","true")<jupyter_output>Starting Spark application<jupyter_text>----- 2.0. Cassandra Keyspace DDL operations  2.0.a. Create keyspace - from Spark<jupyter_code>val cdbConnector = CassandraConnector(sc)
// Create keyspace
cdbConnector.withSessionDo(session => session.execute("CREATE KEYSPACE IF NOT EXISTS books_ks WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1 } "))<jupyter_output>res20: com.datastax.driver.core.ResultSet = ResultSet[ exhausted: true, Columns[]]<jupyter_text>2.0.b. Alter keyspace - from SparkNot supported currently.  2.0.c. Delete keyspace - from Spark<jupyter_code>val cdbConnector = CassandraConnector(sc)
//cdbConnector.withSessionDo(session => session.execute("DROP KEYSPACE books_ks"))<jupyter_output><empty_output><jupyter_text>----- 3.0. Cassandra Table DDL operations **Considerations:**&nbsp;&nbsp;-Throughput can be provisioned at a table level as part of the create table statement.&nbsp;&nbsp;-One partition key can store 10 GB of data.  &nbsp;&nbsp;-One record can be max of 2 MB in size&nbsp;&nbsp;-One partition key range can store multiple partition keys.  3.0.a. Create table<jupyter_code>val cdbConnector = CassandraConnector(sc)
cdbConnector.withSessionDo(session => session.execute("CREATE TABLE IF NOT EXISTS books_ks.books(book_id TEXT PRIMARY KEY,book_author TEXT, book_name TEXT,book_pub_year INT,book_price FLOAT) WITH cosmosdb_provisioned_throughput=4000 , WITH default_time_to_live=630720000;"))<jupyter_output>res21: com.datastax.driver.core.ResultSet = ResultSet[ exhausted: true, Columns[]]<jupyter_text>3.0.b. Alter table **Considerations:**(1) Alter table - add/change columns - on the roadmap(2) Alter provisioned throughput from Spark - on the roadmap (3) Alter table TTL - on the roadmap<jupyter_code>//val cdbConnector = CassandraConnector(sc)
//cdbConnector.withSessionDo(session => session.execute("ALTER TABLE books_ks.books WITH cosmosdb_provisioned_throughput=8000, WITH default_time_to_live=0;"))<jupyter_output><empty_output><jupyter_text>3.0.c. Drop table<jupyter_code>val cdbConnector = CassandraConnector(sc)
//cdbConnector.withSessionDo(session => session.execute("DROP TABLE IF EXISTS books_ks.books;"))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10570866141732284, 'fraction_numerical': 0.020078740157480315, 'mean_word_length': 6.321325648414986, 'pattern_counts': {'":': 2, '<': 27, '<?xml version=': 0, '>': 32, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '178579', 'license': 'MIT', 'n_tokens_mistral': 1703, 'n_tokens_neox': 1602, 'n_words': 392, 'path': 'scala/2.1-DDL-Operations.ipynb', 'repo_name': 'Azure-Samples/azure-cosmos-db-cassandra-api-spark-notebooks-jupyter', 'size': '5083'}