Creating an RDD after retrieving data from cassandra DB

Question

I'm using cassandra and spark for my project, for now i wrote this to retrieve data from the DB:

 results = session.execute("SELECT * FROM foo.test");

 ArrayList<String> supportList = new ArrayList<String>();
 for (Row row : results) {
            supportList.add(row.getString("firstColumn") + "," + row.getString("secondColumn")));
        }
        JavaRDD<String> input = sparkContext.parallelize(supportList);
        JavaPairRDD<String, Double> tuple = input.mapToPair(new PairFunction<String, String, Double>() {
            public Tuple2<String, Double> call(String x) {
                String[] parts = x.split(",");
                return new Tuple2(parts[0],String.valueOf(new Random().nextInt(30) + 1));
            }

It works, but i want to know if there is a pretty way to write the above code, what i want to achieve is:

in scala i can retrieve and fill a RDD simply in this way :

val dataRDD = sc.cassandraTable[TableColumnNames]("keySpace", "table")
how i can write the same thing in Java, without using support list or other "nasty" things.

UPDATE

JavaRDD<String> cassandraRowsRDD = javaFunctions(javaSparkContext).cassandraTable("keyspace", "table")
                .map(new Function<CassandraRow, String>() {
                    @Override
                    public String call(CassandraRow cassandraRow) throws Exception {
                        return cassandraRow.toString();
                    }
                });

i'm getting on this row -> public String call(CassandraRow cassandraRow) this exception:

Exception in thread "main" org.apache.spark.SparkException: Task not serializable
    at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:166)
    at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:158)
    at org.apache.spark.SparkContext.clean(SparkContext.scala:1623)
    at org.apache.spark.rdd.RDD.map(RDD.scala:286)
    at org.apache.spark.api.java.JavaRDDLike$class.map(JavaRDDLike.scala:89)
    at org.apache.spark.api.java.AbstractJavaRDDLike.map(JavaRDDLike.scala:46)
    at org.sparkexamples.cassandraExample.main.KMeans.executeQuery(KMeans.java:271)
    at org.sparkexamples.cassandraExample.main.KMeans.main(KMeans.java:67)
Caused by: java.io.NotSerializableException: org.sparkexamples.cassandraExample.main.KMeans
Serialization stack:
    - object not serializable (class: org.sparkexamples.cassandraExample.main.KMeans, value: org.sparkexamples.cassandraExample.main.KMeans@3015db78)
    - field (class: org.sparkexamples.cassandraExample.main.KMeans$2, name: this$0, type: class org.sparkexamples.cassandraExample.main.KMeans)
    - object (class org.sparkexamples.cassandraExample.main.KMeans$2, org.sparkexamples.cassandraExample.main.KMeans$2@5dbf5634)
    - field (class: org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1, name: fun$1, type: interface org.apache.spark.api.java.function.Function)
    - object (class org.apache.spark.api.java.JavaPairRDD$$anonfun$toScalaFunction$1, <function1>)
    at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:38)
    at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:47)
    at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:80)
    at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:164)
    ... 7 more

thanks in advance.

Why don't you use the Java API if you want to do exactly like in your scala example? http://www.datastax.com/dev/blog/accessing-cassandra-from-spark-in-java — ccheneson, Jul 30 '15 at 08:00
@ccheneson i'm seeing these api. Can you see question updates, i'm getting an error. — OiRc, Jul 30 '15 at 08:07
I don't see any question updates. Please, copy/paste the error you get in your post — ccheneson, Jul 30 '15 at 08:09

score 4 · Accepted Answer · edited May 23 '17 at 12:14

4

Have a look at the answer: RDD not serializable Cassandra/Spark connector java API

The problem may be that the class surrounding the code block you've shown is not Serializable.

edited May 23 '17 at 12:14

Community

1
1

answered Jul 30 '15 at 08:50

mattinbits

10,370
1
26
35

In the link I posted, the class implements `Serializable` indeed. So that could be it `public class JavaDemo implements Serializable {` – ccheneson Jul 30 '15 at 08:56

score 0 · Answer 2 · answered Aug 01 '15 at 08:13

I had the same problem. I implemented the spark interface function in a separate class and supplied that to the map functionality. It worked post that.

Sample

public a implements Function {....}

Used this in the map

.....map(new a())

It got rectified. Some problem with spark deserialisation with respect to anonymous class.

Creating an RDD after retrieving data from cassandra DB

2 Answers2