DatabricksにおけるAmazon S3の取り扱い

[Amazon S3 \| Databricks on AWS](https://docs.databricks.com/data/data-sources/aws/amazon-s3.html#mount-aws-s3) [2021/9/3時点]の翻訳です。

:::note warn
本書は抄訳であり内容の正確性を保証するものではありません。正確な内容に関しては原文を参照ください。
:::

[Amazon S3](https://aws.amazon.com/s3/)は、大量のテキストやバイナリーデータのような非構造化データを格納するためのサービスです。

本書では、AWS S3バケットに対するDBFS(Databricksファイルシステム)を用いたバケットのマウント、あるいはAPIを用いて直接アクセスする方法を説明します。

> **重要！**
Databricksランタイム7.3 LTS以降では、アップグレードされたバージョンのS3コネクターを使用します。以下の変更が既存コードに影響を及ぼす場合があります。

> - S3Aファイルシステムは`FileSystem.close()`でリソースを解放します。ファイルシステムのキャッシュはデフォルトで有効化されているため、キャッシュされたファイルシステムへの参照を保持する他のスレッドが、クローズされた後に不適切にそのファイルを使用する場合があります。このため、`FileSystem.close()` APIを使用するべきではありません。
- S3Aファイルシステムは出力ストリームをクローズする際にディレクトリのマーカーを削除しません。[HADOOP\-13230](https://issues.apache.org/jira/browse/HADOOP-13230)を含まないHadoopバージョンに依存するレガシーなアプリケーションは、ディレクトリの中にファイルが存在する場合でも空のディレクトリと誤解する場合があります。

# DBFSを通じたS3バケットへのアクセス

このセクションでは、DBFSを通じてどのようにS3バケットにアクセスするのかを説明します。以下のことを実行できます。

- [S3バケットのマウント](#s3バケットのマウント)
- [ローカルファイルとしてS3オブジェクトにアクセス](#ローカルファイルとしてs3オブジェクトにアクセス)
- [S3バケットのアンマウント](#s3バケットのアンマウント)

## S3バケットのマウント

[Databricksファイルシステム(DBFS)](https://qiita.com/taka_yayoi/items/897264c486e179d72247)を通じてS3バケットをマウントすることができます。このマウントはS3ロケーションへのポインターであり、決してデータはローカルでは同期されません。

クラスターを通じてマウントポイントが作成されると、当該クラスターのユーザーはすぐにマウントポイントにアクセスすることができます。別の稼働中のクラスターで、このマウントポイントを使用するには、稼働中のクラスターで新規に作成されたマウントポイントを利用可能にするために`dbutils.fs.refreshMounts()`を実行する必要があります。

S3バケットをマウントするには2つの方法が存在します。

- [AWSインスタンスプロファイルを用いたバケットのマウント](#awsインスタンスプロファイルを用いたバケットのマウント)
- [AWSキーを用いたバケットのマウント](#awsキーを用いたバケットのマウント)

### AWSインスタンスプロファイルを用いたバケットのマウント

AWSの[インスタンスプロファイル](https://qiita.com/taka_yayoi/items/446c7971be354f88c679)を用いてS3バケットに対する認証、承認を管理することができます。インスタンスプロファイルに許可されたアクセス権によってバケットのオブジェクトに対するアクセスのタイプが決定されます。ロールに書き込みアクセスがあるのであれば、マウントポイントのユーザーはバケットにオブジェクトを書き込むことができます。ロールに読み込みアクセスがあるのであれば、マウントポイントのユーザーはバケットのオブジェクトを読み込むことができます。

1. クラスターに対するインスタンスプロファイルを設定します。
1. バケットをマウントします。

    ```py:Python
aws_bucket_name = "<aws-bucket-name>"
mount_name = "<mount-name>"
dbutils.fs.mount("s3a://%s" % aws_bucket_name, "/mnt/%s" % mount_name)
display(dbutils.fs.ls("/mnt/%s" % mount_name))
    ```

    ```scala:Scala
val AwsBucketName = "<aws-bucket-name>"
val MountName = "<mount-name>"
dbutils.fs.mount(s"s3a://$AwsBucketName", s"/mnt/$MountName")
display(dbutils.fs.ls(s"/mnt/$MountName"))
    ```

### AWSキーを用いたバケットのマウント

AWSキーを用いてバケットをマウントすることができます。

> **重要！**
キーを用いてS3バケットをマウントする際、S3バケットの*すべてのオブジェクト*に対して、*すべてのユーザー*が読み書き権限を有することになります。

以下の例では、キーを格納するためにDatabricksの[シークレット](https://qiita.com/taka_yayoi/items/338ef0c5394fe4eb87c0)を使用しています。シークレットのキーは*URLエスケープ*する必要があります。

```py:Python
access_key = dbutils.secrets.get(scope = "aws", key = "aws-access-key")
secret_key = dbutils.secrets.get(scope = "aws", key = "aws-secret-key")
encoded_secret_key = secret_key.replace("/", "%2F")
aws_bucket_name = "<aws-bucket-name>"
mount_name = "<mount-name>"

dbutils.fs.mount("s3a://%s:%s@%s" % (access_key, encoded_secret_key, aws_bucket_name), "/mnt/%s" % mount_name)
display(dbutils.fs.ls("/mnt/%s" % mount_name))
```

```scala:Scala
val AccessKey = dbutils.secrets.get(scope = "aws", key = "aws-access-key")
// Encode the Secret Key as that can contain "/"
val SecretKey = dbutils.secrets.get(scope = "aws", key = "aws-secret-key")
val EncodedSecretKey = SecretKey.replace("/", "%2F")
val AwsBucketName = "<aws-bucket-name>"
val MountName = "<mount-name>"

dbutils.fs.mount(s"s3a://$AccessKey:$EncodedSecretKey@$AwsBucketName", s"/mnt/$MountName")
display(dbutils.fs.ls(s"/mnt/$MountName"))
```
## ローカルファイルとしてS3オブジェクトにアクセス

S3バケットをDBFSにマウントすると、ローカルファイルパスを用いてS3オブジェクトにアクセスすることができます。

```py:Python
df = spark.read.text("/mnt/%s/..." % mount_name)
```

あるいは

```py:Python
df = spark.read.text("dbfs:/mnt/%s/..." % mount_name)
```

```scala:Scala
// scala
val df = spark.read.text(s"/mnt/$MountName/...")
```

あるいは

```scala:Scala
val df = spark.read.text(s"dbfs:/mnt/$MountName/...")
```

## S3バケットのアンマウント

```py:Python
dbutils.fs.unmount("/mnt/mount_name")
```

```scala:Scala
dbutils.fs.unmount(s"/mnt/$MountName")
```

# S3バケットへの直接アクセス

この方法では、AWSキーを用いてSparkのワーカーがS3バケットのオブジェクトに直接アクセスすることができます。キーを格納するためにDatabricksの[シークレット](https://qiita.com/taka_yayoi/items/338ef0c5394fe4eb87c0)を使用しています。

```py:Python
access_key = dbutils.secrets.get(scope = "aws", key = "aws-access-key")
secret_key = dbutils.secrets.get(scope = "aws", key = "aws-secret-key")
sc._jsc.hadoopConfiguration().set("fs.s3a.access.key", access_key)
sc._jsc.hadoopConfiguration().set("fs.s3a.secret.key", secret_key)

# If you are using Auto Loader file notification mode to load files, provide the AWS Region ID.
aws_region = "aws-region-id"
sc._jsc.hadoopConfiguration().set("fs.s3a.endpoint", "s3." + aws_region + ".amazonaws.com")

myRDD = sc.textFile("s3a://%s/.../..." % aws_bucket_name)
myRDD.count()
```

## `s3a://`パスに対するKMS暗号化の設定

### ステップ1: インスタンスプロファイルの設定

Databricksで[インスタンスプロファイル](https://qiita.com/taka_yayoi/items/446c7971be354f88c679)を作成します。

### ステップ2: 設定で指定されたKMSキーにおけるキーユーザーとしてインスタンスプロファイルを追加

1. AWSでIAMサービスに移動します。
1. サイドバーの下部にある**Encryption Keys**をクリックします。
1. アクセス権を追加したいキーをクリックします。
1. Key Usersセクションで**Add**をクリックします。
1. IAMロールの隣にあるチェックボックスを選択します。
1. **Attach**をクリックします。

### ステップ3: 暗号化プロパティのセットアップ

[AWS configurationsセッティング](https://qiita.com/taka_yayoi/items/8d951b660cd87c6c5f18#aws%E3%81%AE%E8%A8%AD%E5%AE%9A)あるいは[initスクリプト](https://docs.databricks.com/clusters/init-scripts.html)を用いて、グローバルKMS暗号化プロパティをセットアップします。お使いのキーのARNを`spark.hadoop.fs.s3a.server-side-encryption.key`キーに設定します。

#### Spark設定

```ini:ini
spark.hadoop.fs.s3a.server-side-encryption.key arn:aws:kms:<region>:<aws-account-id>:key/<bbbbbbbb-ddd-ffff-aaa-bdddddddddd>
spark.hadoop.fs.s3a.server-side-encryption-algorithm SSE-KMS
```

[バケットごと](#バケットごとの設定)にKMS暗号化を設定することもできます。

#### Initスクリプト

initスクリプト`set-kms.sh`を作成するために以下のコードをノートブックで実行し、スクリプトを実行するように[クラスターを設定](https://docs.databricks.com/clusters/init-scripts.html#configure-a-cluster-scoped-init-script)することで、グローバル暗号化設定をセットアップします。

```py:Python
dbutils.fs.put("/databricks/scripts/set-kms.sh", """
#!/bin/bash

cat >/databricks/driver/conf/aes-encrypt-custom-spark-conf.conf <<EOL
[driver] {
  "spark.hadoop.fs.s3a.server-side-encryption.key" = "arn:aws:kms:<region>:<aws-account-id>:key/<bbbbbbbb-ddd-ffff-aaa-bdddddddddd>"
  "spark.hadoop.fs.s3a.server-side-encryption-algorithm" = "SSE-KMS"
}
EOL
""", True)
```

暗号化が動作していることを確認したら、[グローバルinitスクリプト](https://docs.databricks.com/clusters/init-scripts.html#global-init-script)を用いて、すべてのクラスターの暗号化を設定します。

# S3バケットデータの暗号化

Databricksではサーバーサイド暗号化を用いたデータの暗号化をサポートしています。このセクションでは、DBFSを通じてS3にファイルを書き込む際に、どのようにサーバーサイド暗号化を用いるのかを説明します。Databricksでは[Amazon S3\-managed encryption keys \(SSE\-S3\)](https://docs.aws.amazon.com/AmazonS3/latest/dev/UsingServerSideEncryption.html)と[AWS KMS–managed encryption keys \(SSE\-KMS\)](https://docs.aws.amazon.com/AmazonS3/latest/dev/UsingKMSEncryption.html)をサポートしています。

## SSE-S3を用いたファイルの書き込み

1. SSE-S3を用いてS3バケットをマウントするには以下を実行します。

    ```scala:Scala
dbutils.fs.mount(s"s3a://$AccessKey:$SecretKey@$AwsBucketName", s"/mnt/$MountName", "sse-s3")
```

1. SSE-S3を用いて対応するS3バケットにファイルを書き込むには以下を実行します。

    ```scala:Scala
dbutils.fs.put(s"/mnt/$MountName", "<file content>")
```

## SSE-KMSを用いたファイルの書き込み

1. 暗号化タイプとして`sse-kms`あるいは`sse-kms:$KmsKey`を引き渡すことでソースディレクトリをマウントします。
    - デフォルトKMSマスターキーを用いて、SSE-KMSを用いたS3バケットのマウントを行うには以下を実行します。

    ```Scala:Scala
    dbutils.fs.mount(s"s3a://$AccessKey:$SecretKey@$AwsBucketName", s"/mnt/$MountName", "sse-kms")
    ```
    - 特定のKMSキーを用いて、SSE-KMSを用いたS3バケットのマウントを行うには以下を実行します。

    ```Scala:Scala
    dbutils.fs.mount(s"s3a://$AccessKey:$SecretKey@$AwsBucketName", s"/mnt/$MountName", "sse-kms:$KmsKey")
    ```

1. SSE-KMSによる暗号化がされたS3バケットにファイルを書き込むには以下を実行します。

    ```scala:Scala
dbutils.fs.put(s"/mnt/$MountName", "<file content>")
```

# 設定

Databricks Runtime 7.3 LTS以降では、[open\-source Hadoop options](https://hadoop.apache.org/docs/r3.1.3/hadoop-aws/tools/hadoop-aws)を用いたS3Aファイルシステムの設定をサポートしています。グローバルプロパティ、あるいはバケットごとのプロパティを設定することができます。

## グローバル設定

```ini:ini
# Global S3 configuration
spark.hadoop.fs.s3a.aws.credentials.provider <aws-credentials-provider-class>
spark.hadoop.fs.s3a.endpoint <aws-endpoint>
spark.hadoop.fs.s3a.server-side-encryption-algorithm SSE-KMS
```

## バケットごとの設定

`spark.hadoop.fs.s3a.bucket.<bucket-name>.<configuration-key>`シンタックスを用いて、バケットごとのプロパティを設定することができます。これによって、異なるクレディンシャル、エンドポイントなどをバケットに設定することができます。

たとえば、グローバルS3設定に加えて、以下のキーを用いることで、それぞれのバケットに設定を行うことができます。

```ini:ini
# Set up authentication and endpoint for a specific bucket
spark.hadoop.fs.s3a.bucket.<bucket-name>.aws.credentials.provider <aws-credentials-provider-class>
spark.hadoop.fs.s3a.bucket.<bucket-name>.endpoint <aws-endpoint>

# Configure a different KMS encryption key for a specific bucket
spark.hadoop.fs.s3a.bucket.<bucket-name>.server-side-encryption.key <aws-kms-encryption-key>
```

# アクセスリクエスト元によるバケット支払い

[Requester Pays](https://docs.aws.amazon.com/AmazonS3/latest/dev/RequesterPaysBuckets.html)バケットへのアクセスを有効化するには、クラスターの[AWS設定](https://qiita.com/taka_yayoi/items/8d951b660cd87c6c5f18#aws%E3%81%AE%E8%A8%AD%E5%AE%9A)に以下の行を追加してください。

```ini:ini
spark.hadoop.fs.s3a.requester-pays.enabled true
```

> **注意**
DatabricksではRequester Paysバケットに対するDelta Lakeの書き込みはサポートしていません。

### Databricks 無料トライアル

[Databricks 無料トライアル](https://databricks.com/jp/try-databricks)