S3Guard とは

Amazon S3 は広く知られている通り、整合性モデルとして以下の特徴をもちます。

新しいオブジェクトの PUT に対する書き込み後の読み取り整合性 (Read-after-write consistency)
上書き PUT および DELETE に対する結果整合性 (eventual consistency)

Hadoop から S3 を DFS として使う場合 S3A などを使う※わけですが、当然このような結果整合性の影響を受けます。
※ちなみに EMR では S3A ではなく EMRFS が推奨されています。

S3Guard とは S3A の拡張で、S3 上のオブジェクトのメタデータを別途保存・活用することで、Hadoop に整合性をもったビューを提供するための新機能です。
※他にもパフォーマンス改善を目的に含みますが、今回の説明では割愛します。

オフィシャルドキュメントはこちら。

S3Guard: Consistency and Metadata Caching for S3A http://hadoop.apache.org/docs/r3.0.0/hadoop-aws/tools/hadoop-aws/s3guard.html

Hadoop と S3 の関係性については imai_factory 先生が去年の Advent Calendar で詳しく解説してくれてますので、そちらをご参照ください。

HadoopとS3 https://qiita.com/imaifactory/items/6866a5c5d8cc363ace12

Hadoop 3 環境の構築

では、早速環境を準備しましょう。
今回は以下の環境に Single Node Cluster を構築します。
- Amazon Linux 2 LTS Candidate AMI 2017.12.0.20171212.2 x86_64 HVM GP2
- c5.xlarge
- IAM ロール/インスタンスプロファイル付与 (DynamoDB/S3 のアクションをすべて許可)

まずは OpenJDK 1.8.0 をインストールして JAVA_HOME を export。

$ sudo yum install java-1.8.0-openjdk
$ export JAVA_HOME=/usr/lib/jvm/jre/

Hadoop 3 GA パッケージをダウンロードして配置。今回は riken さんのミラーを使います。

$ wget http://ftp.riken.jp/net/apache/hadoop/common/hadoop-3.0.0/hadoop-3.0.0.tar.gz
$ tar -zxvf hadoop-3.0.0.tar.gz
$ cd hadoop-3.0.0/

では早速サンプルを動かしてみましょう。正規表現にマッチしたものを表示します。

