py work on peek model.

zinggAI · Mar 10, 2023 · 55708e1 · 55708e1
1 parent 180afef
commit 55708e1
Show file tree

Hide file tree

Showing 5 changed files with 34 additions and 13 deletions.
diff --git a/common/core/src/main/java/zingg/common/core/Context.java b/common/core/src/main/java/zingg/common/core/Context.java
@@ -31,6 +31,11 @@ public interface Context <S,D, R, C,T> extends Serializable {
 
     public void init(String license)
         throws ZinggClientException;
+
+    /**convenience method to set all utils
+     * especially useful when you dont want to create the connection/spark context etc
+     * */
+    public void setUtils();
 
     public S getSession();
 

diff --git a/spark/client/src/main/java/zingg/spark/client/SparkClient.java b/spark/client/src/main/java/zingg/spark/client/SparkClient.java
@@ -24,12 +24,13 @@ public SparkClient(Arguments args, ClientOptions options) throws ZinggClientExce
 	}
 
 	public SparkClient() {
-		SparkSession session = SparkSession
+		/*SparkSession session = SparkSession
                 .builder()
                 .appName("Zingg")
                 .getOrCreate();
-		JavaSparkContext ctx = new JavaSparkContext(session.sparkContext());
+		JavaSparkContext ctx = JavaSparkContext.fromSparkContext(session.sparkContext());
         JavaSparkContext.jarOfClass(IZingg.class);
+		*/
 
 	}
 

diff --git a/.../zingg/spark/core/executor/PeekModel.java → ...g/spark/core/executor/SparkPeekModel.java b/.../zingg/spark/core/executor/PeekModel.java → ...g/spark/core/executor/SparkPeekModel.java
@@ -7,26 +7,35 @@
 import org.apache.commons.logging.LogFactory;
 import org.apache.spark.deploy.PythonRunner;
 
+import org.apache.spark.sql.Column;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.types.DataType;
+
 import zingg.common.client.Arguments;
 import zingg.common.client.ClientOptions;
 import zingg.common.client.ZinggClientException;
 import zingg.common.client.ZinggOptions;
 import zingg.common.core.executor.ZinggBase;
 
-public class PeekModel extends ZinggBase{
+public class SparkPeekModel extends ZinggBase<SparkSession, Dataset<Row>, Row, Column, DataType>{
 
-	protected static String name = "zingg.spark.core.executor.PeekModel";
-	public static final Log LOG = LogFactory.getLog(PeekModel.class); 
+	protected static String name = "zingg.spark.core.executor.SparkPeekModel";
+	public static final Log LOG = LogFactory.getLog(SparkPeekModel.class); 
 
-	public PeekModel() {
+	public SparkPeekModel() {
 		setZinggOptions(ZinggOptions.PEEK_MODEL);
+		setContext(new ZinggSparkContext());
 	}
 
 	@Override
     public void init(Arguments args, String license)
         throws ZinggClientException {
         startTime = System.currentTimeMillis();
-        this.args = args;        
+        this.args = args;     
+		//dont call init here as spark session already created   
+		getContext().setUtils();
     }
 
 	@Override

diff --git a/spark/core/src/main/java/zingg/spark/core/executor/SparkZFactory.java b/spark/core/src/main/java/zingg/spark/core/executor/SparkZFactory.java
@@ -33,6 +33,7 @@ public SparkZFactory() {}
         zinggers.put(ZinggOptions.UPDATE_LABEL, SparkLabelUpdater.name);
         zinggers.put(ZinggOptions.FIND_AND_LABEL, SparkFindAndLabeller.name);
         zinggers.put(ZinggOptions.RECOMMEND, SparkRecommender.name);
+        zinggers.put(ZinggOptions.PEEK_MODEL, SparkPeekModel.name);
     }
 
     public IZingg get(ZinggOptions z) throws InstantiationException, IllegalAccessException, ClassNotFoundException {

diff --git a/spark/core/src/main/java/zingg/spark/core/executor/ZinggSparkContext.java b/spark/core/src/main/java/zingg/spark/core/executor/ZinggSparkContext.java
@@ -68,19 +68,24 @@ public void init(String license)
             //initHashFns();
 
             ctx.setCheckpointDir("/tmp/checkpoint");	
-            setPipeUtil(new SparkPipeUtil(spark));
-            setDSUtil(new SparkDSUtil(spark));
-            setHashUtil(new SparkHashUtil(spark));
-            setGraphUtil(new SparkGraphUtil());
-            setModelUtil(new SparkModelUtil(spark));
-            setBlockingTreeUtil(new SparkBlockingTreeUtil(spark, getPipeUtil()));
+            setUtils();
         }
         catch(Throwable e) {
             if (LOG.isDebugEnabled()) e.printStackTrace();
             throw new ZinggClientException(e.getMessage());
         }
     }
 
+    @Override
+    public void setUtils() {
+        setPipeUtil(new SparkPipeUtil(spark));
+        setDSUtil(new SparkDSUtil(spark));
+        setHashUtil(new SparkHashUtil(spark));
+        setGraphUtil(new SparkGraphUtil());
+        setModelUtil(new SparkModelUtil(spark));
+        setBlockingTreeUtil(new SparkBlockingTreeUtil(spark, getPipeUtil()));
+    }
+
     /** 
     public void initHashFns() throws ZinggClientException {
 		try {