[WIP][DO NOT MERGE] Draft implementation of the new PySpark API for support of both Spark Classic and Spark Connect

.github/workflows/codeql.yml

@@ -13,11 +13,16 @@ name: "CodeQL"
 
 on:
   push:
-
+    branches: [ main, 0.4.0 ]
   pull_request:
+    # The branches below must be a subset of the branches above
+    branches: [ main, 0.4.0 ]
     paths-ignore:
       - '**/*.md'
       - '**/*.txt'
+>>>>>>> 0.4.0
+  schedule:
+    - cron: '22 3 * * 5'
 
 jobs:
   analyze:

.gitignore

@@ -23,3 +23,30 @@ python/docs/_build/_doctrees
 **/python/build/*
 **/assembly/.classpath
 **/.DS_Store
+
+# Python stuff
+.env
+.venv
+
+# Sphinx _build
+**/_build
+
+# Helix stuff
+.helix
+
+# Emacs stuff
+.dir-locals.el
+
+# JDTLS stuff
+.package
+.classpath
+.project
+.settings
+.factorypath
+
+# Metals LSP
+.metals
+.bloop
+
+# Hadoop & Spark binaries
+spark-*

assembly/dependency-reduced-pom.xml

@@ -45,7 +45,7 @@
       </plugin>
       <plugin>
         <artifactId>maven-assembly-plugin</artifactId>
-        <version>2.4.1</version>
+        <version>3.6.0</version>
         <executions>
           <execution>
             <id>make-assembly</id>

assembly/pom.xml

@@ -112,7 +112,7 @@
 			</plugin>
 			<plugin>
 				<artifactId>maven-assembly-plugin</artifactId>
-				<version>2.4.1</version>
+				<version>3.6.0</version>
 				<configuration>
 					<descriptors>
 						<descriptor>${project.basedir}/src/assembly/dist.xml</descriptor>

buf.gen.yaml

@@ -0,0 +1,13 @@
+version: v1
+plugins:
+  # Building the Java classes
+  - plugin: buf.build/protocolbuffers/java:v25.3
+    out: spark/client/src/main/java
+  # Building the Python build and building the mypy interfaces.
+  - plugin: buf.build/protocolbuffers/python:v25.3
+    out: python/zingg_v2/proto
+  - plugin: buf.build/grpc/python:v1.62.0
+    out: python/zingg_v2/proto
+  - plugin: buf.build/community/nipunn1313-mypy:v3.5.0
+    out: python/zingg_v2/proto
+

buf.work.yaml

@@ -0,0 +1,3 @@
+version: v1
+directories:
+  - protobuf

common/client/src/main/java/zingg/common/client/Arguments.java

@@ -4,6 +4,7 @@
 import java.io.Serializable;
 import java.io.StringWriter;
 import java.util.List;
+import java.util.stream.Collectors;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -16,6 +17,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 
 import zingg.common.client.pipe.Pipe;
