Getting started with PySpark

PySpark prerequisites 

Insert Cassandra data 

Access the data using PySpark 

1. Start PySpark using one of the following commands:
   • Installer-Services and Package installations:

     dse pyspark

   • Installer-No Services and Tarball installations:

     install_location/bin/dse pyspark

   Note: The directory in which you run the dse Spark commands must be writable by the current user.
   The Spark prompt will appear.
2. Call the cassandraTable method to obtain an RDD representing the Cassandra table test.kv.

   rdd = sc.cassandraTable("test", "kv")

   Cassandra rows are converted to Python objects of class pyspark.sql.Row, which allows for dictionary-like lookup of column values as well as directly formatting the rows as object fields.

3. Query Cassandra.

   rdd.first()
   Row(key=2, value=u'def')
   rdd.first().key
   2
   rdd.first().value
   u'def'
   rdd.first()[0]
   2
   rdd.first()[1]
   u'def'
   rdd.collect()
   [Row(key=2, value=u'def'), Row(key=1, value=u'abc')]
   rdd.filter(lambda row: row.key > 1).collect()
   [Row(key=2, value=u'def')]
   

4. Call saveToCassandra and pass a keyspace, table, and optionally a list of columns on any RDD of dictionary or pysparksql.Row objects. For example, save the key columns from the kv table to the kv2 table.

   rdd = sc.cassandraTable("test", "kv")
   rdd.saveToCassandra("test", "kv2", ["key"])

5. In cqlsh, confirm that the keys from kv1 were added to table kv2.

   SELECT  * FROM  kv2;
   
    key | value
    ----+-------
      1 |  null
      2 |  null
   
        (2 rows)
   

6. In PySpark, copy all columns from table kv to table kv2.

   rdd.saveToCassandra("test", "kv2")

7. In cqlsh, take a look at the kv2 table to confirm that kv2 has both keys and values now.

   SELECT * FROM kv2;
   
    key | value
   -----+-------
      1 |   abc
      2 |   def
   
   (2 rows)

Insert new data into Cassandra from PySpark 

1. Distribute the current Python collection to form an RDD. The source RDD does not need to be in Cassandra.

   otherRdd = sc.parallelize([{ "key": 3, "value": "foobar" }])  

2. Save otherRdd to the kv2 table in Cassandra.

   otherRdd.saveToCassandra("test", "kv2")

   These steps add the key 3 and the value foobar to table k2.

3. In cqlsh, select all the data from kv2 to confirm the addition of the data.

   SELECT * FROM kv2;
   
    key | value
   -----+--------
      1 |    abc
      2 |    def
      3 | foobar
   
   (3 rows)

Run a Python script using dse spark-submit 

You run a Python script using the spark-submit command. For example, create the following file and save it as standalone.py:

#standalone.py

from pyspark import SparkContext, SparkConf

conf = SparkConf().setAppName("Stand Alone Python Script")
sc = SparkContext(conf=conf)
x = sc.cassandraTable("test","kv").collect()
print x

DataStax Enterprise sets the cassandra.connection.host environment variable, eliminating the need to set the variable in the Python file. Assuming you set up the kv table in the last example, execute standalone.py. On Linux, for example, from the installation directory, execute standalone.py as follows:

$ bin/dse spark-submit /<path>/standalone.py

Run Python jobs independently 

If you create test-suites with nose or other tools that run raw Python code, you cannot explicitly call dse spark-submit. You can execute Python scripts independent of DataStax Enterprise using one or the other of these methods.

• Set the PYTHONPATH environment variable to the path of the DataStax Enterprise integrated python executable, as shown in the next section.
• Use an initialization file that you modify to describe your environment and then run, as shown in the example later.

These methods work with the DataStax Enterprise integrated Spark, not the open source version.

Configure PYTHONPATH to run jobs 

This procedure uses the PYTHONPATH environment variable to find and use PySpark to execute a Python job.

Run a Python job using PYTHONPATH

1. Set the PYTHONPATH environment variable to the location of the Python executable in your DataStax Enterprise Spark environment.
   The default Python location depends on the type of installation:

   Installer-Services and Package installations	/usr/share/dse/spark/python
   Installer-No Services and Tarball installations	install_location/resources/spark/python

2. Run the Python script. For example, run the standalone.py file that you created in "Running a python script using dse spark-submit."

   $ python /<path>/standalone.py

   The output is:

   [Row(key=1, value=u'abc'), Row(key=2, value=u'def')]

Create an initialization file to run a Python job 

You can create an initialization file to run Python jobs independent of DataStax Enterprise.

Use the __name__.py convention to name the initialization file. For example, name the file __init__.py.

import os 
from os import getenv 
from os.path import join 
import sys
from subprocess import check_output 

HOME = getenv("HOME") 
DSE_HOME = getenv("DSE_HOME",join(HOME,"dse-4.7.0"))
SPARK_HOME = join(DSE_HOME,"resources","spark") 
os.environ['SPARK_HOME']=SPARK_HOME

PYSPARK_DIR = join(DSE_HOME,"resources","spark","python") 
ADD_PATH = [ PYSPARK_DIR ] 
for PATH in ADD_PATH: 
  if PATH not in sys.path: 
    sys.path.insert(1,PATH)

Use the initialization file to run jobs 

This procedure uses the initialization file shown in the last section to set up a number of environment variables and execute a sample Python job.

1. Create a directory and subdirectory for storing the python scripts. For example, create a directory named example that has a subdirectory named connector.
2. Create the example initialization file. For example, create the __init__.py file shown in the last section, and save it to example/connector directory.
3. Create scripts to configure Spark and query the kv table in Cassandra from Spark. For example, create the a script named connector.py and save it to the example/connector directory.

   #connector.py
   
   from pyspark import SparkContext, SparkConf
   
   def getSC():
       conf = SparkConf().setAppName("Stand Alone Python Script")
       sc = SparkContext(conf=conf)
       return sc

4. Create another script named moduleexample.py, for example, to print the results. Save moduleexample.py to the example directory.

   #!/usr/local/bin/python
   
   from connector.connector import getSC
   
   sc = getSC()
   print sc.cassandraTable("test","kv").collect()

   The example directory now contains these files:

   • moduleexample.py
   • connector/_init_.py
   • connector/connector.py
5. Execute the Python job as the same user that launched DataStax Enterprise.

   $ python example/moduleexample.py

   The output of the kv table appears.

   [Row(key=1, value=u'abc'), Row(key=2, value=u'def')]