adding session template

client/src/main/java/zingg/client/IZingg.java

@@ -1,7 +1,7 @@
 package zingg.client;
 import org.apache.spark.sql.SparkSession;
 
-public interface IZingg<D,R,C> {
+public interface IZingg<S,D,R,C,T> {
 
 	public void init(Arguments args, String license)
 			throws ZinggClientException;
@@ -30,7 +30,7 @@ public void init(Arguments args, String license)
 
     public Long getUnsureMarkedRecordsStat(ZFrame<D,R,C>  markedRecords);
 
-    public void setSpark(SparkSession session);
+    public void setSession(S session); // method name will have to be changed in Client too
 
 	public void setClientOptions(ClientOptions clientOptions);
 

client/src/main/java/zingg/client/ZFrame.java

@@ -1,67 +1,68 @@
 package zingg.client;
 
 import java.util.List;
+
 //Dataset, Row, column
-public interface ZFrame<T, R, C> {
+public interface ZFrame<D, R, C> {
 
-    public ZFrame<T, R, C> cache();
-    public ZFrame<T, R, C> as(String s);
+    public ZFrame<D, R, C> cache();
+    public ZFrame<D, R, C> as(String s);
     public String[] columns();
-    public ZFrame<T, R, C> select(C... cols);
-    public ZFrame<T, R, C> select(List<C> cols);
-    public ZFrame<T, R, C> select(String col, String... cols);
-    public ZFrame<T, R, C> select(String col);
-    public ZFrame<T, R, C> selectExpr(String... col);
-    public ZFrame <T, R, C> distinct();
+    public ZFrame<D, R, C> select(C... cols);
+    public ZFrame<D, R, C> select(List<C> cols);
+    public ZFrame<D, R, C> select(String col, String... cols);
+    public ZFrame<D, R, C> select(String col);
+    public ZFrame<D, R, C> selectExpr(String... col);
+    public ZFrame <D, R, C> distinct();
     public List<R> collectAsList();
 
-    public ZFrame<T,R,C> toDF(String[] cols);
-    public ZFrame<T,R,C> toDF(String col1, String col2);
+    public ZFrame<D, R, C> toDF(String[] cols);
+    public ZFrame<D, R, C> toDF(String col1, String col2);
 
-    public ZFrame<T,R,C> join(ZFrame<T,R,C> lines1, String joinColumn);
+    public ZFrame<D, R, C> join(ZFrame<D, R, C> lines1, String joinColumn);
 
-    public ZFrame<T,R,C> joinRight(ZFrame<T,R,C> lines1, String joinColumn);
+    public ZFrame<D, R, C> joinRight(ZFrame<D, R, C> lines1, String joinColumn);
 
-    public ZFrame<T,R,C> join(ZFrame<T,R,C> lines1, String joinColumn, boolean addPrefixToCol, String jointype);    
+    public ZFrame<D, R, C> join(ZFrame<D, R, C> lines1, String joinColumn, boolean addPrefixToCol, String jointype);    
 
     public C col(String colname);
 
     public long count();
 
-    public ZFrame<T,R,C> filter(C col);
+    public ZFrame<D, R, C> filter(C col);
 
-    public T df();
+    public D df();
 
-    public ZFrame<T,R,C> withColumnRenamed(String s, String t);
+    public ZFrame<D, R, C> withColumnRenamed(String s, String t);
 
-    public ZFrame<T,R,C> dropDuplicates(String c, String... d);
+    public ZFrame<D, R, C> dropDuplicates(String c, String... d);
 
-    public ZFrame<T,R,C> dropDuplicates(String[] c);
+    public ZFrame<D, R, C> dropDuplicates(String[] c);
 
-    public ZFrame<T,R,C> drop(String c);
-    public ZFrame<T,R,C> drop(String... c);
-    public ZFrame<T,R,C> except(ZFrame<T,R,C> c);
+    public ZFrame<D, R, C> drop(String c);
+    public ZFrame<D, R, C> drop(String... c);
+    public ZFrame<D, R, C> except(ZFrame<D, R, C> c);
 
-    public ZFrame<T,R,C> groupByMinMax(C c);
+    public ZFrame<D, R, C> groupByMinMax(C c);
 
-    public ZFrame<T,R,C> union(ZFrame<T,R,C> other);
+    public ZFrame<D, R, C> union(ZFrame<D, R, C> other);
 
-    public ZFrame<T,R,C> unionByName(ZFrame<T,R,C> other, boolean flag);
+    public ZFrame<D, R, C> unionByName(ZFrame<D, R, C> other, boolean flag);
 
-    public ZFrame<T,R,C> withColumn(String s, int c);
-    public ZFrame<T,R,C> withColumn(String s, String c);
-    public ZFrame<T,R,C> withColumn(String s, double c);
-    public ZFrame<T,R,C> withColumn(String s, C c);
+    public ZFrame<D, R, C> withColumn(String s, int c);
+    public ZFrame<D, R, C> withColumn(String s, String c);
+    public ZFrame<D, R, C> withColumn(String s, double c);
+    public ZFrame<D, R, C> withColumn(String s, C c);
 
 
-    public ZFrame<T,R,C> repartition(int num);
-    public ZFrame<T,R,C> repartition(int num, C c);
+    public ZFrame<D, R, C> repartition(int num);
+    public ZFrame<D, R, C> repartition(int num, C c);
 
-    public ZFrame<T,R,C> sample(boolean repartition, float num);
-    public ZFrame<T,R,C> sample(boolean repartition, double num);
+    public ZFrame<D, R, C> sample(boolean repartition, float num);
+    public ZFrame<D, R, C> sample(boolean repartition, double num);
 
 
-    public ZFrame<T,R,C> coalesce(int num);
+    public ZFrame<D, R, C> coalesce(int num);
 
     public C gt(String c);
 
@@ -80,12 +81,13 @@ public interface ZFrame<T, R, C> {
     public void show();
 
     public String showSchema();
-    public ZFrame<T,R,C> orderBy(String c);
-    public ZFrame<T,R,C> sortAscending(String c);
-    public ZFrame<T,R,C> sortDescending(String c);
+
+    public ZFrame<D, R, C> orderBy(String c);
+    public ZFrame<D, R, C> sortAscending(String c);
+    public ZFrame<D, R, C> sortDescending(String c);
 
 
-    public ZFrame<T,R,C> limit(int l);
+    public ZFrame<D, R, C> limit(int l);
 
     public String getAsString(R r, String colName);
 

core/src/main/java/zingg/ZinggBase.java

@@ -40,7 +40,7 @@
 //Dataset
 //row
 //column
-public abstract class ZinggBase<S,D, R, C, T> implements Serializable, IZingg<D, R, C> {
+public abstract class ZinggBase<S,D, R, C, T> implements Serializable, IZingg<S, D, R, C, T> {
 
     protected Arguments args;
 

core/src/main/java/zingg/documenter/DataColDocumenter.java

@@ -7,17 +7,18 @@
 import org.apache.spark.sql.SparkSession;
 
 import zingg.client.Arguments;
+import zingg.client.ZFrame;
 import zingg.client.ZinggClientException;
 
-public class DataColDocumenter extends DocumenterBase {
+public class DataColDocumenter<S,D,R,C,T> extends DocumenterBase<S,D,R,C,T> {
 	protected static String name = "zingg.DataColDocumenter";
 	public static final Log LOG = LogFactory.getLog(DataColDocumenter.class);
 
-	public DataColDocumenter(SparkSession spark, Arguments args) {
-		super(spark, args);
+	public DataColDocumenter(S session, Arguments args) {
+		super(session, args);
 	}
 
-	public void process(Dataset<Row> data) throws ZinggClientException {
+	public void process(ZFrame<D, R, C> data) throws ZinggClientException {
 	}
 
 }

core/src/main/java/zingg/documenter/DataDocumenter.java

@@ -15,6 +15,7 @@
 import zingg.client.Arguments;
 import zingg.client.ZFrame;
 import zingg.client.ZinggClientException;
+import zingg.util.PipeUtilBase;
 
 public class DataDocumenter<S,D,R,C,T> extends DocumenterBase<S,D,R,C,T> {
 	protected static String name = "zingg.DataDocumenter";
@@ -24,17 +25,18 @@ public class DataDocumenter<S,D,R,C,T> extends DocumenterBase<S,D,R,C,T> {
 	public static final Log LOG = LogFactory.getLog(DataDocumenter.class);
 	protected  ZFrame<D,R,C>  data;
 
-	public DataDocumenter(Arguments args) {
-		super(args);
+	public DataDocumenter(S session,Arguments args) {
+		super(session, args);
 		data = getDFUtil().emptyDataFrame();
 	}
 
+
 	public void process() throws ZinggClientException {
 		try {
 			LOG.info("Data document generation starts");
 
 			try {
-				data = PipeUtil.read(spark, false, false, args.getData());
+				data = getPipeUtil().read(false,false, args.getData());
 				LOG.info("Read input data : " + data.count());
 			} catch (ZinggClientException e) {
 				LOG.warn("No data has been found");
@@ -54,11 +56,11 @@ public void process() throws ZinggClientException {
 	protected void createDataDocument() throws ZinggClientException {
 		if (!data.isEmpty()) {
 			Map<String, Object> root = populateTemplateData();
-			writeMoelDocument(root);
+			writeModelDocument(root);
 		}
 	}
 
-	protected void writeMoelDocument(Map<String, Object> root) throws ZinggClientException {
+	protected void writeModelDocument(Map<String, Object> root) throws ZinggClientException {
 		writeDocument(DATA_DOC_TEMPLATE, root, args.getZinggDataDocFile());
 	}
 

core/src/main/java/zingg/documenter/DocumenterBase.java

@@ -5,7 +5,7 @@
 import java.io.Writer;
 import java.util.Map;
 
-import org.apache.spark.sql.SparkSession;
+// import org.apache.spark.sql.SparkSession;
 
 import freemarker.template.Configuration;
 import freemarker.template.Template;
@@ -16,17 +16,23 @@
 import zingg.client.util.ColName;
 import zingg.util.RowWrapper;
 
-class DocumenterBase<S,D,R,C,T> extends ZinggBase<S,D,R,C,T>{
+abstract class DocumenterBase<S,D,R,C,T> extends ZinggBase<S,D,R,C,T>{
 	protected static Configuration config;
-	protected SparkSession spark;
+	protected S session;
 	protected Arguments args;
 
-	public DocumenterBase(SparkSession spark, Arguments args) {
-		this.spark = spark;
+	public DocumenterBase(S session, Arguments args) {
+		this.session = session;
 		this.args = args;
 		config = createConfigurationObject();
 	}
 
+	public void setSession(S session){
+
+	};
+
+	public abstract void cleanup() throws ZinggClientException;
+
 	public Configuration getTemplateConfig() {
 		if (config == null) {
 			config = createConfigurationObject();

core/src/main/java/zingg/documenter/ModelColDocumenter.java

@@ -21,8 +21,8 @@ public class ModelColDocumenter<S,D,R,C,T> extends DocumenterBase<S,D,R,C,T> {
 	private final String COLUMN_DOC_TEMPLATE = "columnDocTemplate.ftlh";
 	private final String Z_COLUMN_TEMPLATE = "zColumnTemplate.ftlh";
 
-	public ModelColDocumenter(Arguments args) {
-		super(args);
+	public ModelColDocumenter(S session, Arguments args) {
+		super(session, args);
 	}
 
 	public void process( ZFrame<D,R,C>  data) throws ZinggClientException {

core/src/main/java/zingg/documenter/ModelDocumenter.java

@@ -6,6 +6,7 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.spark.sql.Dataset;
@@ -17,6 +18,7 @@
 import zingg.client.ZinggClientException;
 import zingg.client.util.ColName;
 
+
 public class ModelDocumenter<S,D,R,C,T> extends DocumenterBase<S,D,R,C,T> {
 
 	protected static String name = "zingg.ModelDocumenter";
@@ -26,8 +28,8 @@ public class ModelDocumenter<S,D,R,C,T> extends DocumenterBase<S,D,R,C,T> {
 	ModelColDocumenter<S,D,R,C,T> modelColDoc;
 	protected  ZFrame<D,R,C>  markedRecords;
 
-	public ModelDocumenter(Arguments args) {
-		super(args);
+	public ModelDocumenter(S session, Arguments args) {
+		super(session, args);
 		markedRecords = getDFUtil().emptyDataFrame();
 	}
 
@@ -41,7 +43,7 @@ protected void createModelDocument() throws ZinggClientException {
 			LOG.info("Model document generation starts");
 
 			try {
-				markedRecords = PipeUtil.read(spark, false, false, PipeUtil.getTrainingDataMarkedPipe(args));
+				markedRecords = getPipeUtil().read(false, false, getPipeUtil().getTrainingDataMarkedPipe(args));
 			} catch (ZinggClientException e) {
 				LOG.warn("No marked record has been found");
 			}

core/src/main/java/zingg/util/PipeUtilBase.java

@@ -20,6 +20,7 @@ public  ZFrame<D, R, C> readInternal(boolean addLineNo,
 	public  ZFrame<D, R, C> read(boolean addLineNo, int numPartitions,
 			boolean addSource, Pipe... pipes);
 
+
 	public void write(ZFrame<D, R, C> toWriteOrig, Arguments args, Pipe... pipes);