Scylla and Spark integration¶

Prerequisites¶

• Scylla

• sbt

Prepare Scylla¶

Firstly, we need to create keyspace and tables in which data processed by the example application will be stored.

Launch Scylla and connect to it using cqlsh. The following commands will create a new keyspace for our tests and make it the current one.

CREATE KEYSPACE spark_example WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor': 1};
USE spark_example;

Then, tables both for input and output data need to be created:

CREATE TABLE persons (name TEXT PRIMARY KEY, age INT);
CREATE TABLE adults (name TEXT PRIMARY KEY);

Lastly, the database needs to contain some input data for our application to process:

INSERT INTO persons (name, age) VALUES ('Anne', 34);
INSERT INTO persons (name, age) VALUES ('John', 47);
INSERT INTO persons (name, age) VALUES ('Elisabeth', 89);
INSERT INTO persons (name, age) VALUES ('George', 52);
INSERT INTO persons (name, age) VALUES ('Amy', 17);
INSERT INTO persons (name, age) VALUES ('Jack', 16);
INSERT INTO persons (name, age) VALUES ('Treebeard', 36421);

Prepare the application¶

With a database containing all the necessary tables and data, it is now time to write our example application. Create a directory scylla-spark-example, which will contain all source code and build configuration.

First, very important file is build.sbt, which should be created in the project main directory. It contains all the application metadata, including name, version, and dependencies.

name := "scylla-spark-example-simple"
version := "1.0"
scalaVersion := "2.10.5"

libraryDependencies ++= Seq(
        "com.datastax.spark" %% "spark-cassandra-connector" % "1.5.0-M1",
        "org.apache.spark" %% "spark-catalyst" % "1.5.0" % "provided"
    )

Then, we need to enable sbt-assembly plugin. Create directory project and create file plugins.sbt with the following content:

addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.0")

The steps above should cover all build configuration, what is left is the actual logic of the application. Create file src/main/scala/ScyllaSparkExampleSimple.scala:

import org.apache.spark.{SparkContext,SparkConf}
import com.datastax.spark.connector._

object ScyllaSparkExampleSimple {
    def main(args: Array[String]): Unit = {
        val sc = new SparkContext(new SparkConf())

        val persons = sc.cassandraTable("spark_example", "persons")

        val adults = persons.filter(_.getInt("age") >= 18).map(n => Tuple1(n.getString("name")))
        adults.saveToCassandra("spark_example", "adults")

        val out = s"Adults: %d\nTotal: %d\n".format(adults.count(), persons.count())
        println(out)
    }
}

Since we don’t want to hardcode in our application any information about Scylla or Spark we will also need an additional configuration file spark-scylla.conf.

spark.master local
spark.cassandra.connection.host 127.0.0.1

Now it is time to build the application and create a self-containing jar file that we will be able to send to Spark. To do that, execute the command:

sbt assembly

It will download all necessary dependencies, build our example, and create an output jar file in target/scala-2.10/scylla-spark-example-simple-assembly-1.0.jar.

Download and run Spark¶

The next step is to get Spark running. Pre-built binaries can be downloaded from this website. Make sure to choose release 1.5.0. Since we are going to use it with Scylla Hadoop version doesn’t matter.

Once the download has finished, unpack the archive and in its root directory, execute the following command to start Spark Master:

./sbin/start-master.sh -h localhost

Spark Web UI should now be available at http://localhost:8080. The Spark URL used to connect its workers is spark://localhost:7077.

With the master running, the only thing left to have minimal Spark deployment is to start a worker. This can be done with the following command:

./sbin/start-slave.sh spark://localhost:7077

Run application¶

The application is built, Spark is up, and Scylla has all the necessary tables created and contains the input data for our example. This means that we are ready to run the application. Make sure that Scylla is running and execute (still in the Spark directory) the following command):

./bin/spark-submit --properties-file /path/to/scylla-spark-example/spark-scylla.conf \
    --class ScyllaSparkExampleSimple /path/to/scylla-spark-example/target/scala-2.10/scylla-spark-example-simple-assembly-1.0.jar

spark-submit will output some logs and debug information, but among them, there should be a message from the application:

Adults: 5
Total: 7

You can also connect to Scylla with cqlsh, and using the following query, see the results of our example in the database.

SELECT * FROM spark_example.adults;

Expected output:

 name
-----------
 Treebeard
 Elisabeth
    George
      John
      Anne

Based on http://www.planetcassandra.org/getting-started-with-apache-spark-and-cassandra/ and http://koeninger.github.io/spark-cassandra-example/#1.

