MaxCompute:Java UDF

• Java UDF を介して Tinyint、Smallint、Int、Float､Varchar、Binary、Timestamp などの基本データ型を使用する方法は、次のとおりです。
  • UDTF では、 @Resolve アノテーションによって「署名」を取得します。たとえば、 @Resolve("smallint->varchar(10)")と記述します。
  • UDF ではリフレクション分析の「evaluate」によって「署名」が取得されます。 この場合、MaxCompute 組み込み型とJava 型は 1 対 1 に対応しています。
  • UDAF では @Resolve アノテーションを使用して署名が取得されます。Maxcompute2.0 ではアノテーションで新しいデータ型を使用できます。たとえば、 @Resolve("smallint-> varchar (10 )") と記述します。
• JAVA UDF では、「Array｣､｢Map｣､｢Struct｣の 3 つのデータ型が使用されます。
  • UDAF と UDTF では、@Resolveアノテーションによって署名を指定します。たとえば、 @Resolve("array<string>,struct<a1:bigint,b1:string>,string->map<string,bigint>,struct<b1:bigint>") と記述します。
  • UDF では、evaluate メソッドの署名を介して UDF の入力データ型と出力データ型がマッピングさます。このとき、Maxcompute のデータ型と Java のデータ型の対応付けが参照されます。 この関係では、Array 型 は java.util.List を、Map 型は java.util.Map を、Struct 型は com.aliyun.odps.data.Struct をそれぞれマッピングします。
  • UDAF では @Resolve アノテーションを使用して署名が取得されます。Maxcompute2.0 ではアノテーションで新しいでータ型を使用できます。たとえば、 @Resolve("smallint-> varchar (10 )") と記述します。
    注

    • com.aliyun.odps.data.Struct ではリフレクションからフィールド名とフィールドタイプが認識されないので、@Resolveアノテーションで補う必要があります。 つまり、UDF で Struct を使用するには、@Resolve アノテーションを UDF クラスに追加します。 このアノテーションは、パラメーターのオーバーロードまたは com.aliyun.odps.data.Struct を含む戻り値にのみ影響します。
    • 現在、UDF クラスには 1 つの @Resolve アノテーションしか指定できません。 したがって、構造体パラメータまたは戻り値を持つ UDF では、オーバーロードは 1 つだけ定義できます。

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

public final class Lower extends UDF { 
  Public String evaluate (string s ){ 
    If (Stream = NULL ){ 
        return null; 
    } 
        return s.toLowerCase(); 
  } 
}

@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 colors;

public abstract class Aggregator implements ContextFunction {
  @Override
  public void setup(ExecutionContext ctx) throws UDFException {                                                                   
  }
  @Override
  public void close() throws UDFException {    
  }
  /**
   * Create an aggregate buffer
   * @return Writable - Aggregate buffer
   */
  abstract public Writable newBuffer();
  /**
   * @param buffer: aggregation buffer
   * @param args: specified parameter to call UDAF in SQL
   * @throws UDFException
   */
  abstract public void iterate(Writable buffer, Writable[] args) throws UDFException;
  /**
   * generate final result
   * @param buffer
   * @return final result of Object UDAF
   * @throws UDFException
   */
  abstract public Writable terminate(Writable buffer) throws UDFException;
  abstract public void merge(Writable buffer, Writable partial) throws UDFException;
}

• 第一段階で、各ワーカーはスライス内のデータ量と合計をカウントします。 中間結果として、各スライスのデータ量と合計を考慮することができます。

• 第二段階では、ワーカーが第一段階で生成された各スライスの情報を集めます。 最終出力では、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 org.alidata.odps.udtf.examples;
import com.aliyun.odps.udf.UDTF;
import com.aliyun.odps.udf.UDTFCollector;
import com.aliyun.odps.udf.annotation.Resolve;
import com.aliyun.odps.udf.UDFException;
// TODO define input and output types, e.g., "string,string->string,bigint".
   @Resolve("string,bigint->string,bigint")
   public class MyUDTF extends UDTF {
     @Override
     public void process(Object[] args) throws UDFException {
       String a = (String) args[0];
       Long b = (Long) args[1];
       for (String t: a.split("\\s+")) {
         forward(t, b);
       }
     }
   }

select user_udtf(col0, col1) as (c0, c1) from my_table;

+ ------ + ------ +
| col0 | col1 |
+ ------ + ------ +
| A B | 1 |
| C D | 2 |
+ ------ + ------ +

+----+----+
| c0 | c1 |
+----+----+
| A | 1 |
| B | 1 |
| C | 2 |
| D | 2 |
+----+----+

select user_udtf(col0, col1) as (c0, c1) from my_table; 
select user_udtf(col0, col1, col2) as (c0, c1) from (select * from my_table distribute by key sort by key) t;
select reduce_udtf(col0, col1, col2) as (c0, c1) from (select col0, col1, col2 from (select map_udtf(a0, a1, a2, a3) as (col0, col1, col2) from my_table) t1 distribute by col0 sort by col0, col1) t2;

• 同じ SELECT 句内では、他の式を使用することはできません。

  select value, user_udtf(key) as mycol ...

• UDTF はネストできません。

  select user_udtf1(user_udtf2(key)) as mycol...

• 同じ SELECT 句で group by、distribute by、sort by と一緒に使用することはできません。

  select user_udtf(key) as mycol ... group by mycol

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. MaxCompute で、リソーステーブル "table_resource1"、"table_resource2"、および物理テーブル "tmp1" を作成します。 対応するデータをテーブルに挿入します。
5. この 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 |
   +-------+------------+-------+

@Resolve("struct<a:bigint>,string->string")
public class UdfArray extends UDF {
  public String evaluate(List<String> vals, Long len) {
    return vals.get(len.intValue());
  }
  public String evaluate(Map<String,String> 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 colors;

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;
--関数の作成
CREATE FUNCTION hive_collect as 'com.aliyun.odps.compiler.hive.Collect' using 'test.jar';
--関数の使用
set odps.sql.hive.compatible=true;
select hive_collect(4y,5y,6y) from dual;
+------+
| _c0 |
+------+
| [4, 5, 6] |
+------+

• MaxCompute の add jar コマンドにより、プロジェクト内にリソースが恒久的に作成されます。また、UDF の作成時に jar を指定しますが、すべての jar を自動的にクラスパスに追加することはできません。
• 互換性のある Hive UDF を使用するには、 set odps.sql.hive.compatible = true; を SQL 文の前に追加し、SQL 文と一緒に送信します。
• 互換性のある Hive UDF を使用する場合は、 MaxCompute のJAVA サンドボックスの制限にご注意ください。