MaxCompute:例: UDF で複合データ型を使用する

Java UDF のサンプルコード

package com.aliyun; // パッケージ名を指定します。

import com.aliyun.odps.data.Struct;
import com.aliyun.odps.udf.UDF;
import com.aliyun.odps.udf.annotation.Resolve;

import java.text.SimpleDateFormat;
import java.util.*;

@Resolve("struct<input_name:string, input_timestamp:bigint>->map<string,string>")
public class ComplexDataTypeExample extends UDF{
    private static final String PATTERN = "yyyy-MM-dd HH:mm:ss";

    /**
     * タイムスタンプのリストを時間文字列のリストに変換します。
     * @param timestamps タイムスタンプのリストを入力します。
     * @return 時間文字列のリストを取得します。
     */
    public List<String> evaluate(List<Long> timestamps) {
        if (timestamps == null) {
            return null;
        }
        List<String> result = new ArrayList<>();
        SimpleDateFormat formatter = new SimpleDateFormat(PATTERN);
        for (Long timestamp : timestamps) {
            Date date = new Date(timestamp < 9999999999L ? timestamp * 1000 : timestamp);
            String dateString = formatter.format(date);
            result.add(dateString);
        }
        return result;
    }

    /**
     * MAP データ型のタイムスタンプを MAP データ型の時間文字列に変換します。
     * @param timestamps 値がタイムスタンプである MAP データ型を入力します。
     * @return MAP データ型の時間文字列のリストを取得します。
     */
    public Map<String, String> evaluate(Map<String, Long> timestamps) {
        if (timestamps == null) {
            return null;
        }
        Map<String, String> result = new HashMap<>(timestamps.size());
        SimpleDateFormat formatter = new SimpleDateFormat(PATTERN);
        for (String key : timestamps.keySet()) {
            Long timestamp = timestamps.get(key);
            Date date = new Date(timestamp < 9999999999L ? timestamp * 1000 : timestamp);
            String dateString = formatter.format(date);
            result.put(key, dateString);
        }
        return result;
    }

    /**
     * タイムスタンプを時間文字列に変換します。
     * @param input STRUCT データ型のタイムスタンプを入力します。
     * @return STRUCT データ型の時間文字列を取得します。
     */
    public Map<String, String> evaluate(Struct input) {
        if (input == null) {
            return null;
        }
        SimpleDateFormat formatter = new SimpleDateFormat(PATTERN);
        String nameValue = (String) input.getFieldValue("input_name");
        Long timestampValue = (Long) input.getFieldValue("input_timestamp");
        Date date = new Date(timestampValue < 9999999999L ? timestampValue * 1000 : timestampValue);
        String dateString = formatter.format(date);
        Map<String, String> result = new HashMap<>(8);
        result.put("output_name", nameValue);
        result.put("output_time", dateString);
        return result;
    }
}

次の依存関係を pom.xml に導入する必要があります。

<dependency>
  <groupId>com.aliyun.odps</groupId>
  <artifactId>odps-sdk-udf</artifactId>
  <version>0.29.10-public</version>
</dependency>

サンプルコードでは、UDF のオーバーロード用に 3 つの evaluate メソッドが定義されています。

• メソッド 1: java.util.List クラスに対応するパラメーターとして ARRAY を使用します。

• メソッド 2: java.util.Map クラスに対応するパラメーターとして MAP を使用します。

• メソッド 3: com.aliyun.odps.data.Struct クラスに対応するパラメーターとして STRUCT を使用します。

  説明

  com.aliyun.odps.data.Struct クラスの reflection 機能を使用して、フィールドの名前とデータ型を取得することはできません。 UDF に STRUCT データ型を使用する場合、@Resolve アノテーション を com.aliyun.odps.data.Struct クラスに追加する必要があります。 このアノテーションは、入力パラメーターまたは戻り値に com.aliyun.odps.data.Struct クラスが含まれる UDF のオーバーロードにのみ影響します。

Java で UDF を記述する場合は、UDF クラスを継承する必要があります。 この例では、evaluate メソッドは LIST、MAP、および STRUCT 型の 3 つの入力パラメーターを定義し、それぞれ LIST、MAP、および MAP 型の値を返します。 入力パラメーターと戻り値のデータ型は、SQL 文で UDF のシグネチャとして使用されます。 その他のコード仕様と要件の詳細については、「Java UDF」をご参照ください。

Python 3 UDF のサンプルコード

• 次の例では、入力パラメーターは MAP<STRING, BIGINT> データ型です。 サンプルコードは、UDF_COMPLEX_DATA_MAP という名前の UDF を作成します。

  from odps.udf import annotate
  import datetime
  @annotate('map<string,bigint>->map<string,datetime>')
  class MapExample:
      def evaluate(self, input_dict):
          output_dict = dict()
          for key in input_dict:
              value = input_dict[key]
              t = datetime.datetime.fromtimestamp(value)
              output_dict[key] = t
          return output_dict

• 次の例では、入力パラメーターは ARRAY<BIGINT> データ型です。 サンプルコードは、UDF_COMPLEX_DATA_ARRAY という名前の UDF を作成します。

  from odps.udf import annotate
  import datetime
  @annotate('array<bigint>->array<datetime>')
  class ArrayExample:
      def evaluate(self, input_list):
          output_list = list()
          for item in input_list:
              t = datetime.datetime.fromtimestamp(item)
              output_list.append(t)
          return output_list

• 次の例では、入力パラメーターは STRUCT<input_name:string,input_timestamp:bigint> データ型です。 サンプルコードは、UDF_COMPLEX_DATA_STRUCT という名前の UDF を作成します。

  from odps.udf import annotate
  import datetime, collections
  @annotate('struct<input_name:string,input_timestamp:bigint>->struct<output_name:string,output_time:datetime>')
  class StructExample:
      def evaluate(self, input_namedtuple):
          OutputNamedTuple = collections.namedtuple('output_namedtuple', ['output_name', 'output_time'])
          name_val = input_namedtuple.input_name
          time_val = datetime.datetime.fromtimestamp(input_namedtuple.input_timestamp)
          output_namedtuple = OutputNamedTuple(name_val, time_val)
          return output_namedtuple

デフォルトでは、MaxCompute プロジェクトで UDF を実行するために Python 2 が使用されます。 Python 3 で UDF を実行する場合は、セッションレベルで次のコマンドを実行します: set odps.sql.python.version=cp37。 Python 3 UDF 仕様の詳細については、「Python 3 UDF」をご参照ください。