HDInsightのSparkとHive LLAPをHive Warehouse Connecter (HWC) で統合する

## はじめに
HDInsight 4.0以降、具体的にはApache Spark 2.3.1とApache Hive 3.1.0が異なるmetastoreを使うようになったため、これまでのようにmetastoreを共有して運用することが出来なくなったが、HortonworksがGitHub上で公開している [Hive Warehouse Connector (HWC)](https://github.com/hortonworks/hive-warehouse-connector) を使うことで、SparkとHiveのmetastoreを一緒に使用することができるようになる。
HWCライブラリでLLAPデーモンからSpark Executorにデータが並列ロードされるので、JDBC接続よりもスケーラブルに動作するとの事。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/144789/ba64c649-7658-573a-ab67-e3c4d3d71261.png)
(From: https://docs.hortonworks.com/HDPDocuments/HDP3/HDP-3.1.0/developing-spark-applications/content/using_spark_hive_warehouse_connector.html)

HWCを通してできることは下記の通り

- テーブルを記述する (DESCRIBE)
- ORCフォーマットデータ用のテーブルを作成する
- HiveテーブルのデータをSELECTしてDataFrameに読み込む
- DataFrameをHiveテーブルにバッチで書き込む
- HiveのUPDATEステートメントを実行する
- Hiveからテーブルデータを読み取り、Sparkで変換し、新しいHiveテーブルに書き込む
- HiveStreamingを使用してDataFrameまたはSpark StreamをHiveに書き込む

制限事項としては下記の通り

- ORCフォーマットのみサポート
- Spark thrift serverはサポートされていない

## HWCの設定
### 1. 事前準備
今回動作をテストするために新規にHDInsight Spark 4.0とHDInsight Interactive Query (LLAP) 4.0の2つのクラスターを作成した。この2つのクラスターは同じ仮想ネットワークに所属し、同じStorage Accountを使用するように設定した。デプロイの際の注意事項としては以下の2つが挙げられる。

- 同じVNETに作成するときはクラスター名の頭文字が6文字被っているとコケる
- 同じストレージアカウントでデプロイするとき、同じコンテナーは指定できない

### 2. 名前解決の設定
まず、LLAPクラスタとSparkクラスタのHeadNodeにそれぞれSSHログインし、LLAPクラスターのheadnode0の/etc/hostsのファイルの内容を、Sparkクラスターのheadnode0の/etc/hostsにコピーする。これによってSparkクラスターがLLAPクラスターのIPを解決できるようになる。以下のスクリーンショットのような感じ。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/144789/5ae5844b-658b-5fa4-9595-2edf1a43e25e.png)

### 3. 各種パラメータの設定
AzureのポータルのHDInsightのSparkクラスターの管理ブレードから[Cluster dashboard]を選択し、Ambariにアクセス。Ambari画面上で、`[Spark2] > [CONFIGS] > [Custom spark2-default]`にアクセス。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/144789/8f69946d-e994-3f29-d068-37984fb160bc.png)
[Custom spark2-default]セクションの下部に[Add Property...]リンクがあるのでこれをクリックして設定画面を開く。

`"spark.hadoop.hive.llap.daemon.service.hosts"`に"LLAP App name"を設定する。設定する値は`[Advanced hive-interactive-site] > "hive.llap.daemon.service.hosts"`からコピーする。たぶん通常は"@llap0"になってると思う。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/144789/b9d85e60-49d1-0a8e-659e-1e2460509ab5.png)

このようにして下記のパラメータも設定する。

|パラメータ|値|補足|
|---|---|---|
|spark.sql.hive.hiveserver2.jdbc.url|LLAPのJDBC接続文字列|(例) jdbc:hive2://LLAPCLUSTERNAME.azurehdinsight.net:443/;user=admin;password=PWD;ssl=true;transportMode=http;httpPath=/hive2 <br> ※LLAPCLUSTERNAME"は自分のLLAPのクラスター名に置き換える|
|spark.datasource.hive.warehouse.load.staging.dir|ストレージアカウント内のステージングディレクトリを指定|(例) wasb://STORAGE_CONTAINER_NAME@STORAGE_ACCOUNT_NAME.blob.core.windows.net/tmp <br> ※STRAGE_CONTAINER, STORAGE_ACCOUNT_NAMEの値は自分の環境に合わせて置き換える|
|spark.datasource.hive.warehouse.metastoreUri|Ambariの[Hive] > [ADVANCED] > [General] > "hive.metastore.uris"の値をコピー|(例) thrift://hn0-hwclla.0iv2nyrmse1uvp2caa4e34jkmf.cx.internal.cloudapp.net:9083, <br> thrift://hn1-hwclla.0iv2nyrmse1uvp2caa4e34jkmf.cx.internal.cloudapp.net:9083|
|spark.security.credentials.hiveserver2.enabled|false||
|spark.hadoop.hive.zookeeper.quorum|Ambariの[Hive] > [ADVANCED] > [Advanced hive-site] > "hive.zookeeper.quorum"の値をコピー|(例) zk1-nkhvne.0iv2nyrmse1uvp2caa4e34jkmf.cx.internal.cloudapp.net:2181, <br> zk4-nkhvne.0iv2nyrmse1uvp2caa4e34jkmf.cx.internal.cloudapp.net:2181, <br> zk6-nkhvne.0iv2nyrmse1uvp2caa4e34jkmf.cx.internal.cloudapp.net:2181

