package com.mdo;

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

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;

import com.google.common.collect.ObjectArrays;

public class PocAppNew implements Serializable {

	private static final long serialVersionUID = -5666546913223410974L;

	static ArrayList<CFMeta> CFMetaList = new ArrayList<CFMeta>();
	
	public static final CFMeta DEVICE_PERF_BY_MONTH = new CFMeta(Constants.DEVICE_PERF_BY_CF + Constants.AGGREGATION_MONTH, new String[]{"key", Constants.AGGREGATION_MONTH,"make", "model"}, new String[]{"sesslength","sesscntr"});

    static {
    	DEVICE_PERF_BY_MONTH.setStmtKeySuffix("_SESSEND");
    	CFMetaList.add(DEVICE_PERF_BY_MONTH);
    }

    
    public PocAppNew() {
    	
    	System.setProperty("hadoop.home.dir", "F:\\winutil\\");
        // just an initialisation of Spark Context
    	SparkConf sconf= new SparkConf(true)
        .set("spark.cassandra.connection.host", "localhost")
        .set("spark.cassandra.auth.username", "admin")            
        .set("spark.cassandra.auth.password", "*****") 
        .set("spark.cleaner.ttl", "3600")
        .setMaster("local[4]")
        .setAppName("JavaApiDemo");

    	JavaSparkContext sc = new JavaSparkContext(sconf);
    	final String schema = "maa";
    	
    	//1. Get raw sessions data
        JavaRDD<Session> sessions = javaFunctions(sc).cassandraTable(schema, Session.cf, Session.class, Session.columnNameMap)
        		.map(new Function<Session, Session>() {
		        	@Override
		        	public Session call(Session session) throws Exception {
		        		session.generateRowKeys();
		        		return  session;
		        	}
        });
        System.out.println("Session rows returned: " + sessions.count());
        
        //2. Process the raw sessions data
        JavaRDD<DevicePerf> devPerfByMnthRows = sessions.flatMap(new FlatMapFunction<Session, DevicePerf>() {
			@Override
			public Iterable<DevicePerf> call(Session session) throws Exception {
				List<DevicePerf> devicePerfs = new ArrayList<DevicePerf>();
				long endTime = session.getSessend();
                long length = endTime - session.getSesstart();
                
                Set<String> rowkeys= session.getRowkeys();
                DateParser dateParser = new DateParser(session.getSesstart());//subrowkeyVal

                for (String rowkey : rowkeys) {
                	DevicePerf devicePerf= new DevicePerf(Constants.AGGREGATION_MONTH, new String[]{"key", Constants.AGGREGATION_MONTH, "make", "model", "sesslength", "sesscntr"});
                	devicePerf.setKey(rowkey);
                	devicePerf.setMonth(Long.parseLong(dateParser.getMonthColumn()));//subrowkeyVal
                	devicePerf.setMake(session.getMake());//subrowkeyVal
                	devicePerf.setModel(session.getModel());//subrowkeyVal
                	
                	devicePerf.setSesslength(length);//colKeyVal
                	devicePerf.setSesscntr(1l);//colKeyVal
        			
                	devicePerfs.add(devicePerf);
				}
                return devicePerfs;
			}
        });
        System.out.println("DevicePerf rows returned: " + devPerfByMnthRows.count());
        
        //3. Iterate over the processed data(derive from #2) and fetch the previously aggregated data from store for those rowkeys 
        //   Add the values from this batch to previous batch values
        VoidFunction<DevicePerf> aggregateDevicePerf = new VoidFunction<DevicePerf>() {
			@Override
			public void call(DevicePerf devicePerf) throws Exception {
		        JavaRDD<DevicePerf> devicePerfs = javaFunctions(sc).cassandraTable(schema, devicePerf.getCf(), DevicePerf.class)
		        		.select(devicePerf.getSelectCols()).where("key=? AND hour=? AND make=? AND model=? ", devicePerf.getKey(), devicePerf.getKey(), devicePerf.getKey(),devicePerf.getKey());
		        DevicePerf firstRow = devicePerfs.first();
		        devicePerf.setSesscntr(firstRow.getSesscntr() + devicePerf.getSesscntr());
		        devicePerf.setSesslength(firstRow.getSesslength() + devicePerf.getSesslength());
			}
		};
		devPerfByMnthRows.foreach(aggregateDevicePerf);
        
		javaFunctions(devPerfByMnthRows, DevicePerf.class).saveToCassandra(schema, DEVICE_PERF_BY_MONTH.getColFamily(), 
				ObjectArrays.concat(DEVICE_PERF_BY_MONTH.getPrimaryKeys(), DEVICE_PERF_BY_MONTH.getColumnIds(), String.class));
        
        sc.stop();
    }
    
    public static void main(String[] args) {
        new PocAppNew();
    }

}