Java UDF

• MaxCompute数据类型与Java数据类型的对应关系，如下所示。

  MaxCompute Type	Java Type
  TINYINT	java.lang.Byte
  SMALLINT	java.lang.Short
  INT	java.lang.Integer
  BIGINT	java.lang.Long
  FLOAT	java.lang.Float
  DOUBLE	java.lang.Double
  DECIMAL	java.math.BigDecimal
  BOOLEAN	java.lang.Boolean
  STRING	java.lang.String
  VARCHAR	com.aliyun.odps.data.Varchar
  BINARY	com.aliyun.odps.data.Binary
  DATETIME	java.util.Date
  TIMESTAMP	java.sql.Timestamp
  ARRAY	java.util.List
  MAP	java.util.Map
  STRUCT	com.aliyun.odps.data.Struct

  说明

  • 在UDF中输入或输出参数的类型请务必使用Java Type，否则会报错ODPS-0130071。
  • Java中对应的数据类型以及返回值数据类型是对象，数据类型首字母需大写。
  • SQL中的NULL值通过Java中的NULL引用表示，因此不允许使用Java Primitive Type，因为无法表示SQL中的NULL值。
  • 此处ARRAY类型对应的Java类型是List，而不是数组。
• MaxCompute 2.0版本支持定义Java UDF时，使用Writable类型作为参数和返回值。下面为MaxCompute数据类型和Java Writable类型的映射关系。

  MaxCompute Type	Java Writable Type
  TINYINT	ByteWritable
  SMALLINT	ShortWritable
  INT	IntWritable
  BIGINT	LongWritable
  FLOAT	FloatWritable
  DOUBLE	DoubleWritable
  DECIMAL	BigDecimalWritable
  BOOLEAN	BooleanWritable
  STRING	Text
  VARCHAR	VarcharWritable
  BINARY	BytesWritable
  DATETIME	DatetimeWritable
  TIMESTAMP	TimestampWritable
  INTERVAL_YEAR_MONTH	IntervalYearMonthWritable
  INTERVAL_DAY_TIME	IntervalDayTimeWritable
  ARRAY	N/A
  MAP	N/A
  STRUCT	N/A

package org.alidata.odps.udf.examples; 
  import com.aliyun.odps.udf.UDF; 

public final class Lower extends UDF { 
  public String evaluate(String s) { 
    if (s == null) { 
        return null; 
    } 
        return s.toLowerCase(); 
  } 
}

package com.aliyun.odps.udf.example;
import com.aliyun.odps.io.Text;
import com.aliyun.odps.udf.UDF;
public class MyConcat extends UDF {
  private Text ret = new Text();
  public Text evaluate(Text a, Text b) {
    if (a == null || b == null) {
      return null;
    }
    ret.clear();
    ret.append(a.getBytes(), 0, a.getLength());
    ret.append(b.getBytes(), 0, b.getLength());
    return ret;
  }
}

import com.aliyun.odps.udf.ContextFunction;
import com.aliyun.odps.udf.ExecutionContext;
import com.aliyun.odps.udf.UDFException;
public abstract class Aggregator implements ContextFunction {
    @Override
    public void setup(ExecutionContext ctx) throws UDFException {
    }
    @Override
    public void close() throws UDFException {
    }
    /**
     * 创建聚合Buffer * @return Writable 聚合buffer。
     */
    abstract public Writable newBuffer();
    /**
     * @param buffer 聚合buffer * @param args SQL中调用UDAF时指定的参数，不能为null，但是args里面的元素可以为null，代表对应的输入数据是null * @throws UDFException。
     */
    abstract public void iterate(Writable buffer, Writable[] args) throws UDFException;

    /**
     * 生成最终结果 * @param buffer * @return Object UDAF的最终结果 * @throws UDFException。
     */
    abstract public Writable terminate(Writable buffer) throws UDFException;
    abstract public void merge(Writable buffer, Writable partial) throws UDFException;
}