設定が終わったら[Save]して、サービスをRestartして反映させる。

## HWCを使った接続

### サポートされているクライアントツール
サポートされているクライアントツールは下記の通り。

- spark-shell
- PySpark
- Spark-submit
- Zeppelin
- Livy

### spark-shellの起動

今回はspark-shellを使って接続を試すため、SparkクラスターのHeadNodeにSSHでログインして、以下のようにしてspark-shellを起動する。{VERSION}のところは実際のバージョンに置き換えて実行。

```bash
$ spark-shell --master yarn \
--jars /usr/hdp/current/hive_warehouse_connector/hive-warehouse-connector-assembly-{VERSION}.jar \
--conf spark.security.credentials.hiveserver2.enabled=false
```

そうするといつものようにspark-shellが起動する
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/144789/1b1a3f62-ced2-1845-dff7-c409516d3a8f.png)

### Sparkセッションの作成

最初にHWCを使ったSparkセッションを作成する。

```scala
import com.hortonworks.hwc.HiveWarehouseSession
val hive = HiveWarehouseSession.session(spark).build()
```

### HiveテーブルにSELECTしてDataFrameに読み込む

標準で作成されているhivesampletableにアクセスしてクエリを実行してみる。

```scala
hive.setDatabase("default")
val df = hive.executeQuery("select * from hivesampletable where state = 'Colorado'")
val df = hive.executeQuery("select * from hivesampletable")
df.filter("state = 'Colorado'").show()
```

こんな感じで結果が出てくればOK。

```
+--------+---------+------+--------------+----------+-----------+--------+-------------+--------------+---------+--------------------+
|clientid|querytime|market|deviceplatform|devicemake|devicemodel|   state|      country|querydwelltime|sessionid|sessionpagevieworder|
+--------+---------+------+--------------+----------+-----------+--------+-------------+--------------+---------+--------------------+
|      28| 01:37:50| en-US|       Android|  Motorola|    Droid X|Colorado|United States|    20.3095339|        1|                   1|
|      28| 00:53:31| en-US|       Android|  Motorola|    Droid X|Colorado|United States|    16.2981668|        0|                   0|
|      28| 00:53:50| en-US|       Android|  Motorola|    Droid X|Colorado|United States|     1.7715228|        0|                   1|
|      28| 01:37:19| en-US|       Android|  Motorola|    Droid X|Colorado|United States|    28.9811416|        1|                   0|
|     186| 18:31:39| en-US|       Android|        LG|      VS660|Colorado|United States|     3.1242108|        3|                   6|
|     186| 00:33:42| en-US|       Android|        LG|      VS660|Colorado|United States|      0.407366|        0|                   0|
|     186| 00:33:43| en-US|       Android|        LG|      VS660|Colorado|United States|     2.8991561|        0|                   1|
|     186| 00:33:45| en-US|       Android|        LG|      VS660|Colorado|United States|     0.9904373|        0|                   2|
|     186| 15:53:14| en-US|       Android|        LG|      VS660|Colorado|United States|     0.5970943|        1|                   0|
|     186| 15:53:14| en-US|       Android|        LG|      VS660|Colorado|United States|     1.4811596|        1|                   1|
|     186| 15:53:16| en-US|       Android|        LG|      VS660|Colorado|United States|     2.5451684|        1|                   2|
|     186| 16:46:28| en-US|       Android|        LG|      VS660|Colorado|United States|      0.312012|        2|                   0|
|     186| 16:46:29| en-US|       Android|        LG|      VS660|Colorado|United States|     2.6448547|        2|                   1|
|     186| 16:46:31| en-US|       Android|        LG|      VS660|Colorado|United States|     1.7891041|        2|                   2|
|     186| 16:49:00| en-US|       Android|        LG|      VS660|Colorado|United States|     0.0200286|        2|                   3|
|     186| 16:49:00| en-US|       Android|        LG|      VS660|Colorado|United States|     6.1997895|        2|                   4|
|     186| 16:49:04| en-US|       Android|        LG|      VS660|Colorado|United States|     1.0788553|        2|                   5|
|     186| 18:09:48| en-US|       Android|        LG|      VS660|Colorado|United States|    20.9030322|        3|                   0|
|     186| 18:31:18| en-US|       Android|        LG|      VS660|Colorado|United States|     0.7973807|        3|                   1|
|     186| 18:31:19| en-US|       Android|        LG|      VS660|Colorado|United States|     14.875574|        3|                   2|
+--------+---------+------+--------------+----------+-----------+--------+-------------+--------------+---------+--------------------+
only showing top 20 rows
```

## 参考情報
Hive Warehouse Connector を使用して Apache Spark と Apache Hive を統合する
https://docs.microsoft.com/ja-jp/azure/hdinsight/interactive-query/apache-hive-warehouse-connector
Integrating Apache Hive with Kafka, Spark, and BI
https://docs.hortonworks.com/HDPDocuments/HDP3/HDP-3.1.0/integrating-hive/content/hive_hivewarehouseconnector_for_handling_apache_spark_data.html