pyspark rdd/dataframe not creating table in cassandra automatically

Question

After checking all sources found datastax-spark-cassandra connector supports automatic creation of table in cassandra with rdd in scala and java. For pyspark specifically another package is available to do this job -- https://github.com/anguenot/pyspark-cassandra. But even with this package unable to create table automatically. With dataframe I didn't found any option at all. Am new to pyspark and cassandra, any help is highly appreciated. Tried using only anguenot package as dependency as well. Spark version : 2.4.7 Cassandra : latest docker image

Pyspark shell >> pyspark --packages anguenot/pyspark-cassandra:2.4.0,com.datastax.spark:spark-cassandra-connector_2.11:2.5.1
>>> spark = SparkSession.builder.master('local[*]').appName('cassandra').config("spark.cassandra.connection.host", "ip").config("spark.cassandra.connection.port", "port").config("spark.cassandra.auth.username", "username").config("spark.cassandra.auth.password", "password").getOrCreate()
>>> from datetime import datetime
>>> rdd = sc.parallelize([{
...     "key": k,
...     "stamp": datetime.now(),
...     "tags": ["a", "b", "c"],
...     "options": {
...             "foo": "bar",
...             "baz": "qux",
...     }
... } for k in ["x", "y", "z"]])

>>> rdd.saveToCassandra("test", "testTable")
Traceback (most recent call last):
  File "", line 1, in 
AttributeError: 'RDD' object has no attribute 'saveToCassandra'

Alex_Ott_18627 · Answer

Generally, it's possible to create tables from Spark Cassandra Connector for RDDs (saveAsCassandraTable or saveAsCassandraTableEx) or Dataframes (createCassandraTable and createCassandraTableEx), but this functionality is available only in Scala API.

Since version 3.0, Spark Cassandra Connector supports Catalogs API (Spark 3+), so you'll be able to work with keyspace & tables (create/alter/drop) using the Spark SQL, like this:

spark.sql("""
CREATE TABLE casscatalog.ksname.testTable (
     key_1 Int, key_2 Int, key_3 Int, 
     cc1 STRING, cc2 String, cc3 String, value String) 
  USING cassandra
  PARTITIONED BY (key_1, key_2, key_3)
  TBLPROPERTIES (
    clustering_key='cc1.asc, cc2.desc, cc3.asc'
  )
""")

user_1563022 · Answer

You should import pyspark_cassandra before creating rdd:

>>> import pyspark_cassandra
>>> rdd = sc.parallelize(...)
>>> rdd.saveToCassandra("test", "testTable")

See https://github.com/anguenot/pyspark-cassandra#examples.

pyspark rdd/dataframe not creating table in cassandra automatically

2 Answers