HiveのGenericUDFを使ってArray,Map,Struct型を処理する

# 概要
HiveのカスタムUDF(User Defined Function)を作る際に継承するクラスをUDFにした場合(extends UDF)、struct型などの複雑な型を処理できません。
今回、struct型などを処理するために**GenericUDF**クラスを継承したカスタムUDFを作成します。

もちろんlateral view explodeなどを使いクエリ中でstructなどを展開すればUDFを使わずに処理できるデータもあります。
めったに使う機会がなく、間違いなく使い方を忘れるので記録しています。

## 環境
* hadoop-3.1.2
* apache-hive-3.1.2

# GenericUDF
今回はGenericUDFを使ってArray,Map,Structがネストしたデータを処理することにします。
GenericUDFを継承する際にはinitialize,evaluate,getDisplayStringという3つのメソッドをoverrideする必要があります。
## pom

```xml:pom.xml
	<build>
		<plugins>
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-compiler-plugin</artifactId>
				<version>3.2</version>
				<configuration>
					<source>1.8</source>
					<target>1.8</target>
				</configuration>
			</plugin>
		</plugins>
	</build>
<dependency>
  <groupId>org.apache.hive</groupId>
  <artifactId>hive-exec</artifactId>
  <version>2.3.6</version>
</dependency>
```

## initialize
* この関数が受け取る型の定義
* 返却する型の定義
* 受け取る型のチェック(必要であれば)
を行います。

### 受け取る型、返却する型
どちらもObjectInspectorのいずれかになります。
例：
* ListObjectInspector(hiveのarrayに使用)
* MapObjectInspector(hiveのmapに使用)
* StructObjectInspector(hiveのstructに使用)
* javaIntObjectInspector(hiveのint)

## evaluate
* 関数に渡される各レコードの値についての処理を記載します。
* ObjectInspectorをreturnします。

## getDisplayString
explainで表示されるテキストを記載します。

# Array,Map,Structの処理サンプル

## Array型の処理
### 引数のarray長を返却するudfを作る

```java:UDFArraySample.java
package hiveudf.sample;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ListObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;


public class UDFArraySample extends GenericUDF {
  private ListObjectInspector listInputObjectInspector;

  public ObjectInspector initialize(ObjectInspector[] args) throws UDFArgumentException {
    if (args.length != 1) {
      throw new UDFArgumentLengthException("just 1 argument can be taken");
    }
    if (args[0].getCategory() != Category.LIST) {
      throw new UDFArgumentException("an argument need to be an array");
    } ;
    // クエリから受け取るオブジェクトを作成
    listInputObjectInspector = (ListObjectInspector) args[0];

    // 今回返却するのは数字である。
    return PrimitiveObjectInspectorFactory.javaIntObjectInspector;
  }

  public Object evaluate(DeferredObject[] args) throws HiveException {

    Object args0 = args[0].get();

    int listLength = listInputObjectInspector.getListLength(args0);
    return listLength;
  }

  @Override
  public String getDisplayString(String[] args) {
    return "explain の結果にここに記載した内容が出力されます。";
  }
}

```

### 挙動確認

```sql
add jar hiveudf-0.0.1-SNAPSHOT.jar;
CREATE TEMPORARY FUNCTION udfArraySample AS 'hiveudf.sample.UDFArraySample';
select udfArraySample(array(1,2,3));
```
#### 結果
3

### explainで表示されるメッセージの確認

```sql
explain select udfArraySample(array(1,2,3));
-- 出力結果の中に以下のような表示が出てきます。
expressions: explain の結果にここに記載した内容が出力されます。
```

## Map型の処理

```java:UDFMapSample.java
package hiveudf.sample;
import java.util.Map;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.MapObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

public class UDFMapSample extends GenericUDF {
  private MapObjectInspector mapOI;

  @Override
  public ObjectInspector initialize(ObjectInspector[] arguments)
  throws UDFArgumentException {
    if (arguments.length != 1) {
      throw new UDFArgumentLengthException("just 1 argument can be taken");
    } else if (!(arguments[0] instanceof MapObjectInspector)) {
      throw new UDFArgumentTypeException(0, "just map can be used");
    }

    mapOI = (MapObjectInspector) arguments[0];
    // 文字列を返す
    return PrimitiveObjectInspectorFactory.javaStringObjectInspector;
  }

  @Override
  public Object evaluate(DeferredObject[] arguments) throws HiveException {
    Object mapObject = arguments[0].get();
    Map<?,?> map = mapOI.getMap(mapObject);

    String returnString = "";
    for(Object v :map.values()){
      returnString = returnString + v + ",";
    }
    return returnString;
  }

  @Override
  public String getDisplayString(String[] args) {
    return "mapのvalueを,区切りで文字列にして返します";
  }
}
```

### 挙動確認

```sql
add jar hiveudf-0.0.1-SNAPSHOT.jar;
CREATE TEMPORARY FUNCTION udfMapSample AS 'hiveudf.sample.UDFMapSample';
select udfMapSample(map("price1",100,"price2",200) );
```

