added support for withColumns in ZFrame

Author: Nitish1814

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

@@ -76,6 +76,7 @@ public interface ZFrame<D, R, C> {
     public ZFrame<D, R, C> unionByName(ZFrame<D, R, C> other, boolean flag);
 
     public <A> ZFrame<D, R, C> withColumn(String s, A c);
+    public ZFrame<D, R, C> withColumns(String[] columns, C[] columnValues);
 
 
     public ZFrame<D, R, C> repartition(int num);

common/core/src/main/java/zingg/common/core/preprocess/casenormalize/CaseNormalizer.java

@@ -15,7 +15,7 @@ public abstract class CaseNormalizer<S,D,R,C,T> implements IMultiFieldPreprocess
 
     private static final long serialVersionUID = 1L;
     private static final String STRING_TYPE = "string";
-    protected static String name = "zingg.common.core.preprocess.caseNormalize.CaseNormalizer";
+    protected static String name = "zingg.common.core.preprocess.casenormalize.CaseNormalizer";
     public static final Log LOG = LogFactory.getLog(CaseNormalizer.class);
 
     private IContext<S, D, R, C, T> context;

spark/client/src/main/java/zingg/spark/client/SparkFrame.java

@@ -1,5 +1,6 @@
 package zingg.spark.client;
 
+import java.util.Arrays;
 import java.util.List;
 
 import org.apache.spark.sql.Column;
@@ -10,6 +11,7 @@
 import org.apache.spark.sql.types.StructField;
 
 import scala.collection.JavaConverters;
+import scala.collection.Seq;
 import zingg.common.client.FieldData;
 import zingg.common.client.ZFrame;
 import zingg.common.client.util.ColName;
@@ -203,6 +205,18 @@ public <A> ZFrame<Dataset<Row>, Row, Column> withColumn(String s, A c){
         return new SparkFrame(df.withColumn(s, functions.lit(c)));
     }
 
+    @Override
+    public ZFrame<Dataset<Row>, Row, Column> withColumns(String[] columns, Column[] columnValues) {
+        Seq<String> columnsSeq = JavaConverters.asScalaIteratorConverter(Arrays.asList(columns).iterator())
+                .asScala()
+                .toSeq();
+        Seq<Column> columnValuesSeq = JavaConverters.asScalaIteratorConverter(Arrays.asList(columnValues).iterator())
+                .asScala()
+                .toSeq();
+
+        return new SparkFrame(df.withColumns(columnsSeq, columnValuesSeq));
+    }
+
     public ZFrame<Dataset<Row>, Row, Column> repartition(int nul){
         return new SparkFrame(df.repartition(nul));
     }

spark/core/src/main/java/zingg/spark/core/preprocess/caseNormalize/SparkCaseNormalizer.java

@@ -20,7 +20,7 @@
 
 public class SparkCaseNormalizer extends CaseNormalizer<SparkSession, Dataset<Row>, Row, Column, DataType> {
     private static final long serialVersionUID = 1L;
-    protected static String name = "zingg.spark.core.preprocess.caseNormalize.SparkCaseNormalizer";
+    protected static String name = "zingg.spark.core.preprocess.casenormalize.SparkCaseNormalizer";
 
     public SparkCaseNormalizer() {
         super();
@@ -32,16 +32,11 @@ public SparkCaseNormalizer(IContext<SparkSession, Dataset<Row>, Row, Column, Dat
     @Override
     protected ZFrame<Dataset<Row>, Row, Column> applyCaseNormalizer(ZFrame<Dataset<Row>, Row, Column> incomingDataFrame, List<String> relevantFields) {
         String[] incomingDFColumns = incomingDataFrame.columns();
-        Seq<String> columnsSeq = JavaConverters.asScalaIteratorConverter(relevantFields.iterator())
-                .asScala()
-                .toSeq();
-        List<Column> caseNormalizedValues = new ArrayList<>();
-        for (String relevantField : relevantFields) {
-            caseNormalizedValues.add(lower(incomingDataFrame.col(relevantField)));
+        Column[] caseNormalizedValues = new Column[relevantFields.size()];
+        for (int idx = 0; idx < relevantFields.size(); idx++) {
+            caseNormalizedValues[idx] = lower(incomingDataFrame.col(relevantFields.get(idx)));
         }
-        Seq<Column> caseNormalizedSeq = JavaConverters.asScalaIteratorConverter(caseNormalizedValues.iterator())
-                .asScala()
-                .toSeq();
-        return new SparkFrame(incomingDataFrame.df().withColumns(columnsSeq, caseNormalizedSeq)).select(incomingDFColumns);
+
+        return incomingDataFrame.withColumns(incomingDFColumns, caseNormalizedValues).select(incomingDFColumns);
     }
 }