Hadoop と Oracle Cloud Infrastructure HDFS Connector を使って Object Storage をファイルシステムのように扱う

はじめに

Hadoop の主要コンポーネントに HDFS（Hadoop Distributed File System）という仮想的な分散ファイル・システムがあって、これは Hadoop エコシステムの基本となる入出力インターフェースとなっています。そして、Oracle は Object Storage を HDFS のように扱うための HDFS Connector を提供していて、これを使うと Hadoop や Spark から HDFSと同じ方式で Object Storage のファイルにアクセスすることができます（oci://~ というプレフィクスをつけて読み書きできます）。
私も最初はこの HDFS Connector を Spark アプリケーションのファイル入出力のために使っていたのですが、ある時 hadoop fs コマンドを使うとかなり便利に Object Storage の操作ができることに気づきました。特に Oracle Cloud Infrastructure (OCI) の Object Storage はネームスペース内でバケットおよびオブジェクトはフラットに存在する一方で、1つ以上のスラッシュ(/)を含む接頭辞文字列をオブジェクト名に追加することで、ディレクトリ構造をシミュレートできるという面白い特長を備えています。この仮想的なディレクトリ構造をストレスなく扱うことができるのが hadoop fs コマンドなのです。
ここでは、Hadoop を使って Object Storage の「ディレクトリ＆ファイル操作」を色々と試してみようと思います。

セットアップ

事前準備

今回は Hadoop 3.3.1 を使いますので、Java 8 か Java 11 を予めインストールしておいて下さい。
Java 8 だと余計な warning が出ないです。
OCI CLI も最後に使いますが、インストールしてなければ OCI コンソールから操作しても構いません。

Hadoop & OCI HDFS Connector のインストール

手順としては、Hadoop のバイナリと OCI HDFS Connector のバイナリをダウンロード＆解凍して、OCI HDFS Connector の jar ファイルを Hadoop の lib ディレクトリにコピーします。

Hadoop は https://hadoop.apache.org/releases.html からダウンロードします。
OCI HDFS Connector は https://github.com/oracle/oci-hdfs-connector/releases からダウンロードします。

$ wget --no-check-certificate https://dlcdn.apache.org/hadoop/common/hadoop-3.3.1/hadoop-3.3.1.tar.gz
$ tar xvf hadoop-3.3.1.tar.gz

$ wget https://github.com/oracle/oci-hdfs-connector/releases/download/v3.3.1.0.3.0/oci-hdfs.zip
$ unzip oci-hdfs.zip -d oci-hdfs

ディレクトリはこんな状況。

$ ls
hadoop-3.3.1  hadoop-3.3.1.tar.gz  oci-hdfs  oci-hdfs.zip

OCI HDFS Connector の jar を Hadoop のライブラリ・ディレクトリにコピーします。

