/**
 * Illustrates loading a json file and finding out if people like pandas
 */
package com.oreilly.learningsparkexamples.java;
import java.io.Serializable;

import java.io.StringReader;
import java.util.ArrayList;
import java.util.List;
import java.util.Iterator;
import java.lang.Iterable;
import scala.Tuple2;

import org.apache.commons.lang.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.DoubleFunction;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;

import com.datastax.spark.connector.CassandraRow;
import static com.datastax.spark.connector.CassandraJavaUtil.javaFunctions;

public class BasicQueryCassandra {
  public static void main(String[] args) throws Exception {
		if (args.length != 2) {
      throw new Exception("Usage BasicLoadJson [sparkMaster] [cassandraHost]");
		}
    String sparkMaster = args[0];
    String cassandraHost = args[1];
    SparkConf conf = new SparkConf(true)
      .set("spark.cassandra.connection.host", cassandraHost);

		JavaSparkContext sc = new JavaSparkContext(
      sparkMaster, "basicquerycassandra", conf);
    // entire table as an RDD
    // assumes your table test was created as CREATE TABLE test.kv(key text PRIMARY KEY, value int);
    JavaRDD<CassandraRow> data = javaFunctions(sc).cassandraTable("test" , "kv");
    // print some basic stats
    System.out.println(data.mapToDouble(new DoubleFunction<CassandraRow>() {
        public double call(CassandraRow row) {
          return row.getInt("value");
        }}).stats());
    // write some basic data to Cassandra
    ArrayList<KeyValue> input = new ArrayList<KeyValue>();
    input.add(new KeyValue("mostmagic", 3));
    JavaRDD<KeyValue> kvRDD = sc.parallelize(input);
    javaFunctions(kvRDD, KeyValue.class).saveToCassandra("test", "kv");
	}
}

class KeyValue implements Serializable {
  private String key;
  private Integer value;

  public KeyValue() {
  }

  public KeyValue(String key_, Integer value_) {
    key = key_;
    value = value_;
  }

  public Integer getValue() {
    return value;
  }

  public String getKey() {
    return key;
  }

  public void setValue(Integer v) {
    this.value = v;
  }
  public void setKey(String k) {
    this.key = k;
  }
}


@@ -1843,20 +1843,16 @@
 add(
-new 
 KeyValue
 (%22mo
@@ -1847,16 +1847,28 @@
 KeyValue
+.newInstance
 (%22mostma
@@ -2134,24 +2134,31 @@
  public 
+static 
 KeyValue
 (String 
@@ -2149,16 +2149,28 @@
 KeyValue
+ newInstance
 (String 
@@ -2170,19 +2170,16 @@
 String k
-ey_
 , Intege
@@ -2185,21 +2185,16 @@
 er v
-alue_
 ) %7B%0A    
 key 
@@ -2193,38 +2193,89 @@
     
-key = key_;%0A    value = value_
+KeyValue kv = new KeyValue();%0A    kv.setKey(k);%0A    kv.setValue(v);%0A    return kv
 ;%0A  

change the constructor for KeyValue in ajva