/**
 * 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(KeyValue.newInstance("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 static KeyValue newInstance(String k, Integer v) {
    KeyValue kv = new KeyValue();
    kv.setKey(k);
    kv.setValue(v);
    return kv;
  }

  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;
  }
}


@@ -2006,19 +2006,32 @@
 v%22);%0A%09%7D%0A
-%7D%0A%0A
+  public static 
 class Ke
@@ -2065,16 +2065,18 @@
 ble %7B%0A  
+ 
+ 
 private 
@@ -2087,16 +2087,18 @@
 ng key;%0A
+  
   privat
@@ -2113,24 +2113,26 @@
 r value;%0A%0A  
+ 
+ 
 public stati
@@ -2181,16 +2181,18 @@
 ) %7B%0A    
+  
 KeyValue
@@ -2213,24 +2213,26 @@
 alue();%0A    
+ 
+ 
 kv.setKey(k)
@@ -2233,16 +2233,18 @@
 Key(k);%0A
+  
     kv.s
@@ -2255,24 +2255,26 @@
 lue(v);%0A    
+ 
+ 
 return kv;%0A 
@@ -2272,23 +2272,27 @@
 urn kv;%0A
+  
   %7D%0A%0A  
+ 
+ 
 public I
@@ -2311,24 +2311,26 @@
 lue() %7B%0A    
+  
 return value
@@ -2329,27 +2329,31 @@
 rn value;%0A  
+ 
+ 
 %7D%0A%0A
+  
   public Str
@@ -2367,24 +2367,26 @@
 Key() %7B%0A    
+  
 return key;%0A
@@ -2385,23 +2385,27 @@
 rn key;%0A
+  
   %7D%0A%0A  
+ 
+ 
 public v
@@ -2430,24 +2430,26 @@
 er v) %7B%0A    
+  
 this.value =
@@ -2454,18 +2454,22 @@
  = v;%0A  
+ 
+ 
 %7D%0A
+  
   public
@@ -2493,24 +2493,26 @@
 ng k) %7B%0A    
+ 
+ 
 this.key = k
@@ -2519,9 +2519,15 @@
 ;%0A  
-%7D%0A
+  %7D%0A  
 %7D%0A%0A
+%7D%0A

make inner class instead