$ cp oci-hdfs/lib/*.jar oci-hdfs/third-party/lib/*.jar hadoop-3.3.1/share/hadoop/hdfs/lib/

インストール作業は以上、Hadoop ディレクトリ以外は削除して構いません。
では、次に Hadoop の設定を行います。

Hadoop の設定

OCI HDFS Connector を認識させるための設定を行います。まずはベーシックにユーザ認証方式 (User Principal) の設定を行ってみます。

HADOOP_DIR/etc/hadoop/core-site.xml を編集します。

$ vi hadoop-3.3.1/etc/hadoop/core-site.xml

core-site.xml の設定内容は、~/.oci/config と同じです(こちらを参照)。fs.oci.client.hostname プロパティで接続先のリージョンを指定しています。

<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
  <property>
    <name>fs.oci.client.hostname</name>
    <value>https://objectstorage.us-ashburn-1.oraclecloud.com</value>
  </property>
  <property>
    <name>fs.oci.client.auth.tenantId</name>
    <value>ocid1.tenancy.oc1..</value>
  </property>
  <property>
    <name>fs.oci.client.auth.userId</name>
    <value>ocid1.user.oc1..</value>
  </property>
  <property>
    <name>fs.oci.client.auth.fingerprint</name>
    <value>xx:xx:xx:xx:xx:xx:xx:xx:xx:xx:..</value>
  </property>
  <property>
    <name>fs.oci.client.auth.pemfilepath</name>
    <value>~/.oci/oci_api_key.pem</value>
  </property>
</configuration>

OCI HDFS Connector は Instance Principal や Resource Principal にも対応していますので、適切なポリシーを設定すれば、ユーザ認証しなくても利用できます。Compute や OKE 上の Pod なら Instance Principal、OCI Functions や OCI Data Science であれば Resource Principal ですね。
以下は、Instance Principal の設定例（繰り返しますが OCI 側でポリシーの設定は必要です）。

<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
  <property>
    <name>fs.oci.client.hostname</name>
    <value>https://objectstorage.us-ashburn-1.oraclecloud.com</value>
  </property>
  <property>
    <name>fs.oci.client.custom.authenticator</name>
    <value>com.oracle.bmc.hdfs.auth.InstancePrincipalsCustomAuthenticator</value>
  </property>
</configuration>

以上で設定完了です。

Hadoop で Object Storage にアクセスする

では、いよいよ実際に Hadoop を使ってみましょう。

$ cd hadoop-3.3.1/bin

$ ./hadoop fs
Usage: hadoop fs [generic options]
        [-appendToFile <localsrc> ... <dst>]
        [-cat [-ignoreCrc] <src> ...]
        [-checksum [-v] <src> ...]
        [-chgrp [-R] GROUP PATH...]
        [-chmod [-R] <MODE[,MODE]... | OCTALMODE> PATH...]
        [-chown [-R] [OWNER][:[GROUP]] PATH...]
        [-concat <target path> <src path> <src path> ...]
        [-copyFromLocal [-f] [-p] [-l] [-d] [-t <thread count>] <localsrc> ... <dst>]
        [-copyToLocal [-f] [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]
        [-count [-q] [-h] [-v] [-t [<storage type>]] [-u] [-x] [-e] [-s] <path> ...]
        [-cp [-f] [-p | -p[topax]] [-d] <src> ... <dst>]
        [-createSnapshot <snapshotDir> [<snapshotName>]]
        [-deleteSnapshot <snapshotDir> <snapshotName>]
        [-df [-h] [<path> ...]]
        [-du [-s] [-h] [-v] [-x] <path> ...]
        [-expunge [-immediate] [-fs <path>]]
        [-find <path> ... <expression> ...]
        [-get [-f] [-p] [-ignoreCrc] [-crc] <src> ... <localdst>]
        [-getfacl [-R] <path>]
        [-getfattr [-R] {-n name | -d} [-e en] <path>]
        [-getmerge [-nl] [-skip-empty-file] <src> <localdst>]
        [-head <file>]
        [-help [cmd ...]]
        [-ls [-C] [-d] [-h] [-q] [-R] [-t] [-S] [-r] [-u] [-e] [<path> ...]]
        [-mkdir [-p] <path> ...]
        [-moveFromLocal [-f] [-p] [-l] [-d] <localsrc> ... <dst>]
        [-moveToLocal <src> <localdst>]
        [-mv <src> ... <dst>]
        [-put [-f] [-p] [-l] [-d] [-t <thread count>] <localsrc> ... <dst>]
        [-renameSnapshot <snapshotDir> <oldName> <newName>]
        [-rm [-f] [-r|-R] [-skipTrash] [-safely] <src> ...]
        [-rmdir [--ignore-fail-on-non-empty] <dir> ...]
        [-setfacl [-R] [{-b|-k} {-m|-x <acl_spec>} <path>]|[--set <acl_spec> <path>]]
        [-setfattr {-n name [-v value] | -x name} <path>]
        [-setrep [-R] [-w] <rep> <path> ...]
        [-stat [format] <path> ...]
        [-tail [-f] [-s <sleep interval>] <file>]
        [-test -[defswrz] <path>]
        [-text [-ignoreCrc] <src> ...]
        [-touch [-a] [-m] [-t TIMESTAMP (yyyyMMdd:HHmmss) ] [-c] <path> ...]
        [-touchz <path> ...]
        [-truncate [-w] <length> <path> ...]
        [-usage [cmd ...]]

Generic options supported are:
-conf <configuration file>        specify an application configuration file
-D <property=value>               define a value for a given property
-fs <file:///|hdfs://namenode:port> specify default filesystem URL to use, overrides 'fs.defaultFS' property from configurations.
-jt <local|resourcemanager:port>  specify a ResourceManager
-files <file1,...>                specify a comma-separated list of files to be copied to the map reduce cluster
-libjars <jar1,...>               specify a comma-separated list of jar files to be included in the classpath
-archives <archive1,...>          specify a comma-separated list of archives to be unarchived on the compute machines

The general command line syntax is:
command [genericOptions] [commandOptions]

結構色々なディレクトリ＆ファイル操作ができそうですね。ただし、Hadoop コマンドではバケットを新規作成することはできないので、最初に OCI CLI を使って砂場となるバケットを作っておきましょう。

$ oci os bucket create --name hadoop1
$ oci os bucket create --name hadoop2

hadoop1, hadoop2 という二つのバケットを作成しました。では、このバケットにアクセスする方法ですが、Hadoop からは、バケットのルートディレクトリは、

oci://<バケット>@<オブジェクト・ストレージ・ネームスペース>/

という形式で表現されます。オブジェクト・ストレージ・ネームスペースは OCI コンソール右上の「プロファイル」（人マーク）からテナンシをクリックするとテナンシ詳細画面に遷移しますので、そこで確認することができます。

$ ./hadoop fs -ls -d oci://hadoop1@NAMESPACE/
drwxrwxrwx   -          0 1970-01-01 09:00 oci://hadoop1@NAMESPACE/

$ ./hadoop fs -ls -d oci://hadoop2@NAMESPACE/
drwxrwxrwx   -          0 1970-01-01 09:00 oci://hadoop2@NAMESPACE/

では、色々な操作を行ってみましょう。

# hadoop1バケットに dir1 ディレクトリを作成
$ ./hadoop fs -mkdir oci://hadoop1@NAMESPACE/dir1

# dir1 ディレクトリに インターネットからファイル(README.md)をコピー
$ ./hadoop fs -cp https://raw.githubusercontent.com/oracle/oci-hdfs-connector/master/README.md oci://hadoop1@NAMESPACE/dir1

# hadoop2バケットに dir2/dir3 ディレクトリ（階層）を作成
$ ./hadoop fs -mkdir -p oci://hadoop2@NAMESPACE/dir2/dir3

# hadoop1バケットのdir1ディレクトリ配下を丸ごとhadoop2バケットにコピー
$ ./hadoop fs -cp oci://hadoop1@NAMESPACE/dir1 oci://hadoop2@NAMESPACE/dir2/dir3

# ファイルを確認
$ ./hadoop fs -ls oci://hadoop2@NAMESPACE/dir2/dir3/dir1
Found 1 items
-rw-rw-rw-   1       4043 2022-01-04 18:24 oci://hadoop2@NAMESPACE/dir2/dir3/dir1/README.md

# ローカル・ディレクトリにコピー
$ ./hadoop fs -copyToLocal oci://hadoop2@NAMESPACE/dir2/dir3/dir1/README.md

OCIコンソールから見ると、hadoop2 バケットはこのようになっています。

リージョンを跨いだ操作

勘の良い方なら「でも core-site.xml で指定したエンドポイントは1つだけだったので、特定のリージョンに閉じた操作しかできないのは？」と疑問に思うでしょう。
リージョンを跨いだ操作は、バケットに対応したエンドポイントを core-site.xml 内であらかじめ指定しておくことで実現できます。
東京リージョンの nrt_bucket バケットと 大阪リージョンの kix_bucket バケットのエンドポイントを各々設定してみます。

まずは、各々のリージョンにバケットを作成して

$ oci os bucket create --name nrt_bucket --region ap-tokyo-1
$ oci os bucket create --name kix_bucket --region ap-osaka-1

core-site.xml に設定を追加します。

  <property>
    <name>fs.oci.client.hostname.nrt_bucket.NAMESPACE</name>
    <value>https://objectstorage.ap-tokyo-1.oraclecloud.com</value>
  </property>
  <property>
    <name>fs.oci.client.hostname.kix_bucket.NAMESPACE</name>
    <value>https://objectstorage.ap-osaka-1.oraclecloud.com</value>
  </property>

それでは、リージョン間でファイルのコピーを行ってみましょう。

# ファイルをアップロード
$ ./hadoop fs -copyFromLocal README.md oci://nrt_bucket@NAMESPACE/

# リージョン間でファイルをコピー
$ ./hadoop fs -cp oci://nrt_bucket@NAMESPACE/README.md oci://kix_bucket@NAMESPACE/

# 確認
$ ./hadoop fs -ls oci://kix_bucket@NAMESPACE/
Found 1 items
-rw-rw-rw-   1       1034 2022-01-06 16:31 oci://kix_bucket@NAMESPACE/README.md

東京リージョンのバケットから大阪リージョンのバケットにファイルのコピーができました。

まとめ

Object Storage のバケットを階層型のファイルシステムのように扱えると色々と便利ですよね。私自身も、バケット内の特定の "仮想" ディレクトリ単位でファイルの管理ができるようになるので、無駄にバケットを横並びに作ることが無くなりました。それと、ちょっとしたObject storageのバックアップ作業にも hadoop fs コマンドは使えると思います（大量コピーであれば hadoop distcp です）。Hadoop って使ったことが無い方が結構多いと思いますが、是非この機会に試してみて下さい。

参考

• OCI HDFS Connector ドキュメント
  https://docs.oracle.com/ja-jp/iaas/Content/API/SDKDocs/hdfsconnector.htm