• 第一阶段：每个Worker统计分片内数据的个数及汇总值。您可以将每个分片内的数据个数及汇总值视为一个中间结果。
• 第二阶段：Worker汇总上一个阶段中每个分片内的信息。在最终输出时，r.sum/r.count即是所有输入数据的平均值。

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import com.aliyun.odps.io.DoubleWritable;
import com.aliyun.odps.io.Writable;
import com.aliyun.odps.udf.Aggregator;
import com.aliyun.odps.udf.UDFException;
import com.aliyun.odps.udf.annotation.Resolve;
@Resolve("double->double")
public class AggrAvg extends Aggregator {
  private static class AvgBuffer implements Writable {
    private double sum = 0;
    private long count = 0;
    @Override
    public void write(DataOutput out) throws IOException {
      out.writeDouble(sum);
      out.writeLong(count);
    }
    @Override
    public void readFields(DataInput in) throws IOException {
      sum = in.readDouble();
      count = in.readLong();
    }
  }
  private DoubleWritable ret = new DoubleWritable();
  @Override
  public Writable newBuffer() {
    return new AvgBuffer();
  }
  @Override
  public void iterate(Writable buffer, Writable[] args) throws UDFException {
    DoubleWritable arg = (DoubleWritable) args[0];
    AvgBuffer buf = (AvgBuffer) buffer;
    if (arg != null) {
      buf.count += 1;
      buf.sum += arg.get();
    }
  }
  @Override
  public Writable terminate(Writable buffer) throws UDFException {
    AvgBuffer buf = (AvgBuffer) buffer;
    if (buf.count == 0) {
      ret.set(0);
    } else {
      ret.set(buf.sum / buf.count);
    }
    return ret;
  }
  @Override
  public void merge(Writable buffer, Writable partial) throws UDFException {
    AvgBuffer buf = (AvgBuffer) buffer;
    AvgBuffer p = (AvgBuffer) partial;
    buf.sum += p.sum;
    buf.count += p.count;
  }
}

package com.aliyun.odps.udf.example;
import com.aliyun.odps.io.Text;
import com.aliyun.odps.udf.UDF;
public class MyConcat extends UDF {
  private Text ret = new Text();
  public Text evaluate(Text a, Text b) {
    if (a == null || b == null) {
      return null;
    }
    ret.clear();
    ret.append(a.getBytes(), 0, a.getLength());
    ret.append(b.getBytes(), 0, b.getLength());
    return ret;
  }
}

• Writable[] writables：表示一行数据。代码中是指传入的列，例如writables[0]表示第一列，writables[1]表示第二列。
• iterate(Writable writable,Writable[] writables)方法：writable是指一个阶段性的汇总数据，在不同的Map任务中，group by后得出的数据（可理解为一个集合），每行执行一次。
• merge()方法：将不同的Map直接结算的结果进行汇总。
• terminate()方法：返回数据。
• newBuffer()方法：创建初始返回结果的值。
• Writable的readFields方法， 由于partial的writable对象是重用的，同一个对象的readFields方法会被调用多次。该方法每次调用的时候重置整个对象，如果对象中包含Collection，则需要清空。
• UDAF在SQL中的使用语法与普通的内建聚合函数相同，详情请参见聚合函数。
• 运行UDTF的方法与UDF类似，详情请参见运行 UDF。
• STRING对应的Writable类型为Text。

1. 编写UDTF程序，编译成功后导出Jar包（udtfexample1.jar）。

   package com.aliyun.odps.examples.udf;
   import java.io.BufferedReader;
   import java.io.IOException;
   import java.io.InputStream;
   import java.io.InputStreamReader;
   import java.util.Iterator;
   import com.aliyun.odps.udf.ExecutionContext;
   import com.aliyun.odps.udf.UDFException;
   import com.aliyun.odps.udf.UDTF;
   import com.aliyun.odps.udf.annotation.Resolve;
   /**
    * project: example_project 
    * table: wc_in2 
    * partitions: p2=1,p1=2 
    * columns: colc,colb
    */
   @Resolve("string,string->string,bigint,string")
   public class UDTFResource extends UDTF {
     ExecutionContext ctx;
     long fileResourceLineCount;
     long tableResource1RecordCount;
     long tableResource2RecordCount;
     @Override
     public void setup(ExecutionContext ctx) throws UDFException {
     this.ctx = ctx;
     try {
      InputStream in = ctx.readResourceFileAsStream("file_resource.txt");
      BufferedReader br = new BufferedReader(new InputStreamReader(in));
      String line;
      fileResourceLineCount = 0;
      while ((line = br.readLine()) != null) {
        fileResourceLineCount++;
      }
      br.close();
      Iterator<Object[]> iterator = ctx.readResourceTable("table_resource1").iterator();
      tableResource1RecordCount = 0;
      while (iterator.hasNext()) {
        tableResource1RecordCount++;
        iterator.next();
      }
      iterator = ctx.readResourceTable("table_resource2").iterator();
      tableResource2RecordCount = 0;
      while (iterator.hasNext()) {
        tableResource2RecordCount++;
        iterator.next();
      }
    } catch (IOException e) {
      throw new UDFException(e);
    }
   }
      @Override
      public void process(Object[] args) throws UDFException {
        String a = (String) args[0];
        long b = args[1] == null ? 0 : ((String) args[1]).length();
        forward(a, b, "fileResourceLineCount=" + fileResourceLineCount + "|tableResource1RecordCount="
        + tableResource1RecordCount + "|tableResource2RecordCount=" + tableResource2RecordCount);
       }
   }