#### 結果
100,200,

## struct型の処理
structのprice1とprice2の和を返す処理

```java:UDFStructSample.java
package hiveudf.sample;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentLengthException;
import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
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.StructField;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.io.IntWritable;

public class UDFStructSample extends GenericUDF {
  private StructObjectInspector structOI;
  private StructField price1Field;
  private StructField price2Field;

  @Override
  public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {
    if (arguments.length != 1) {
      throw new UDFArgumentLengthException("just 1 argument can be taken");
    } else if (!(arguments[0] instanceof StructObjectInspector)) {
      throw new UDFArgumentTypeException(0, "just struct can be used");
    }
    this.structOI = (StructObjectInspector) arguments[0];

    this.price1Field = structOI.getStructFieldRef("price1");
    this.price2Field = structOI.getStructFieldRef("price2");

    // 数字を返す
    return PrimitiveObjectInspectorFactory.javaIntObjectInspector;
  }

  @Override
  public Object evaluate(DeferredObject[] arguments) throws HiveException {

    Object price1 = structOI.getStructFieldData(arguments[0].get(), price1Field);
    Object price2 = structOI.getStructFieldData(arguments[0].get(), price2Field);
    // price1とprice2の和を計算
    int priceSum = ((IntWritable) price1).get() + ((IntWritable) price2).get();
    return priceSum;
  }

  @Override
  public String getDisplayString(String[] children) {
    return "structのprice1とprice2の和を返します。";
  }
}
```

### 挙動確認
```sql
add jar hiveudf-0.0.1-SNAPSHOT.jar;
CREATE TEMPORARY FUNCTION udfStructSample AS 'hiveudf.sample.UDFStructSample';

select udfStructSample(NAMED_STRUCT('city_name','city_a','area',100, 'price1',100,'price2',200));
```
#### 結果
300

## Array,Map,Structの組み合わせ型の処理
Arrayの中にMap,Mapの中にStructの入ったレコードを処理します。
最もpriceの和が大きいproductの名前とpriceの和を**Map**として返します。

```java:UDFArrayStructMapSample.java
public class UDFArrayStructMapSample extends GenericUDF {
  private ListObjectInspector listOI;
  private StructObjectInspector structOI;
  private MapObjectInspector mapOI;
  private StructField productNameField;
  private StructField priceField;

  @Override
  public ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {
    listOI = (ListObjectInspector) arguments[0];
    // listからstructを取得します。
    structOI = (StructObjectInspector) listOI.getListElementObjectInspector();
    productNameField = structOI.getStructFieldRef("product_name");
    priceField = structOI.getStructFieldRef("price");
    // structからmapを取得します。
    mapOI = (MapObjectInspector) priceField.getFieldObjectInspector();

    return ObjectInspectorFactory.getStandardMapObjectInspector(
        PrimitiveObjectInspectorFactory.writableStringObjectInspector,
        PrimitiveObjectInspectorFactory.writableIntObjectInspector);
  }

  @Override
  public Object evaluate(DeferredObject[] arguments) throws HiveException {
    List<?> productStructList = listOI.getList(arguments[0].get());
    int highestPrice = -1;
    String mostExpensiveProductName = null;

    for (Object productStruct : productStructList) {
      Text productNameText = (Text) structOI.getStructFieldData(productStruct, productNameField);
      String productName = productNameText.toString();
      Object structOIFieldData = structOI.getStructFieldData(productStruct, priceField);
      Map<?, ?> priceMap = mapOI.getMap(structOIFieldData);
      int totalPrice = 0;

      // 合計金額を求めます
      for (Object v : priceMap.values()) {
        totalPrice = totalPrice + ((IntWritable) v).get();
      }
      // 合計金額を比較します
      if (totalPrice > highestPrice) {
        mostExpensiveProductName = productName;
        highestPrice = totalPrice;
      }
    }
    // Text,IntWritableを使います。
    Map<Text, IntWritable> mostExpensiveProductMap = new HashMap<>();
    mostExpensiveProductMap.put(new Text(mostExpensiveProductName), new IntWritable(highestPrice));

    return mostExpensiveProductMap;
  }

  @Override
  public String getDisplayString(String[] children) {
    return "最もpriceの和が大きいproductの名前とpriceの和を返します。";
  }
}
```


### 挙動確認
```sql
add jar hiveudf-0.0.1-SNAPSHOT.jar;
CREATE TEMPORARY FUNCTION udfArrayStructMapSample AS 'hiveudf.sample.UDFArrayStructMapSample';

select udfArrayStructMapSample(array(
NAMED_STRUCT('product_name','p1','price',map("price1",100,'price1',200)),
NAMED_STRUCT('product_name','p2','price',map("price1",300,'price2',200))
)
)
;
```
#### 結果
{"p2":500}

# 参考資料
https://blog.dataiku.com/2013/05/01/a-complete-guide-to-writing-hive-udf
https://blog.matthewrathbone.com/2013/08/10/guide-to-writing-hive-udfs.html
https://data-flair.training/blogs/hive-udf/