Removing inMemory Pipes (#1171)

* removing inmemory pipes

* moving to ent

* moving to ent

* access change

Author: sania-16

common/client/src/main/java/zingg/common/client/pipe/Pipe.java

@@ -33,7 +33,6 @@ public class Pipe<D,R,C> implements Serializable{ // St:StructType, Sv:SaveMode
 	public static final String FORMAT_ELASTIC = "org.elasticsearch.spark.sql";
 	public static final String FORMAT_EXASOL = "com.exasol.spark";
 	public static final String FORMAT_BIGQUERY = "bigquery";
-	public static final String FORMAT_INMEMORY = "inMemory";
 
 	String name;
 	String format;
@@ -44,25 +43,18 @@ public class Pipe<D,R,C> implements Serializable{ // St:StructType, Sv:SaveMode
 	String schema;
 	String mode;
 
-	
-
-	
-
 	public String getSchema() {
 		return schema;
 	}
 
-
 	public void setSchema(String schema) {
 		this.schema = schema;
 	}
 
-
 	public String getName() {
 		return name;
 	}
 
-	
 	@JsonValue
 	public void setName(String name) {
 		this.name = name;		
@@ -76,7 +68,6 @@ public String getFormat() {
 	public void setFormat(String sinkType) {
 		this.format = sinkType;
 	}
-
 
 	@JsonValue
 	public void setProps(Map<String, String> props) {
@@ -102,23 +93,18 @@ public String get(String key) {
 		return props.get(key);
 	}
 
-	
 	public String getPreprocessors() {
 		return preprocessors;
 	}
 
-
 	public void setPreprocessors(String preprocessors) {
 		this.preprocessors = preprocessors;
 	}
 
-
-
 	public int getId() {
 		return id;
 	}
 
-
 	public void setId(int recId) {
 		this.id = recId;
 	}

common/client/src/main/java/zingg/common/client/util/PipeUtil.java

@@ -9,7 +9,6 @@
 import zingg.common.client.ZFrame;
 import zingg.common.client.ZinggClientException;
 import zingg.common.client.pipe.FilePipe;
-//import zingg.common.client.pipe.InMemoryPipe;
 import zingg.common.client.pipe.Pipe;
 
 //import com.datastax.spark.connector.cql.*;
@@ -55,20 +54,18 @@ protected  ZFrame<D,R,C> read(DFReader<D,R,C> reader, Pipe<D,R,C> p, boolean add
 		LOG.warn("Reading " + p);
 		try {
 
-		if (p.getFormat().equals(Pipe.FORMAT_INMEMORY)) {
-			input = p.getDataset(); //.df();
-		}
-		else {		
+				
 			if (p.getProps().containsKey(FilePipe.LOCATION)) {
 				input = reader.load(p.get(FilePipe.LOCATION));
 			}
 			else {
 				input = reader.load();
 			}
-    }
+    
 			if (addSource) {
 				input = input.withColumn(ColName.SOURCE_COL, p.getName());
 			}
+
 			p.setDataset(input);
 		} catch (Exception ex) {
 			LOG.warn(ex.getMessage());
@@ -195,10 +192,6 @@ public  void write(ZFrame<D,R,C> toWriteOrig,
 
 			LOG.warn("Writing output " + p);
 
-			if (p.getFormat().equals(Pipe.FORMAT_INMEMORY)) {
- 				p.setDataset(toWriteOrig);
-				return;
-			}
 			//SparkPipe sPipe = (SparkPipe) p;
 			if (p.getMode() != null) {
 				writer.setMode(p.getMode()); //SaveMode.valueOf(p.getMode()));

docs/python/markdown/index.md

@@ -102,9 +102,6 @@ Requires **python 3.6+**; **spark 3.5.0.** Otherwise, [`zingg.client.Zingg()`](z
     * [`CsvPipe.setDelimiter()`](zingg.md#zingg.pipes.CsvPipe.setDelimiter)
     * [`CsvPipe.setHeader()`](zingg.md#zingg.pipes.CsvPipe.setHeader)
     * [`CsvPipe.setLocation()`](zingg.md#zingg.pipes.CsvPipe.setLocation)
-  * [`InMemoryPipe`](zingg.md#zingg.pipes.InMemoryPipe)
-    * [`InMemoryPipe.getDataset()`](zingg.md#zingg.pipes.InMemoryPipe.getDataset)
-    * [`InMemoryPipe.setDataset()`](zingg.md#zingg.pipes.InMemoryPipe.setDataset)
   * [`Pipe`](zingg.md#zingg.pipes.Pipe)
     * [`Pipe.addProperty()`](zingg.md#zingg.pipes.Pipe.addProperty)
     * [`Pipe.getPipe()`](zingg.md#zingg.pipes.Pipe.getPipe)

docs/python/markdown/zingg.md

@@ -643,32 +643,6 @@ Method to set location of pipe
 * **Parameters:**
   **location** (*String*) – location from where we read data
 
-### *class* zingg.pipes.InMemoryPipe(name, df=None)
-
-Bases: [`Pipe`](#zingg.pipes.Pipe)
-
-Pipe Class for working with InMemory pipeline
-
-* **Parameters:**
-  * **name** (*String*) – name of the pipe
-  * **df** (*Dataset* *or* *None*) – provide dataset for this pipe (optional)
-
-#### getDataset()
-
-Method to get Dataset from pipe
-
-* **Returns:**
-  dataset of the pipe in the format of spark dataset
-* **Return type:**
-  Dataset<Row>
-
-#### setDataset(df)
-
-Method to set DataFrame of the pipe
-
-* **Parameters:**
-  **df** (*DataFrame*) – pandas or spark dataframe for the pipe
-
 ### *class* zingg.pipes.Pipe(name, format)
 
 Bases: `object`
@@ -677,7 +651,7 @@ Pipe class for working with different data-pipelines. Actual pipe def in the arg
 
 * **Parameters:**
   * **name** (*String*) – name of the pipe
-  * **format** (*Format*) – formate of pipe e.g. bigquery,InMemory, etc.
+  * **format** (*Format*) – formate of pipe e.g. bigquery,csv, etc.
 
 #### addProperty(name, value)
 

python/zingg/pipes.py

@@ -38,7 +38,7 @@ class Pipe:
 
     :param name: name of the pipe
     :type name: String
-    :param format: formate of pipe e.g. bigquery,InMemory, etc.
+    :param format: formate of pipe e.g. bigquery,csv, etc.
     :type format: Format
     """
 
@@ -251,47 +251,4 @@ def setDbTable(self, dbtable):
         :param dbtable: provide bucket parameter.
         :type dbtable: String
         """
-        Pipe.addProperty(self, "dbtable", dbtable)     
-        
-
-class InMemoryPipe(Pipe):
-    """ Pipe Class for working with InMemory pipeline
-
-    :param name: name of the pipe
-    :type name: String
-    :param df: provide dataset for this pipe (optional)
-    :type df: Dataset or None
-    """    
-
-    def __init__(self, name, df = None):
-        setupPipes()
-        Pipe.__init__(self, name, JPipe.FORMAT_INMEMORY)
-        if (df is not None):
-            self.setDataset(df)
-
-    def setDataset(self, df):
-        """ Method to set DataFrame of the pipe
-        
-        :param df: pandas or spark dataframe for the pipe
-        :type df: DataFrame
-        """
-        if (isinstance(df, pd.DataFrame)):
-            print('schema of pandas df is ' , Pipe.getPipe(self).getSchema())
-            if (Pipe.getPipe(self).getSchema() is not None):
-                ds = getSparkSession().createDataFrame(df, schema=Pipe.getPipe(self).getSchema())
-            else:
-                ds = getSparkSession().createDataFrame(df)
-            
-            Pipe.getPipe(self).setDataset(ds._jdf)
-        elif (isinstance(df, DataFrame)):
-            Pipe.getPipe(self).setDataset(df._jdf)
-        else:
-            LOG.error(" setDataset(): NUll or Unsupported type: %s", type(df))
-
-    def getDataset(self):
-        """ Method to get Dataset from pipe
-        
-        :return: dataset of the pipe in the format of spark dataset 
-        :rtype: Dataset<Row>
-        """
-        return Pipe.getPipe(self).getDataset().df()
+        Pipe.addProperty(self, "dbtable", dbtable)     

spark/client/src/main/java/zingg/spark/client/util/SparkDFReader.java

@@ -14,8 +14,8 @@
 
 public class SparkDFReader implements DFReader<Dataset<Row>, Row, Column> {
 
-    private SparkSession session;
-    private DataFrameReader reader;
+    protected SparkSession session;
+    protected DataFrameReader reader;
 
     public SparkDFReader(SparkSession s) {
         this.session = s;

spark/client/src/main/java/zingg/spark/client/util/SparkDFWriter.java

@@ -10,30 +10,33 @@
 import zingg.common.client.util.DFWriter;
 
 public class SparkDFWriter implements DFWriter<Dataset<Row>, Row, Column>{
-    private DataFrameWriter writer;
+
+    protected DataFrameWriter writer;
 
     public SparkDFWriter(ZFrame<Dataset<Row>, Row, Column> toWriteOrig) {
         Dataset<Row> toWrite = toWriteOrig.df();
 		this.writer = toWrite.write();
-        
     }
 
-
     public void setMode(String s) {
         this.writer.mode(SaveMode.valueOf(s));
 
     }
+
     public DFWriter<Dataset<Row>, Row, Column> format(String f) {
         writer.format(f);
         return this;
     }
+
     public DFWriter<Dataset<Row>, Row, Column> option(String k, String v) {
         writer.option(k,v);
         return this;
     }
+
     public void save(String location) {
         writer.save(location);
     }
+    
     public void save() {
         writer.save();
     }

spark/client/src/main/java/zingg/spark/client/util/SparkPipeUtil.java

@@ -7,7 +7,6 @@
 import org.apache.spark.sql.Row;
 
 import zingg.common.client.ZFrame;
-//import zingg.common.client.pipe.InMemoryPipe;
 import zingg.common.client.util.DFReader;
 import zingg.common.client.util.DFWriter;
 import zingg.common.client.util.PipeUtil;