2. 添加资源到MaxCompute。

   Add file file_resource.txt;
   Add jar udtfexample1.jar;
   Add table table_resource1 as table_resource1;
   Add table table_resource2 as table_resource2;

3. 在MaxCompute中创建UDTF函数（my_udtf）。

   create function mp_udtf as com.aliyun.odps.examples.udf.UDTFResource using 
   'udtfexample1.jar, file_resource.txt, table_resource1, table_resource2';

4. 运行该UDTF。

   select mp_udtf("10","20") as (a, b, fileResourceLineCount) from tmp1;  
   --返回结果如下。
   +-------+------------+-------+
   | a | b      | fileResourceLineCount |
   +-------+------------+-------+
   | 10    | 2          | fileResourceLineCount=3|tableResource1RecordCount=0|tableResource2RecordCount=0 |
   | 10    | 2          | fileResourceLineCount=3|tableResource1RecordCount=0|tableResource2RecordCount=0 |
   +-------+------------+-------+

import com.aliyun.odps.udf.UDF;
import com.aliyun.odps.udf.annotation.Resolve;
@Resolve("struct,string->string")
public class UdfArray extends UDF {
    public String evaluate(List vals, Long len) {
        return vals.get(len.intValue());
    }

    public String evaluate(Map map, String key) {
        return map.get(key);
    }

    public String evaluate(Struct struct, String key) {
        return struct.getFieldValue("a") + key;
    }
}

create function my_index as 'UdfArray' using 'myjar.jar'; 
select id, my_index(array('red', 'yellow', 'green'), colorOrdinal) as color_name from co

package com.aliyun.odps.compiler.hive;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
public class Collect extends GenericUDF {
  @Override
  public ObjectInspector initialize(ObjectInspector[] objectInspectors) throws UDFArgumentException {
    if (objectInspectors.length == 0) {
      throw new UDFArgumentException("Collect: input args should >= 1");
    }
    for (int i = 1; i < objectInspectors.length; i++) {
      if (objectInspectors[i] != objectInspectors[0]) {
        throw new UDFArgumentException("Collect: input oi should be the same for all args");
      }
    }
    return ObjectInspectorFactory.getStandardListObjectInspector(objectInspectors[0]);
  }
  @Override
  public Object evaluate(DeferredObject[] deferredObjects) throws HiveException {
    List<Object> objectList = new ArrayList<>(deferredObjects.length);
    for (DeferredObject deferredObject : deferredObjects) {
      objectList.add(deferredObject.get());
    }
    return objectList;
  }
  @Override
  public String getDisplayString(String[] strings) {
    return "Collect";
  }
}

--添加资源。
Add jar test.jar;
--创建function。
CREATE FUNCTION hive_collect as 'com.aliyun.odps.compiler.hive.Collect' using 'test.jar';
--使用function。
set odps.sql.hive.compatible=true;
select hive_collect(4y,5y,6y) from dual;
+------+
| _c0  |
+------+
| [4, 5, 6] |
+------+

• 创建UDF时需要指定Jar包，无法自动将所有Jar包加入Classpath。
  说明 MaxCompute的add jar命令会永久地在Project中创建一个资源。
• 执行SQL语句时，需要在SQL前加set odps.sql.hive.compatible=true;语句，set语句和SQL语句一起提交执行。
• 请注意MaxCompute的Java沙箱限制。