Prerequisites¶

• Scylla

• Maven

• Git

Get the source code¶

You can get the source code of this example by cloning the following repository:

https://github.com/jsebrien/spark-tests

Disable Apache Cassandra¶

spark-tests are configured to launch Cassandra, which is not what we want to achieve here. The following patch disables Cassandra. It can be applied, for example, using git apply --ignore-whitespace -.

diff --git a/src/main/java/blog/hashmade/spark/util/CassandraUtil.java b/src/main/java/blog/hashmade/spark/util/CassandraUtil.java
index 37bbc2e..bfe5517 100644
--- a/src/main/java/blog/hashmade/spark/util/CassandraUtil.java
+++ b/src/main/java/blog/hashmade/spark/util/CassandraUtil.java
@@ -14,7 +14,7 @@ public final class CassandraUtil {
    }

    static Session startCassandra() throws Exception {
-       EmbeddedCassandraServerHelper.startEmbeddedCassandra();
+       //EmbeddedCassandraServerHelper.startEmbeddedCassandra();
        Thread.sleep(EMBEDDED_CASSANDRA_SERVER_WAITING_TIME);
        Cluster cluster = new Cluster.Builder().addContactPoints("localhost")
                .withPort(9142).build();

Update connector¶

spark-tests use Spark Cassandra Connector in version 1.1.0 which is too old for our purposes. Before 1.3.0 the connector used to use Thrift as well CQL and that won’t work with Scylla. Updating the example isn’t very complicated and can be accomplished by applying the following patch:

diff --git a/pom.xml b/pom.xml
index 673e22b..1245ffc 100644
--- a/pom.xml
+++ b/pom.xml
@@ -142,7 +142,7 @@
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.10</artifactId>
-           <version>1.1.0</version>
+           <version>1.3.0</version>
            <exclusions>
                <exclusion>
                    <groupId>com.google.guava</groupId>
@@ -157,7 +157,7 @@
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.10</artifactId>
-           <version>1.1.0</version>
+           <version>1.3.0</version>
        </dependency>
        <dependency>
            <groupId>org.cassandraunit</groupId>
@@ -173,18 +173,18 @@
        <dependency>
            <groupId>com.datastax.cassandra</groupId>
            <artifactId>cassandra-driver-core</artifactId>
-           <version>2.1.2</version>
+           <version>2.1.7.1</version>
        </dependency>
        <!-- Datastax -->
        <dependency>
            <groupId>com.datastax.spark</groupId>
            <artifactId>spark-cassandra-connector_2.10</artifactId>
-           <version>1.1.0-beta2</version>
+           <version>1.3.0</version>
        </dependency>
        <dependency>
            <groupId>com.datastax.spark</groupId>
            <artifactId>spark-cassandra-connector-java_2.10</artifactId>
-           <version>1.1.0-beta2</version>
+           <version>1.3.0</version>
        </dependency>
        <dependency>
            <groupId>net.sf.supercsv</groupId>
diff --git a/src/main/java/blog/hashmade/spark/DatastaxSparkTest.java b/src/main/java/blog/hashmade/spark/DatastaxSparkTest.java
index 1027e42..190eb3d 100644
--- a/src/main/java/blog/hashmade/spark/DatastaxSparkTest.java
+++ b/src/main/java/blog/hashmade/spark/DatastaxSparkTest.java
@@ -43,8 +43,7 @@ public class DatastaxSparkTest {
                .setAppName("DatastaxTests")
                .set("spark.executor.memory", "1g")
                .set("spark.cassandra.connection.host", "localhost")
-               .set("spark.cassandra.connection.native.port", "9142")
-               .set("spark.cassandra.connection.rpc.port", "9171");
+               .set("spark.cassandra.connection.port", "9142");
        SparkContext ctx = new SparkContext(conf);
        SparkContextJavaFunctions functions = CassandraJavaUtil.javaFunctions(ctx);
        CassandraJavaRDD<CassandraRow> rdd = functions.cassandraTable("roadtrips", "roadtrip");

Build the example¶

The example can be built with Maven:

mvn compile

Start Scylla¶

The application we are trying to run will try to connect with Scylla using custom port 9142. That’s why when starting Scylla, an additional flag is needed to make sure that’s the port it listens on (alternatively, you can change all occurrences of 9142 to 9042 in the example source code).

scylla --native-transport-port=9142

Run the application¶

With the example compiled and Scylla running all that is left to be done is to actually run the application:

mvn exec:java