/**
 * 
 */
package alokawi.poc.spark.core;

import java.util.HashMap;
import java.util.Map;

import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.sql.Column;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.SparkSession;

/**
 * @author alokkumar
 *
 */
public class SparkCassandraUtils {

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		SparkCassandraUtils cassandraUtils = new SparkCassandraUtils();
		cassandraUtils.execute();
	}

	@SuppressWarnings("deprecation")
	private void execute() {
		SparkConf conf = new SparkConf();
		conf.setAppName("cassandra-spark-poc");
		conf.setMaster("local[*]");

		SparkContext sparkContext = new SparkContext(conf);

		System.out.println(sparkContext);

		SparkSession sparkSession = SparkSession.builder().appName("cassandra-spark-poc").master("local[*]")
				.getOrCreate();

		SQLContext sqlContext = new SQLContext(sparkSession);

		Map<String, String> options = new HashMap<String, String>();
		options.put("keyspace", "wootag");
		options.put("table", "video_view");

		Dataset<Row> dataset = sqlContext.read().format("org.apache.spark.sql.cassandra").options(options).load()
				.cache();

		dataset.registerTempTable("temptable");
		dataset.filter(new Column("view_duration_in_second").equalTo("15")).show();

		String query = "select * from temptable where video_id = 'v-44'";

		sqlContext.sql(query).show();

		System.out.println(sqlContext);

	}

}

@@ -81,19 +81,80 @@
 va.util.
-Map
+List;%0Aimport java.util.Map;%0Aimport java.util.concurrent.TimeUnit
 ;%0A%0Aimpor
@@ -1394,158 +1394,294 @@
 %22);%0A
+%0A
 %09%09
-dataset.filter(new Column(%22view_duration_in_second%22).equalTo(%2215%22)).show();%0A%0A%09%09String query = %22select * from temptable where video_id = 'v-44'%22;%0A%0A%09%09
+String query = %22select video_id, view_duration_in_second, count(*) from temptable group by 1, 2%22;%0A%0A%09%09List%3CRow%3E collectAsList = sqlContext.sql(query).collectAsList();%0A%09%09for (Row row : collectAsList) %7B%0A%09%09%09System.out.println(row.get(0) + %22,%22 + row.get(1) + %22,%22 + row.get(2));%0A%09%09%7D%0A%0A%09%09// 
 sqlC
@@ -1707,46 +1707,604 @@
 how(
+1000
 );%0A%0A%09%09
-System.out.println(sqlContext);%0A
+long startTime = 1485907200000L;%0A%09%09long endTime = 1487226374000L;%0A%0A%09%09for (long i = startTime; i %3C= endTime; i = i + TimeUnit.DAYS.toMillis(1)) %7B%0A%0A%09%09%09dataset.filter(new Column(%22event_start_timestamp%22).geq(i))%0A%09%09%09%09%09.filter(new Column(%22event_start_timestamp%22).leq(i + TimeUnit.DAYS.toMillis(1)))%0A%09%09%09%09%09.groupBy(new Column(%22view_duration_in_second%22), new Column(%22video_id%22)).count()%0A%09%09%09%09%09.orderBy(%22view_duration_in_second%22).show(1000);%0A%09%09%09sleepDelay();%0A%09%09%09%0A%09%09%7D%0A%0A%09%7D%0A%0A%09private void sleepDelay() %7B%0A%09%09try %7B%0A%09%09%09Thread.sleep(3000L);%0A%09%09%7D catch (InterruptedException e) %7B%0A%09%09%09e.printStackTrace();%0A%09%09%7D
 %0A%09%7D%0A

Add loop to calculate per day drop-off