+import zingg.common.client.util.JsonStringify;
 
 
 /**
@@ -169,7 +171,7 @@ public void setLabelDataSampleSize(float labelDataSampleSize) throws ZinggClient
 	public List<? extends FieldDefinition> getFieldDefinition() {
 		return fieldDefinition;
 	}
-
+	
 	/**
 	 * Set the field definitions consisting of match field indices, types and
 	 * classes
@@ -308,18 +310,7 @@ public void checkNullBlankEmpty(Pipe[] field, String fieldName) throws ZinggClie
 
 	@Override
 	public String toString() {
-		ObjectMapper mapper = new ObjectMapper();
-		mapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_CONTROL_CHARS,
-				true);
-		//mapper.configure(JsonParser.Feature.FAIL_ON_EMPTY_BEANS, true)
-		try {
-			StringWriter writer = new StringWriter();
-			return mapper.writeValueAsString(this);
-		} catch (IOException e) {
-			// TODO Auto-generated catch block
-			e.printStackTrace();
-			return null;
-		}
+		return JsonStringify.toString(this);
 	}
 
 	/**

common/client/src/main/java/zingg/common/client/ArgumentsUtil.java

@@ -19,7 +19,7 @@
 
 public class ArgumentsUtil {
 
-	protected Class<? extends Arguments> argsClass;
+	protected Class<? extends IArguments> argsClass;
 	private static final String ENV_VAR_MARKER_START = "$";
 	private static final String ENV_VAR_MARKER_END = "$";
 	private static final String ESC = "\\";
@@ -31,7 +31,7 @@ public ArgumentsUtil() {
 		this(Arguments.class);
 	}
 
-	public ArgumentsUtil( Class<? extends Arguments> argsClass) {
+	public ArgumentsUtil( Class<? extends IArguments> argsClass) {
 		this.argsClass = argsClass;
 	}
 

common/client/src/main/java/zingg/common/client/Client.java

@@ -5,9 +5,17 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
-import zingg.common.client.license.IZinggLicense;
+import zingg.common.client.event.events.IEvent;
+import zingg.common.client.event.events.ZinggStartEvent;
+import zingg.common.client.event.events.ZinggStopEvent;
+import zingg.common.client.event.listeners.EventsListener;
+import zingg.common.client.event.listeners.IEventListener;
+import zingg.common.client.event.listeners.ZinggStartListener;
+import zingg.common.client.event.listeners.ZinggStopListener;
+import zingg.common.client.options.ZinggOptions;
 import zingg.common.client.util.Email;
 import zingg.common.client.util.EmailBody;
+import zingg.common.client.util.PipeUtilBase;
 
 /**
  * This is the main point of interface with the Zingg matching product.
@@ -22,9 +30,11 @@ public abstract class Client<S,D,R,C,T> implements Serializable {
 	protected IZingg<S,D,R,C> zingg;
 	protected ClientOptions options;
 	protected S session;
-
+	protected PipeUtilBase<S,D,R,C> pipeUtil;
 	public static final Log LOG = LogFactory.getLog(Client.class);
 
+	protected String zFactoryClassName;
+
 
 	/**
 	 * Construct a client to Zingg using provided arguments and spark master.
@@ -36,10 +46,14 @@ public abstract class Client<S,D,R,C,T> implements Serializable {
 	 *             if issue connecting to master
 	 */
 
-	public Client() {}
+	public Client(String zFactory) {
+		setZFactoryClassName(zFactory);
+	}
 