$ mkdir /tmp/input
$ cp etc/hadoop/*.xml /tmp/input/
$ bin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-3.0.0.jar grep /tmp/input /tmp/output 'dfs[a-z.]+'
$ cat /tmp/output/*
1       dfsadmin

うまく動きました。

S3A

次に、S3A を使って S3 にアクセスしてみます。
S3A を使うには hadoop-aws-3.0.0.jar をクラスパスに含める必要があるので、hadoop-env.sh を以下のように編集します。

etc/hadoop/hadoop-env.sh

export HADOOP_CLASSPATH=$HADOOP_CLASSPATH:share/hadoop/tools/lib/*

これだけですね。では早速 S3A を使って S3 上の入力データを S3 に出力してみます。

$ bin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-3.0.0.jar wordcount s3a://us-east-1.elasticmapreduce.samples/wordcount/data/ s3a://sample-bucket/hadoop3/output_wc/
$ bin/hdfs dfs -cat s3a://sample-bucket/hadoop3/output_wc/*

できました。WordCount の結果がどばーっと出ました。（ここでは長いので省略しています。）

S3Guard

ようやく本題。S3Guard を試してみます。
S3Guard を使うには core-site.xml に設定を追加する必要があります。
S3Guard ではメタデータ保存用のデータストアとして DynamoDB を使用します。S3 を使ってるんだから AWS アカウントをもってるだろうということで、DynamoDB を使うのは自然ですね。
ちなみに、EMR には EMRFS Consistent View という機能があり、そちらでも DynamoDB が使用されています。

etc/hadoop/core-site.xml

<property>
    <name>fs.s3a.metadatastore.impl</name>
    <value>org.apache.hadoop.fs.s3a.s3guard.DynamoDBMetadataStore</value>
</property>

<property>
    <name>fs.s3a.s3guard.ddb.table</name>
    <value>s3guard-table</value>
</property>

<property>
  <name>fs.s3a.s3guard.ddb.region</name>
  <value>us-east-1</value>
</property>

<property>
    <name>fs.s3a.s3guard.ddb.table.create</name>
    <value>true</value>
</property>

それではサンプルジョブを実行します。

$ bin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples-3.0.0.jar wordcount s3a://us-east-1.elasticmapreduce.samples/wordcount/data/ s3a://sample-bucket/hadoop3/output_wc_s3guard/
$ bin/hdfs dfs -cat s3a://sample-bucket/hadoop3/output_wc_s3guard/*

できました。またもや WordCount の結果がどばーっと出ました。（長いので省略します。）

。。。と、S3Guard のあり/なしの影響がよくわからないですよね。
整合性が影響するようなワークロードじゃないと効果のほどはかなりわかりづらいです。
同時に複数のジョブがアドホックに走る状況で、結果整合性を許容できないようなワークロードで初めて生きてくるわけです。

ただ、DynamoDB に実際に保存されたメタデータの状態には興味が出てきますよね。
そこで、ジョブの実行が終わったこの時点のメタデータを見てみます。

まずはテーブルがちゃんとできているか Describe してみましょう。

$ aws dynamodb describe-table --table-name s3guard-table
{
    "Table": {
        "TableArn": "arn:aws:dynamodb:us-east-1:123456789101:table/s3guard-table",
        "AttributeDefinitions": [
            {
                "AttributeName": "child",
                "AttributeType": "S"
            },
            {
                "AttributeName": "parent",
                "AttributeType": "S"
            }
        ],
        "ProvisionedThroughput": {
            "NumberOfDecreasesToday": 0,
            "WriteCapacityUnits": 100,
            "ReadCapacityUnits": 500
        },
        "TableSizeBytes": 0,
        "TableName": "s3guard-table",
        "TableStatus": "ACTIVE",
        "TableId": "a0227da8-6f98-40b7-b973-67e83a206532",
        "KeySchema": [
            {
                "KeyType": "HASH",
                "AttributeName": "parent"
            },
            {
                "KeyType": "RANGE",
                "AttributeName": "child"
            }
        ],
        "ItemCount": 0,
        "CreationDateTime": 1513514910.2
    }
}

ちゃんとできてました。
それでは、このテーブルの中身をスキャンしてみましょう。
（出力がかなり長いので興味がない方は読み飛ばしてください。）

$ aws dynamodb scan --table-name s3guard-table
{
    "Count": 24,
    "Items": [
        {
            "is_dir": {
                "BOOL": true
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/us-east-1.elasticmapreduce.samples"
            },
            "child": {
                "S": "wordcount"
            }
        },
        {
            "mod_time": {
                "N": "1513514946338"
            },
            "is_deleted": {
                "BOOL": true
            },
            "parent": {
                "S": "/sample-bucket/hadoop3/output_wc_s3guard/_temporary"
            },
            "child": {
                "S": "0"
            },
            "file_length": {
                "N": "0"
            },
            "block_size": {
                "N": "0"
            }
        },
        {
            "mod_time": {
                "N": "1513514944878"
            },
            "is_deleted": {
                "BOOL": true
            },
            "parent": {
                "S": "/sample-bucket/hadoop3/output_wc_s3guard/_temporary/0/_temporary/attempt_local1505268872_0001_r_000000_0"
            },
            "child": {
                "S": "part-r-00000"
            },
            "file_length": {
                "N": "0"
            },
            "block_size": {
                "N": "0"
            }
        },
        {
            "is_dir": {
                "BOOL": true
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/us-east-1.elasticmapreduce.samples/wordcount"
            },
            "child": {
                "S": "data"
            }
        },
        {
            "mod_time": {
                "N": "1513514948217"
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/sample-bucket/hadoop3/output_wc_s3guard"
            },
            "child": {
                "S": "_SUCCESS"
            },
            "file_length": {
                "N": "0"
            },
            "block_size": {
                "N": "33554432"
            }
        },
        {
            "mod_time": {
                "N": "1513514946328"
            },
            "is_deleted": {
                "BOOL": true
            },
            "parent": {
                "S": "/sample-bucket/hadoop3/output_wc_s3guard"
            },
            "child": {
                "S": "_temporary"
            },
            "file_length": {
                "N": "0"
            },
            "block_size": {
                "N": "0"
            }
        },
        {
            "mod_time": {
                "N": "1513514944034"
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/sample-bucket/hadoop3/output_wc_s3guard"
            },
            "child": {
                "S": "part-r-00000"
            },
            "file_length": {
                "N": "733216"
            },
            "block_size": {
                "N": "33554432"
            }
        },
        {
            "mod_time": {
                "N": "1513514946347"
            },
            "is_deleted": {
                "BOOL": true
            },
            "parent": {
                "S": "/sample-bucket/hadoop3/output_wc_s3guard/_temporary/0"
            },
            "child": {
                "S": "_temporary"
            },
            "file_length": {
                "N": "0"
            },
            "block_size": {
                "N": "0"
            }
        },
        {
            "mod_time": {
                "N": "1513514946357"
            },
            "is_deleted": {
                "BOOL": true
            },
            "parent": {
                "S": "/sample-bucket/hadoop3/output_wc_s3guard/_temporary/0/_temporary"
            },
            "child": {
                "S": "attempt_local1505268872_0001_r_000000_0"
            },
            "file_length": {
                "N": "0"
            },
            "block_size": {
                "N": "0"
            }
        },
        {
            "mod_time": {
                "N": "1406157796000"
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/us-east-1.elasticmapreduce.samples/wordcount/data"
            },
            "child": {
                "S": "0001"
            },
            "file_length": {
                "N": "2392524"
            },
            "block_size": {
                "N": "33554432"
            }
        },
        {
            "mod_time": {
                "N": "1406157796000"
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/us-east-1.elasticmapreduce.samples/wordcount/data"
            },
            "child": {
                "S": "0002"
            },
            "file_length": {
                "N": "2396618"
            },
            "block_size": {
                "N": "33554432"
            }
        },
        {
            "mod_time": {
                "N": "1406157796000"
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/us-east-1.elasticmapreduce.samples/wordcount/data"
            },
            "child": {
                "S": "0003"
            },
            "file_length": {
                "N": "1593915"
            },
            "block_size": {
                "N": "33554432"
            }
        },
        {
            "mod_time": {
                "N": "1406157796000"
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/us-east-1.elasticmapreduce.samples/wordcount/data"
            },
            "child": {
                "S": "0004"
            },
            "file_length": {
                "N": "1720885"
            },
            "block_size": {
                "N": "33554432"
            }
        },
        {
            "mod_time": {
                "N": "1406157796000"
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/us-east-1.elasticmapreduce.samples/wordcount/data"
            },
            "child": {
                "S": "0005"
            },
            "file_length": {
                "N": "2216895"
            },
            "block_size": {
                "N": "33554432"
            }
        },
        {
            "mod_time": {
                "N": "1406157797000"
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/us-east-1.elasticmapreduce.samples/wordcount/data"
            },
            "child": {
                "S": "0006"
            },
            "file_length": {
                "N": "1906322"
            },
            "block_size": {
                "N": "33554432"
            }
        },
        {
            "mod_time": {
                "N": "1406157798000"
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/us-east-1.elasticmapreduce.samples/wordcount/data"
            },
            "child": {
                "S": "0007"
            },
            "file_length": {
                "N": "1930660"
            },
            "block_size": {
                "N": "33554432"
            }
        },
        {
            "mod_time": {
                "N": "1406157798000"
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/us-east-1.elasticmapreduce.samples/wordcount/data"
            },
            "child": {
                "S": "0008"
            },
            "file_length": {
                "N": "1913444"
            },
            "block_size": {
                "N": "33554432"
            }
        },
        {
            "mod_time": {
                "N": "1406157798000"
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/us-east-1.elasticmapreduce.samples/wordcount/data"
            },
            "child": {
                "S": "0009"
            },
            "file_length": {
                "N": "2707527"
            },
            "block_size": {
                "N": "33554432"
            }
        },
        {
            "mod_time": {
                "N": "1406157798000"
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/us-east-1.elasticmapreduce.samples/wordcount/data"
            },
            "child": {
                "S": "0010"
            },
            "file_length": {
                "N": "327050"
            },
            "block_size": {
                "N": "33554432"
            }
        },
        {
            "mod_time": {
                "N": "1406157798000"
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/us-east-1.elasticmapreduce.samples/wordcount/data"
            },
            "child": {
                "S": "0011"
            },
            "file_length": {
                "N": "8"
            },
            "block_size": {
                "N": "33554432"
            }
        },
        {
            "mod_time": {
                "N": "1406157798000"
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/us-east-1.elasticmapreduce.samples/wordcount/data"
            },
            "child": {
                "S": "0012"
            },
            "file_length": {
                "N": "8"
            },
            "block_size": {
                "N": "33554432"
            }
        },
        {
            "is_dir": {
                "BOOL": true
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/sample-bucket"
            },
            "child": {
                "S": "hadoop3"
            }
        },
        {
            "table_created": {
                "N": "1513514920327"
            },
            "table_version": {
                "N": "100"
            },
            "parent": {
                "S": "../VERSION"
            },
            "child": {
                "S": "../VERSION"
            }
        },
        {
            "is_dir": {
                "BOOL": true
            },
            "is_deleted": {
                "BOOL": false
            },
            "parent": {
                "S": "/sample-bucket/hadoop3"
            },
            "child": {
                "S": "output_wc_s3guard"
            }
        }
    ],
    "ScannedCount": 24,
    "ConsumedCapacity": null
}

入出力の両方に関係する S3 オブジェクトのメタデータが格納されていることが確認できます。

おわりに

今回は Hadoop 3 GA 記念ということで S3Guard を触ってみました。
Hadoop 3 には他にも面白い機能がたくさんありますので、どんどん試して使っていきたいですね。