added check before setting checkpoint directory

Author: Nitish1814

spark/core/src/main/java/zingg/spark/core/executor/ZinggSparkContext.java

@@ -2,6 +2,7 @@
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.spark.SparkContext;
 import org.apache.spark.api.java.JavaSparkContext;
 import org.apache.spark.sql.Column;
 import org.apache.spark.sql.Dataset;
@@ -71,12 +72,16 @@ public void init(IZinggLicense license)
             	zSession = new ZSparkSession(spark, license);
             }
             if (ctx==null) {
-				ctx = JavaSparkContext.fromSparkContext(zSession.getSession().sparkContext());
+                SparkContext sparkContext = zSession.getSession().sparkContext();
+                if (sparkContext.getCheckpointDir().isEmpty()) {
+                    sparkContext.setCheckpointDir("/tmp/checkpoint");
+                }
+				ctx = JavaSparkContext.fromSparkContext(sparkContext);
 				JavaSparkContext.jarOfClass(IZingg.class);
 				LOG.debug("Context " + ctx.toString());
 				//initHashFns();
                 if (!ctx.getCheckpointDir().isPresent()) {
-                    ctx.setCheckpointDir("/tmp/checkpoint");
+                    ctx.setCheckpointDir(sparkContext.getCheckpointDir().get());
                 }
 				setUtils();
 			}

spark/core/src/test/java/zingg/spark/core/executor/ZinggSparkTester.java

@@ -6,6 +6,7 @@
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.spark.SparkContext;
 import org.apache.spark.api.java.JavaSparkContext;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
@@ -49,15 +50,20 @@ public static void setup() {
     				.master("local[*]")
     				.appName("Zingg" + "Junit")
     				.getOrCreate();
-    		ctx = new JavaSparkContext(spark.sparkContext());
+			SparkContext sparkContext = spark.sparkContext();
+			if (sparkContext.getCheckpointDir().isEmpty()) {
+				sparkContext.setCheckpointDir("/tmp/checkpoint");
+			}
+    		ctx = new JavaSparkContext(sparkContext);
     		JavaSparkContext.jarOfClass(IZingg.class);    
 			args = new Arguments();
 			zsCTX = new ZinggSparkContext();
 			zsCTX.ctx = ctx;
 			zSession = new ZSparkSession(spark, null);
 			zsCTX.zSession = zSession;
-			
-            ctx.setCheckpointDir("/tmp/checkpoint");	
+			if (!ctx.getCheckpointDir().isPresent()) {
+				ctx.setCheckpointDir(sparkContext.getCheckpointDir().get());
+			}
             zsCTX.setPipeUtil(new SparkPipeUtil(zSession));
             zsCTX.setDSUtil(new SparkDSUtil(zSession));
             zsCTX.setHashUtil(new SparkHashUtil(zSession));