-	public Client(IArguments args, ClientOptions options) throws ZinggClientException {
-		setOptions(options);
+	public Client(IArguments args, ClientOptions options, String zFactory) throws ZinggClientException {
+		setZFactoryClassName(zFactory);
+		this.options = options;
+    	setOptions(options);
 		try {
 			buildAndSetArguments(args, options);
 			printAnalyticsBanner(arguments.getCollectMetrics());
@@ -51,14 +65,28 @@ public Client(IArguments args, ClientOptions options) throws ZinggClientExceptio
 		}
 	}
 
-	public Client(IArguments args, ClientOptions options, S s) throws ZinggClientException {
-		this(args, options);
+
+    public String getZFactoryClassName() {
+        return zFactoryClassName;
+    }
+
+    public void setZFactoryClassName(String s) {
+        this.zFactoryClassName = s;
+    }
+
+	public Client(IArguments args, ClientOptions options, S s, String zFactory) throws ZinggClientException {
+		this(args, options, zFactory);
 		this.session = s;
 		LOG.debug("Session passed is " + s);
 		if (session != null) zingg.setSession(session);
 	}
 
-	public abstract IZinggFactory getZinggFactory() throws Exception;//(IZinggFactory) Class.forName("zingg.ZFactory").newInstance();
+
+	public IZinggFactory getZinggFactory() throws InstantiationException, IllegalAccessException, ClassNotFoundException{
+        LOG.debug("z factory is " + getZFactoryClassName());
+		return (IZinggFactory) Class.forName(getZFactoryClassName()).newInstance();
+	}
+
 
 
 
@@ -70,9 +98,10 @@ public void setZingg(IArguments args, ClientOptions options) throws Exception{
 		catch(Exception e) {
 			e.printStackTrace();
 			//set default
-			setZingg(zf.get(ZinggOptions.getByValue(ZinggOptions.PEEK_MODEL.getValue())));
+			setZingg(zf.get(ZinggOptions.getByValue(ZinggOptions.PEEK_MODEL.getName())));
 		}
 	}
+
 
 	public void setZingg(IZingg<S,D,R,C> zingg) {
 		this.zingg = zingg; 
@@ -120,7 +149,7 @@ else if (args.getJobId() != -1) {
 	}
 
 	public void printBanner() {
-		String versionStr = "0.4.0";
+		String versionStr = "0.4.1-SNAPSHOT";
 		LOG.info("");
 		LOG.info("********************************************************");
 		LOG.info("*                    Zingg AI                          *");
@@ -155,23 +184,27 @@ public void printAnalyticsBanner(boolean collectMetrics) {
 	}
 
 	public abstract Client<S,D,R,C,T> getClient(IArguments args, ClientOptions options) throws ZinggClientException;
+
+	public ClientOptions getClientOptions(String ... args){
+		return new ClientOptions(args);
+	}
 
 	public void mainMethod(String... args) {
 		printBanner();
 		Client<S,D,R,C,T> client = null;
 		ClientOptions options = null;
 		try {
+
 			for (String a: args) LOG.debug("args " + a);
-			options = new ClientOptions(args);
+			options = getClientOptions(args);
 			setOptions(options);
-		
+
 			if (options.has(options.HELP) || options.has(options.HELP1) || options.get(ClientOptions.PHASE) == null) {
 				LOG.warn(options.getHelp());
 				System.exit(0);
 			}
 			String phase = options.get(ClientOptions.PHASE).value.trim();
 			ZinggOptions.verifyPhase(phase);
-			IArguments arguments = null;
 			if (options.get(ClientOptions.CONF).value.endsWith("json")) {
 					arguments = getArgsUtil().createArgumentsFromJSON(options.get(ClientOptions.CONF).value, phase);
 			}
@@ -184,6 +217,7 @@ else if (options.get(ClientOptions.CONF).value.endsWith("env")) {
 
 			client = getClient(arguments, options);
 			client.init();
+			// after setting arguments etc. as some of the listeners need it
 			client.execute();
 			client.postMetrics();
 			LOG.warn("Zingg processing has completed");				
@@ -212,6 +246,7 @@ else if (options.get(ClientOptions.CONF).value.endsWith("env")) {
 		}
 		finally {
 			try {
+				EventsListener.getInstance().fireEvent(new ZinggStopEvent());
 				if (client != null) {
 					//client.postMetrics();
 					client.stop();
@@ -228,14 +263,12 @@ else if (options.get(ClientOptions.CONF).value.endsWith("env")) {
 	}
 
 	public void init() throws ZinggClientException {
-		zingg.setClientOptions(options);
-		zingg.init(getArguments(), getLicense(options.get(ClientOptions.LICENSE).value.trim()));
+		zingg.init(getArguments(), getSession());
 		if (session != null) zingg.setSession(session);
-
+		initializeListeners();
+		EventsListener.getInstance().fireEvent(new ZinggStartEvent());					
 	}
 
-	protected abstract IZinggLicense getLicense(String license)  throws ZinggClientException ;
-
 	/**
 	 * Stop the Spark job running context
 	 */
@@ -305,5 +338,26 @@ protected ArgumentsUtil getArgsUtil() {
 		}
 		return argsUtil;
 	}    
+
+	public void addListener(Class<? extends IEvent> eventClass, IEventListener listener) {
+        EventsListener.getInstance().addListener(eventClass, listener);
+    }
+
+    public void initializeListeners() {
+        addListener(ZinggStartEvent.class, new ZinggStartListener());
+        addListener(ZinggStopEvent.class, new ZinggStopListener());
+    }
+
+    public abstract S getSession();
+
+    public void setSession(S s) {
+    	this.session = s;
+    }
+
+	public abstract PipeUtilBase<S, D, R, C> getPipeUtil();
+
+	public void setPipeUtil(PipeUtilBase<S, D, R, C> pipeUtil) {
+		this.pipeUtil = pipeUtil;
+	}
 
 }