[Oracle Cloud] Apache Spark から HDFS Connector for Object Storage を使ってみた

# はじめに

Apache Spark などのビッグデータ処理プラットフォームを使うとき、データの保管場所は考慮点がいくつかあります。ローカルのディスクで HDFS を作ってアクセスしても良いのですが、スケーラビリティを確保するためにはノードのスケールアウトをしていく必要があり、運用性やコストが気になります。また、同一のデータを複数の Spark クラスタで参照したいときに、膨大なデータのコピーが必要になります。

これらを解消するために、コンピューティングリソースと、ストレージリソースを分離する考え方があります。Apache Spark から、Object Storage にストレージデータを参照する仕組みにすることで、低コスト・無限のスケーラビリティ・データ参照の柔軟性が手に入ります。

Oracle Cloud Infrastructure(以下OCI)では、Object Storage という名前のサービスを提供しています（そのまんまの名前ですね）。今回の記事では、Apache Spark から Object Storage へデータを書き込み・読み込みをする手順を確認していきます。

HDFS Connector for Object Storage という HDFS Connector をOracleが提供しており、これを使って Apache Spark からアクセスをします。



# 前提条件

[以前の記事](https://qiita.com/sugimount/items/8f44808409d1518b8424)で、CentOS 上に Apache Spark をインストールしました。この環境を使っていきます。



# JVM の DNS ルックアップ TTL 設定

Apache Spark は、Java で動作しています。Java の JVM はDNSの名前解決を一定期間キャッシュする挙動をします。デフォルトの設定では `-1`で設定されていることが多く、永久にキャッシュします。Object Storage の名前解決先は、あるタイミングへ変更してしまうこともあるので、キャッシュする時間を設定します。

環境によって変わりますが、yum で OpenJDK を Install した場合は、以下の場所のファイルを編集します

```
sudo vim /usr/lib/jvm/java-1.8.0-openjdk-1.8.0.242.b08-0.el7_7.x86_64/jre/lib/security/java.security
```



編集します

```
省略
#networkaddress.cache.ttl=-1
networkaddress.cache.ttl=60
省略
```





# HDFS Connector for Object Storage Install

次の GitHub にリリースタグで公開されている zip ファイルをダウンロードします。まずは、GitHub にアクセスします。
https://github.com/oracle/oci-hdfs-connector/releases



URL をコピーします。

![1587395747915.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/ffe18118-99e5-4f8b-befa-c39bc4b36f28.png)



Spark が稼働している CentOS にダウンロードします

```
mkdir ~/oci-hdfs-connector
cd ~/oci-hdfs-connector
wget https://github.com/oracle/oci-hdfs-connector/releases/download/v2.9.2.4/oci-hdfs.zip
```



zip を解凍します

```
unzip oci-hdfs.zip
```



以下のディレクトリ構造となっています。

```
[opc@spark01-63591 oci-hdfs-connector]$ tree /home/opc/oci-hdfs-connector
/home/opc/oci-hdfs-connector
|-- CHANGELOG.md
|-- LICENSE.txt
|-- Thirdpartyreadme.txt
|-- apidocs
|   |-- allclasses-frame.html
|   |-- allclasses-noframe.html
|   |-- com
|   |   `-- oracle
|   |       `-- bmc
|   |           `-- hdfs
|   |               |-- Bmc.html
|   |               |-- BmcConstants.Deprecated.html
|   |               |-- BmcConstants.html
|   |               |-- BmcFilesystem.html
|   |               |-- BmcProperties.html
|   |               |-- DeprecatedBmcFilesystem.html
|   |               |-- auth
|   |               |   |-- InstancePrincipalsCustomAuthenticator.html
|   |               |   |-- class-use
|   |               |   |   `-- InstancePrincipalsCustomAuthenticator.html
|   |               |   |-- package-frame.html
|   |               |   |-- package-summary.html
|   |               |   |-- package-tree.html
|   |               |   `-- package-use.html
|   |               |-- class-use
|   |               |   |-- Bmc.html
|   |               |   |-- BmcConstants.Deprecated.html
|   |               |   |-- BmcConstants.html
|   |               |   |-- BmcFilesystem.html
|   |               |   |-- BmcProperties.html
|   |               |   `-- DeprecatedBmcFilesystem.html
|   |               |-- package-frame.html
|   |               |-- package-summary.html
|   |               |-- package-tree.html
|   |               `-- package-use.html
|   |-- constant-values.html
|   |-- deprecated-list.html
|   |-- help-doc.html
|   |-- index-all.html
|   |-- index.html
|   |-- overview-frame.html
|   |-- overview-summary.html
|   |-- overview-tree.html
|   |-- package-list
|   |-- script.js
|   `-- stylesheet.css
|-- lib
|   |-- oci-hdfs-full-2.9.2.4-javadoc.jar
|   `-- oci-hdfs-full-2.9.2.4.jar
|-- oci-hdfs.zip
`-- third-party
    `-- lib
        |-- bcpkix-jdk15on-1.60.jar
        |-- bcprov-jdk15on-1.60.jar
        `-- jsr305-3.0.2.jar

11 directories, 44 files
[opc@spark01-63591 oci-hdfs-connector]$ 
```



HDFS Connector for Object Storage から clone したファイルを、Spark 配下の jar ファイル格納先にコピーします。Spark が Object Storage にアクセスするために使います。

```
cp -p ~/oci-hdfs-connector/lib/oci-hdfs-full-2.9.2.4.jar ~/sparkinstall/spark-2.4.4-bin-hadoop2.7/jars
cp -p ~/oci-hdfs-connector/third-party/lib/bcpkix-jdk15on-1.60.jar ~/sparkinstall/spark-2.4.4-bin-hadoop2.7/jars
cp -p ~/oci-hdfs-connector/third-party/lib/bcprov-jdk15on-1.60.jar ~/sparkinstall/spark-2.4.4-bin-hadoop2.7/jars
cp -p ~/oci-hdfs-connector/third-party/lib/jsr305-3.0.2.jar ~/sparkinstall/spark-2.4.4-bin-hadoop2.7/jars
```



# IAM API キーペアを登録

OCI IAM用 APIキーペアを作成するために、まずは、OCI CLI をインストールします。

```
bash -c "$(curl -L https://raw.githubusercontent.com/oracle/oci-cli/master/scripts/install/install.sh)"
```



以下のコマンドを実行して生成します。

```
oci setup keys
```



自動的に API キー用の秘密鍵・公開鍵が生成されます

```
[opc@spark01-63591 ~]$ ls -la ~/.oci/
total 12
drwx------.  2 opc opc   59 Apr 20 15:35 .
drwx------. 15 opc opc 4096 Apr 20 15:35 ..
-rw-------.  1 opc opc 1679 Apr 20 15:35 oci_api_key.pem
-rw-------.  1 opc opc  451 Apr 20 15:35 oci_api_key_public.pem
```



公開鍵の値をコピーします (以下の値はダミーです)

```
[opc@spark01-63591 ~]$ cat ~/.oci/oci_api_key_public.pem 
-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA1SQ5gGBbSgPR5yIQSOcz
Pf8upJyL7OtsgZ/y0DLif4Qe/usPzZZ4jexY7Y01Rb6tC2AnnAz1x9AWxoUMKzal
BEP8sSx3YpjLzscDhNkb+WPzMsPPjLRa0JIovT5xob1D1yuU62qR3XJHzR0Q2EpV
B9lXRAdhUD5tGNSNEZuL3dnhCOM1d9yhAlEcuuZjrnK1aaQURhKRWxp2/kYAJk2R
/O81gFgbZW7fSpqk7RnHC1GUO4Wfh817Y4vPi57sxkPNcgMtaNoHdE2G5AK/jYDg
MzgDOuQgrJDY/8aXTBnF/hct7gkZX1e6eUd6mqO88ybjDNuChqfpgMQ0jtS8W4Sk
tQIDAQAA
-----END PUBLIC KEY-----
[opc@spark01-63591 ~]$
```



OCI Console で公開鍵を登録します。ログインしたユーザーページに移動します。

![1587397218222.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/bf01b972-9eb5-5cee-51cc-7d0cefa6eb52.png)



Add

![1587397277064.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/1417ca75-2e03-32cf-9119-423c9ceb423b.png)



貼り付けて Add します (値はダミーです。実際には使えません)

![1587397304698.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/73e2e203-ceaf-ddec-c3d5-f3dd25b0da53.png)



# Spark Config ファイル編集

Spark をインストールしたディレクトリに、`core-site.xml` ファイルを作成します。

```
cat <<'EOF' > $SPARK_HOME/conf/core-site.xml
<configuration>
  <property>
    <name>fs.oci.client.hostname</name>
    <value>https://objectstorage.ap-tokyo-1.oraclecloud.com</value>
  </property>
  <property>
    <name>fs.oci.client.auth.tenantId</name>
    <value>ocid1.tenancy.oc1..aaaaaaaaftfptyiomnr5vmsxmrhoqsz7tsgquz3kghjcjy4janm7rfzfaokq</value>
  </property>
  <property>
    <name>fs.oci.client.auth.userId</name>
    <value>ocid1.user.oc1..aaaaaaaaqer2ilvgh5asdibn37gluzrzn4mv2bnrr4pcrke7v6xnw7itmceq</value>
  </property>
  <property>
    <name>fs.oci.client.auth.fingerprint</name>
    <value>a3:cf:03:a2:df:94:c4:c2:59:58:f1:88:9e:4d:86:4a</value>
  </property>
  <property>
    <name>fs.oci.client.auth.pemfilepath</name>
    <value>~/.oci/oci_api_key.pem</value>
  </property>
</configuration>
EOF
```



`spark-defaults.conf` ファイルを作成します。template からコピーします

```
cd ~/sparkinstall/spark-2.4.4-bin-hadoop2.7/conf
cp spark-defaults.conf.template spark-defaults.conf
```



編集

```
vim spark-defaults.conf
```



末尾に追加

```
spark.sql.hive.metastore.sharedPrefixes = shaded.oracle,com.oracle.bmc
```



以上で、Spark 側で準備が出来ました。



# 動作確認 : csvファイルを Object Storage へ書き込み

Spark Shell を起動

```
spark-shell 
```



1, 2, 3, 4, 5 というデータで、Dataset を作ります。

```
val ds = Seq(1, 2, 3, 4, 5).toDS()
```



Dataset から、write.csv API を使用して csv ファイルを書き込みます。`oci://` という文字列を指定することで、OCI Object Storage を書き込み先に出来ます。これは、HDFS Connector for Object Storage をインストールすることで、使える指定先です。

書式は `oci:://<バケット名>@<Namespace名>/<オブジェクト名>` という形です。

```
ds.write.csv("oci://inputbucket@nryjxkqe0mhq/testcsv.csv")
```



Object Storage で書き込みされた結果を確認します。1個の csv ファイルを書き込みすると、Object Storage 上では 6個のオブジェクトで見えています。

![1587752766546.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/318293c8-c396-d254-39a6-cbfe72bd4955.png)



実際の csv ファイルとしては、4つのファイルに分割されています。

- testcsv.csv/part-00000-fba0a510-a442-4bca-9feb-a455eef170f1-c000.csv
- testcsv.csv/part-00001-fba0a510-a442-4bca-9feb-a455eef170f1-c000.csv
- testcsv.csv/part-00002-fba0a510-a442-4bca-9feb-a455eef170f1-c000.csv
- testcsv.csv/part-00003-fba0a510-a442-4bca-9feb-a455eef170f1-c000.csv



中身を確認すると、Dataset の配列が4分割されて csv ファイルとして保存されています。複数の Object で保存することで並列にダウンロード・アップロード出来て、性能が期待できるのではないかなと思います。

```
> cat testcsv.csv_part-00000-fba0a510-a442-4bca-9feb-a455eef170f1-c000.csv 
1
```

```
> cat testcsv.csv_part-00001-fba0a510-a442-4bca-9feb-a455eef170f1-c000.csv 
2
```

```
> cat testcsv.csv_part-00002-fba0a510-a442-4bca-9feb-a455eef170f1-c000.csv 
3
```

```
> cat testcsv.csv_part-00003-fba0a510-a442-4bca-9feb-a455eef170f1-c000.csv 
4
5
```





# 動作確認 : csvファイルを読み込み

適当に csv ファイルを作成します。

```
id,place,rain,day
1,okinawa,129,2018/7/04
2,okinawa,130,2018/7/05
3,gihu,100,2018/7/05
4,koti,101,2018/7/05
5,koti,102,2018/7/05
```



Object Storage へのアップロードします。次のようにアップロード済みです。

![1587753897017.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/b4d323a6-f4a2-b3c2-fcfe-19947aa9b5a4.png)



Spark Shell を起動

```
spark-shell
```



読み込みします。csv ファイルの1行目に列名を書いているので、`option("header", "true")` を入れています。

```
val opendata1 = spark.read.option("header", "true").csv("oci://inputbucket@nryjxkqe0mhq/pre_1h.csv")
```



出力

```
opendata1.show()
```



実行例

```
scala> opendata1.show()
+---+-------+----+---------+
| id|  place|rain|      day|
+---+-------+----+---------+
|  1|okinawa| 129|2018/7/04|
|  2|okinawa| 130|2018/7/05|
|  3|   gihu| 100|2018/7/05|
|  4|   koti| 101|2018/7/05|
|  5|   koti| 102|2018/7/05|
+---+-------+----+---------+


scala>
```



# 動作確認 : 複数csvファイル読み込み

複数のcsvファイルをまとめて Spark に読み込む方法を確認します。Object Storage に、以下の2種類の csvファイルをアップロードします。

multi1.csv

```
rank,name
1,satou
2,suzuki
3,tanaka
```



multi2.csv

```
rank,name
4,sugiyama
5,kato
```



Object Storage へ2つのcsvファイルをアップロードします。

![1587810707077.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/317a24b6-0c5c-3d6e-ae67-5c84a8769fba.png)



Spark-shell を起動して、`oci://inputbucket@nryjxkqe0mhq/multi` と指定して読み込みます。実際の csv ファイル名までは指定していないことがポイントで、このディレクトリ配下の Object が全て読み込まれます。

```
val opendata1 = spark.read.option("header", "true").csv("oci://inputbucket@nryjxkqe0mhq/multi")
```



show() で中身を確認します。

```
opendata1.show()
```



実行例

```
scala> opendata1.show()
+----+--------+
|rank|    name|
+----+--------+
|   1|   satou|
|   2|  suzuki|
|   3|  tanaka|
|   4|sugiyama|
|   5|    kato|
+----+--------+


scala>
```





# 参考URL

https://github.com/oracle/oci-hdfs-connector

https://docs.cloud.oracle.com/en-us/iaas/Content/API/SDKDocs/hdfsconnector.htm

https://www.slideshare.net/KenichiSonoda/oci-object-storage-deep-dive-20190329-ss

https://blogs.oracle.com/bigdata-dataintegration-jp/ocios2spark