Merge pull request #52 from mahmoudhanafy/port-UDF-to-java

Port UDFs to Java

Author: holdenk

src/main/java/com/highperformancespark/examples/dataframe/JavaUDFs.java

@@ -0,0 +1,74 @@
+package com.highperformancespark.examples.dataframe;
+
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SQLContext;
+import org.apache.spark.sql.expressions.MutableAggregationBuffer;
+import org.apache.spark.sql.expressions.UserDefinedAggregateFunction;
+import org.apache.spark.sql.types.*;
+
+public class JavaUDFs {
+
+  public static void setupUDFs(SQLContext sqlContext) {
+    sqlContext.udf().register("strlen", (String s) -> s.length(), DataTypes.StringType);
+  }
+
+  public static void setupUDAFs(SQLContext sqlContext) {
+
+    class Avg extends UserDefinedAggregateFunction {
+
+      @Override
+      public StructType inputSchema() {
+        StructType inputSchema =
+          new StructType(new StructField[]{new StructField("value", DataTypes.DoubleType, true, Metadata.empty())});
+        return inputSchema;
+      }
+
+      @Override
+      public StructType bufferSchema() {
+        StructType bufferSchema =
+          new StructType(new StructField[]{
+            new StructField("count", DataTypes.LongType, true, Metadata.empty()),
+            new StructField("sum", DataTypes.DoubleType, true, Metadata.empty())
+          });
+
+        return bufferSchema;
+      }
+
+      @Override
+      public DataType dataType() {
+        return DataTypes.DoubleType;
+      }
+
+      @Override
+      public boolean deterministic() {
+        return true;
+      }
+
+      @Override
+      public void initialize(MutableAggregationBuffer buffer) {
+        buffer.update(0, 0L);
+        buffer.update(1, 0.0);
+      }
+
+      @Override
+      public void update(MutableAggregationBuffer buffer, Row input) {
+        buffer.update(0, buffer.getLong(0) + 1);
+        buffer.update(1, buffer.getDouble(1) + input.getDouble(0));
+      }
+
+      @Override
+      public void merge(MutableAggregationBuffer buffer1, Row buffer2) {
+        buffer1.update(0, buffer1.getLong(0) + buffer2.getLong(0));
+        buffer1.update(1, buffer1.getDouble(1) + buffer2.getDouble(1));
+      }
+
+      @Override
+      public Object evaluate(Row buffer) {
+        return buffer.getDouble(1) / buffer.getLong(0);
+      }
+    }
+
+    Avg average = new Avg();
+    sqlContext.udf().register("ourAvg", average);
+  }
+}

src/main/scala/com/high-performance-spark-examples/dataframe/UDFs.scala

@@ -47,7 +47,7 @@ object UDFs {
       }
 
       def evaluate(buffer: Row): Any = {
-        math.pow(buffer.getDouble(1), 1.toDouble / buffer.getLong(0))
+        buffer.getDouble(1) / buffer.getLong(0)
       }
     }
     // Optionally register