翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
# Amazon S3 で Spark のパフォーマンスを向上させる
Amazon EMR には、Spark を使用して Amazon S3 に保存されたデータのクエリ、読み取り、および書き込みを行うときのパフォーマンスを最適化するのに役立つ機能が用意されています。
[S3 Select](https://aws.amazon.com/blogs/aws/s3-glacier-select/) では、Amazon S3 に処理を「プッシュダウン」することで一部のアプリケーションの CSV および JSON ファイルのクエリパフォーマンスを向上させることができます。
EMRFS S3 向けに最適化されたコミッターは [OutputCommitter](https://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapreduce/OutputCommitter.html) クラスに代わるものであり、EMRFS のマルチパートアップロード機能を使用して、Spark、DataFrames、および Datasets で Amazon S3 に Parquet ファイルを書き込むときのパフォーマンスを向上させます。
**Topics**
+ [S3 Select と Spark を使用してクエリパフォーマンスを向上させる](emr-spark-s3select.md)
+ [EMR Spark MagicCommitProtocol](emr-spark-magic-commit-protocol.md)
+ [EMRFS S3 向けに最適化されたコミッターを使用する](emr-spark-s3-optimized-committer.md)
+ [EMRFS S3 向けに最適化されたコミットプロトコルを使用する](emr-spark-s3-optimized-commit-protocol.md)
+ [EMRFS で Amazon S3 リクエストを再試行する](emr-spark-emrfs-retry.md)
# S3 Select と Spark を使用してクエリパフォーマンスを向上させる
**重要**
新規のお客様への Amazon S3 Select の提供は終了しました。Amazon S3 Select をご利用の既存のお客様は、今後も通常どおり使用できます。[詳細はこちら](https://aws.amazon.com/blogs/storage/how-to-optimize-querying-your-data-in-amazon-s3/)
Amazon EMR リリース 5.17.0 以降では、Amazon EMR での Spark を使用した [S3 Select](https://aws.amazon.com/blogs/aws/s3-glacier-select/) を使用できます。*S3 Select* では、アプリケーションはオブジェクトに含まれるデータのサブセットのみを取得できます。Amazon EMR では、大量のデータセットをフィルタリングして処理する計算作業をクラスターから Amazon S3 に「プッシュダウン」することにより、一部のアプリケーションのパフォーマンスを高めることができます。また、Amazon EMR と Amazon S3 の間で転送されるデータの量も削減されます。
S3 Select は、`s3selectCSV` および `s3selectJSON` 値を使用してデータ形式を指定する CSV ファイルと JSON ファイルでサポートされます。詳細な説明と例については、[コードで S3 Select を指定する](#emr-spark-s3select-specify) を参照してください。
## S3 Select が使用するアプリケーションに適しているかどうかを確認する
S3 Select が使用するアプリケーションに適しているかどうかを確認するために、S3 Select を使用した場合と使用しない場合のアプリケーションのベンチマークを行うことをお勧めします。
アプリケーションが S3 Select を使用する候補となるかどうかを判断するには、次のガイドラインを使用します。
+ クエリは元のデータセットの半分以上を除外する。
+ Amazon S3 と Amazon EMR クラスター間のネットワーク接続は、転送速度と使用可能な帯域幅が良好です。Amazon S3 は、HTTP 応答を圧縮しないため、応答サイズは圧縮された入力ファイルと比較して増加する可能性があります。
## 考慮事項と制限事項
+ お客様が用意した暗号化キーを使用した Amazon S3 サーバー側の暗号化 (SSE-C) とクライアント側の暗号化はサポートされていません。
+ `AllowQuotedRecordDelimiters` プロパティはサポートされていません。このプロパティを指定した場合、クエリは失敗します。
+ UTF-8 形式の CSV ファイルと JSON ファイルのみがサポートされています。複数行の CSV はサポートされません。
+ 圧縮されていないファイルか gzip ファイルのみがサポートされます。
+ `nanValue`、`positiveInf`、`negativeInf` などの Spark CSV と JSON のオプションや破損した記録に関連するオプション (failfast および dropmalformed モードなど) はサポートされません。
+ 小数でのカンマ (,) の使用はサポートされません。たとえば、`10,000` はサポートされませんが、`10000` はサポートされます。
+ 最後の行のコメント文字はサポートされていません。
+ ファイルの末尾にある空の行は処理されません。
+ 次のフィルターは Amazon S3 にプッシュダウンされません。
+ `COUNT()` や `SUM()` などの集計関数。
+ 属性を `CAST()` するフィルター。例えば、`CAST(stringColumn as INT) = 1`。
+ オブジェクトの属性、または複雑な属性を持つフィルター。例えば、`intArray[1] = 1, objectColumn.objectNumber = 1`。
+ 値がリテラル値ではないフィルター。例: `intColumn1 = intColumn2`
+ 確認された制限のある [S3 Select がサポートするデータ型](https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-glacier-select-sql-reference-data-types.html)のみがサポートされます。
## コードで S3 Select を指定する
次の例では、Scala、SQL、R、および PySpark を使用して CSV 用の S3 Select を指定する方法を示します。JSON 用の S3 Select も同じ方法で使用できます。オプション、デフォルト値、および制限の一覧については、「[オプション](#emr-spark-s3select-specify-options)」を参照してください。
------
#### [ PySpark ]
```
spark
.read
.format("s3selectCSV") // "s3selectJson" for Json
.schema(...) // optional, but recommended
.options(...) // optional
.load("s3://path/to/my/datafiles")
```
------
#### [ R ]
```
read.df("s3://path/to/my/datafiles", "s3selectCSV", schema, header = "true", delimiter = "\t")
```
------
#### [ Scala ]
```
spark
.read
.format("s3selectCSV") // "s3selectJson" for Json
.schema(...) // optional, but recommended
.options(...) // optional. Examples:
// .options(Map("quote" -> "\'", "header" -> "true")) or
// .option("quote", "\'").option("header", "true")
.load("s3://path/to/my/datafiles")
```
------
#### [ SQL ]
```
CREATE TEMPORARY VIEW MyView (number INT, name STRING) USING s3selectCSV OPTIONS (path "s3://path/to/my/datafiles", header "true", delimiter "\t")
```
------
### オプション
`s3selectCSV` と `s3selectJSON` を使用するときは、以下のオプションを利用できます。指定しない場合はデフォルト値が使用されます。
#### S3selectCSV のオプション
| オプション | デフォルト | 使用方法 |
| --- | --- | --- |
| `compression` | `"none"` | 圧縮が使用されているかどうかを示します。`"none"` 以外にサポートされる設定は `"gzip"` のみです。 |
| `delimiter` | "," | フィールドの区切り記号を指定します。 |
| `quote` | `'\"'` | 引用文字を指定します。空の文字例は指定できず、文字列を空にすると XML が不正というエラーが表示されます。 |
| `escape` | `'\\'` | エスケープ文字を指定します。 |
| `header` | `"false"` | `"false"` はヘッダーがないことを指定します。`"true"` はヘッダーが最初の行にあることを指定します。最初の行のヘッダーのみがサポートされ、ヘッダーの前の空の行はサポートされません。 |
| コメント | `"#"` | コメント文字を指定します。コメントインジケーターを無効にすることはできません。つまり、`\u0000` という値はサポートされません。 |
| `nullValue` | "" | |
#### S3selectJSON のオプション
| オプション | デフォルト | 使用方法 |
| --- | --- | --- |
| `compression` | `"none"` | 圧縮が使用されているかどうかを示します。`"none"` 以外にサポートされる設定は `"gzip"` のみです。 |
| `multiline` | "false" | `"false"` は JSON が S3 Select `LINES` 形式であることを指定し、入力データの各行に 1 つの JSON オブジェクトが含まれていることを意味します。`"true"` は JSON が S3 Select `DOCUMENT` 形式であることを指定し、入力データの複数の行に JSON オブジェクトをまたがらせられることを意味します。 |
# EMR Spark MagicCommitProtocol
EMR 6.15.0 以降、S3A ファイルシステムを利用することにより、MagicCommitProtocol は Spark のデフォルトの FileCommitProtocol になります。
## MagicCommitProtocol
MagicCommitProtocol は、[FileCommitProtocol](https://dlcdn.apache.org/spark/docs/2.4.2/api/java/org/apache/spark/internal/io/FileCommitProtocol.html) の代替実装であり、S3A ファイルシステムを使用する時に EMR Spark を使用して Amazon S3 にファイルを書き込むように最適化されています。このプロトコルは、ジョブおよびタスクコミットフェーズ中の Amazon S3 での名前変更操作を回避することで、アプリケーションのパフォーマンスを向上させます。
MagicCommitProtocol は、S3A ファイルシステムが使用されているときに Amazon Elastic Map Reduce (EMR) で実行されている Spark で使用されるデフォルトの FileCommitProtocol 実装です。MagicCommitProtocol は、内部的に [MagicV2Committer](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/s3a-magicv2-committer.html) を使用して Amazon S3 へのファイル書き込みを実行します。
静的挿入オペレーションの場合、MagicCommitProtocol はタスクコミットフェーズ中にジョブの出力場所にファイルを書き込みます。対照的に、動的挿入上書きオペレーションの場合、タスク試行によって書き込まれたファイルは、ジョブのコミット時にジョブの出力場所にのみ表示されます。これは、コミットメタデータをタスクコミット呼び出しで Spark ドライバーにエクスポートすることで実現されます。
## MagicCommitProtocol の有効化
MagicCommitProtocol は、S3A ファイルシステムを使用する時に Amazon Elastic Map Reduce (EMR) で実行される Spark に対してデフォルトで有効になっています。
S3A ファイルシステムを使用するには、以下のいずれかを実行します:
1. テーブル、パーティション、またはディレクトリを定義するときは、ファイルスキームを `s3a://` として使用します。
1. core-site.xml で設定 `fs.s3.impl=org.apache.hadoop.fs.s3a.S3AFileSystem` を行います。
## MagicCommitProtocol の無効化
1. `spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol.leverageMagicCommitProtocol` は、`SparkConf` でハードコーディングするか、Spark シェルまたは `spark-submit` および `spark-sql` ツールで `--conf` パラメータとして渡すか、`conf/spark-defaults.conf` で false に設定できます。詳細については、Apache Spark ドキュメントの「[Spark Configuration](https://spark.apache.org/docs/latest/configuration.html)」を参照してください。
次の例は、`spark-sql` コマンドの実行中に MagicCommitProtocol を無効にする方法を示しています。
```
spark-sql \
--conf spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol.leverageMagicCommitProtocol=false \
-e "INSERT OVERWRITE TABLE target_table SELECT * FROM source_table;"
```
1. `spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol.leverageMagicCommitProtocol` プロパティを false に設定するには、`spark-defaults` 設定分類を使用します。詳細については、「[アプリケーションの設定](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-configure-apps.html)」を参照してください。
## MagicCommitProtocol に関する考慮事項
+ 静的パーティション挿入の場合、Spark エグゼキュターで、MagicCommitProtocol 向けに最適化されたコミットプロトコルは、タスクがコミットされるか中止されるまで、タスクの試行によって書き込まれた各ファイルのために少量のメモリを消費します。ほとんどのジョブで消費されるメモリの量は無視できる程度です。Spark ドライバーに追加のメモリ要件はありません
+ 動的パーティション挿入の場合、Spark ドライバーでは、ジョブがコミットまたは中止されるまで、MagicCommitProtocol はコミットされた各ファイルのメタデータ情報を保存するメモリを必要とします。ほとんどのジョブでは、Spark ドライバーのデフォルトのメモリ設定はごくわずかです。
多数のファイルを書き込む長時間のタスクを含むジョブの場合、コミットプロトコルが消費するメモリが多くなり、Spark、特に Spark エグゼキュターに割り当てられたメモリの調整が必要になることがあります。Spark ドライバーの `spark.driver.memory` プロパティと Spark エグゼキュターの `spark.executor.memory` プロパティを使用してメモリを調整できます。ガイドラインとして、100,000 個のファイルを書き込む 1 つのタスクでは、一般的に 200 MB のメモリを追加する必要があります。詳細については、Apache Spark Configuration ドキュメントの「[Application Properties](https://spark.apache.org/docs/latest/configuration.html#application-properties)」を参照してください。
# EMRFS S3 向けに最適化されたコミッターを使用する
EMRFS S3 向けに最適化されたコミッターは代替 [OutputCommitter](https://hadoop.apache.org/docs/current/api/org/apache/hadoop/mapreduce/OutputCommitter.html) 実装であり、EMRFS を使用するときの Amazon S3 へのファイル書き込みに対して最適化されています。EMRFS S3 向けに最適化されたコミッターは、ジョブおよびタスクコミットフェーズ中に Amazon S3 で行われるリストオペレーションと名前変更オペレーションを回避することにより、アプリケーションのパフォーマンスを向上させることができます。コミッターは Amazon EMR リリース 5.19.0 以降で使用でき、Amazon EMR 5.20.0 以降ではデフォルトで有効になっています。コミッターは、Spark、DataFrame、または Dataset を使用する Spark ジョブに使用されます。Amazon EMR 6.4.0 以降では、Parquet、ORC、テキストベースの形式 (CSV と JSON を含む) など、一般的なあらゆる形式にこのコミッターを使用できます。Amazon EMR 6.4.0 より前のリリースでは、Parquet 形式のみがサポートされています。コミッターが使用されない状況があります。詳細については、「[EMRFS S3 向けに最適化されたコミッターの要件](emr-spark-committer-reqs.md)」を参照してください。
**Topics**
+ [EMRFS S3 向けに最適化されたコミッターの要件](emr-spark-committer-reqs.md)
+ [EMRFS S3 向けに最適化されたコミッターとマルチパートアップロード](emr-spark-committer-multipart.md)
+ [ジョブの調整に関する考慮事項](emr-spark-committer-tuning.md)
+ [Amazon EMR 5.19.0 の EMRFS S3 向けに最適化されたコミッターを有効にする](emr-spark-committer-enable.md)
# EMRFS S3 向けに最適化されたコミッターの要件
以下の条件が満たされる場合に、EMRFS S3 向けに最適化されたコミッターが使用されます。
+ Spark、DataFrame、または Dataset を使用して Amazon S3 にファイルを書き込む Spark ジョブを実行する。Amazon EMR 6.4.0 以降では、Parquet、ORC、テキストベースの形式 (CSV と JSON を含む) など、一般的なあらゆる形式にこのコミッターを使用できます。Amazon EMR 6.4.0 より前のリリースでは、Parquet 形式のみがサポートされています。
+ Amazon EMR でマルチパートアップロードが有効になっている。これがデフォルトです。詳細については、「[EMRFS S3 向けに最適化されたコミッターとマルチパートアップロード](emr-spark-committer-multipart.md)」を参照してください。
+ Spark の組み込みファイル形式のサポートが使用されます。組み込みファイル形式のサポートは以下の状況で使用されます。
+ Hive メタストアテーブルの場合、Parquet テーブルに対して `spark.sql.hive.convertMetastoreParquet` が `true` に設定される場合、または、Amazon EMR 6.4.0 以降の Orc テーブルに対して `spark.sql.hive.convertMetastoreOrc` が `true` に設定される場合。これらはデフォルトの設定です。
+ ジョブによってファイル形式のデータソースまたはテーブルに書き込まれる場合。例えば、ターゲットテーブルが `USING parquet` 句で作成される場合などです。
+ ジョブでパーティション分割されていない Hive メタストア Parquet テーブルに書き込む場合。Spark の組み込み Parquet サポートはパーティション分割された Hive テーブルをサポートしていません。これは既知の制限です。詳細については、Apache Spark, DataFrames and Datasets Guide の「[Hive metastore Parquet table conversion](https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#hive-metastore-parquet-table-conversion)」を参照してください。
+ デフォルトのパーティションの場所 (`${table_location}/k1=v1/k2=v2/` など) に書き込む Spark ジョブオペレーションでコミッターが使用される。ジョブオペレーションによってカスタムのパーティション場所に書き込まれる場合、例えば、カスタムのパーティション場所が `ALTER TABLE SQL` コマンドを使用して設定されている場合、コミッターは使用されません。
+ Spark で以下の値を使用する。
+ `spark.sql.parquet.fs.optimized.committer.optimization-enabled` プロパティを `true` に設定する必要があります。これは、Amazon EMR 5.20.0 以降でのデフォルト設定です。Amazon EMR 5.19.0 では、デフォルト値は `false` です。この値の設定については、「[Amazon EMR 5.19.0 の EMRFS S3 向けに最適化されたコミッターを有効にする](emr-spark-committer-enable.md)」を参照してください。
+ パーティション化されていない Hive メタストアテーブルに書き込む場合は、Parquet と Orc のファイル形式のみがサポートされます。パーティション化されていない Parquet Hive メタストアテーブルに書き込む場合は、`spark.sql.hive.convertMetastoreParquet` は `true` に設定する必要があります。パーティション化されていない Orc Hive メタストアテーブルに書き込む場合は、`spark.sql.hive.convertMetastoreOrc` は `true` に設定する必要があります。これらはデフォルトの設定です。
+ `spark.sql.parquet.output.committer.class` を `com.amazon.emr.committer.EmrOptimizedSparkSqlParquetOutputCommitter` に設定する必要があります。これはデフォルトの設定です。
+ `spark.sql.sources.commitProtocolClass` は `org.apache.spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol` または `org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol` に設定する必要があります。Amazon EMR 5.x シリーズバージョン 5.30.0 以降、および Amazon EMR 6.x シリーズバージョン 6.2.0 以降の場合、`org.apache.spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol` がデフォルト設定です。それよりも前の Amazon EMR バージョンの場合、`org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol` がデフォルト設定です。
+ Spark ジョブでパーティション分割された Parquet データセットを動的パーティション列で上書きする場合は、`partitionOverwriteMode` 書き込みオプションと `spark.sql.sources.partitionOverwriteMode` を `static` に設定する必要があります。これはデフォルトの設定です。
**注記**
`partitionOverwriteMode` 書き込みオプションは Spark 2.4.0 で導入されました。Amazon EMR リリース 5.19.0 に含まれている Spark バージョン 2.3.2 では、`spark.sql.sources.partitionOverwriteMode` プロパティを設定します。
## EMRFS S3 用に最適化されたコミッターが使用されていない場合
一般的に、EMRFS S3 用に最適化されたコミッターは次の状況では、使用されません。
****
| 状況 | コミッターが使われない理由 |
| --- | --- |
| HDFS に書き込む場合 | コミッターは、EMRFS を使用した Amazon S3 への書き込みのみをサポートします。 |
| S3A ファイルシステムを使用する場合 | コミッターは EMRFS のみをサポートします。 |
| MapReduce または Spark の RDD API を使用する場合 | コミッターは SparkSQL、DataFrame、またはデータセット API の使用のみをサポートします。 |
以下の Scala の例では、いくつかの追加の状況を示しています。EMRFS S3 向けに最適化されたコミッターを全体に使用しないもの (最初の例) と、部分的に使用しないもの (2 番目の例) です。
**Example - 動的パーティション上書きモード**
以下の Scala の例では、別のコミットアルゴリズムを使用するように Spark に指示しています。これでは、EMRFS S3 向けに最適化されたコミッターはまったく使用されません。このコードは、データを書き込むパーティションのみを上書きするように `partitionOverwriteMode` プロパティを `dynamic` に設定します。次に、動的パーティション列が `partitionBy` で指定され、書き込みモードが `overwrite` に設定されます。
```
val dataset = spark.range(0, 10)
.withColumn("dt", expr("date_sub(current_date(), id)"))
dataset.write.mode("overwrite")
.option("partitionOverwriteMode", "dynamic")
.partitionBy("dt")
.parquet("s3://amzn-s3-demo-bucket1/output")
```
EMRFS S3 向けに最適化されたコミッターが使用されないようにするには、3 つの設定をすべて構成する必要があります。これを行うと、Spark は Spark のコミットプロトコルで指定されている別のコミットアルゴリズムを実行します。5.30.0 より前の Amazon EMR 5.x リリースと 6.2.0 より前の Amazon EMR 6.x リリースの場合、コミットプロトコルは Spark のステージングディレクトリを使用します。これは、`.spark-staging` で始まる出力場所に作成された一時ディレクトリです。このアルゴリズムではパーティションディレクトリの名前が順番に変更されるため、パフォーマンスが低下する可能性があります。Amazon EMR リリース 5.30.0 以降および 6.2.0 以降の詳細については、「[EMRFS S3 向けに最適化されたコミットプロトコルを使用する](emr-spark-s3-optimized-commit-protocol.md)」を参照してください。
Spark 2.4.0 のアルゴリズムは以下の手順に従います。
1. タスク試行により、Spark のステージングディレクトリ (`${outputLocation}/spark-staging-${jobID}/k1=v1/k2=v2/` など) の下のパーティションディレクトリに出力が書き込まれます。
1. 書き込まれたパーティションごとに、タスク試行は相対パーティションパス (`k1=v1/k2=v2` など) を管理します。
1. タスクが正常に完了すると、追跡されたすべての相対パーティションパスがドライバーに渡されます。
1. すべてのタスクが完了した後、ジョブのコミットフェーズでは、成功したタスクの試行によって Spark のステージングディレクトリに書き込まれたすべてのパーティションディレクトリが収集されます。ディレクトリツリーの名前変更オペレーションを使用して、これらの各ディレクトリの名前が最終的な出力場所に順番に変更されます。
1. ステージングディレクトリがジョブのコミットフェーズの完了前に削除されます。
**Example - カスタムのパーティション場所**
この例では、Scala コードは 2 つのパーティションを挿入します。1 つのパーティションはカスタムのパーティション場所を使用します。もう 1 つのパーティションはデフォルトのパーティション場所を使用します。EMRFS S3 向けに最適化されたコミッターは、デフォルトのパーティション場所を使用するパーティションへのタスク出力の書き込みにのみ使用されます。
```
val table = "dataset"
val location = "s3://bucket/table"
spark.sql(s"""
CREATE TABLE $table (id bigint, dt date)
USING PARQUET PARTITIONED BY (dt)
LOCATION '$location'
""")
// Add a partition using a custom location
val customPartitionLocation = "s3://bucket/custom"
spark.sql(s"""
ALTER TABLE $table ADD PARTITION (dt='2019-01-28')
LOCATION '$customPartitionLocation'
""")
// Add another partition using default location
spark.sql(s"ALTER TABLE $table ADD PARTITION (dt='2019-01-29')")
def asDate(text: String) = lit(text).cast("date")
spark.range(0, 10)
.withColumn("dt",
when($"id" > 4, asDate("2019-01-28")).otherwise(asDate("2019-01-29")))
.write.insertInto(table)
```
Scala コードは以下の Amazon S3 オブジェクトを作成します。
```
custom/part-00001-035a2a9c-4a09-4917-8819-e77134342402.c000.snappy.parquet
custom_$folder$
table/_SUCCESS
table/dt=2019-01-29/part-00000-035a2a9c-4a09-4917-8819-e77134342402.c000.snappy.parquet
table/dt=2019-01-29_$folder$
table_$folder$
```
カスタムのパーティション場所に書き込むとき、先ほどの例と同様のコミットアルゴリズムが使用されます。先ほどの例と同様、このアルゴリズムでは名前が順番に変更されるため、パフォーマンスが低下する可能性があります。
1. カスタムのパーティション場所に出力を書き込むとき、タスクでは最終的な出力場所に作成される Spark のステージングディレクトリにファイルを書き込みます。ファイルの名前には、ファイルの競合から保護するためのランダムな UUID が含まれます。タスクの試行によって各ファイルが最終的な出力パスと共に追跡されます。
1. タスクが正常に完了すると、ドライバーにファイルとそれらの最終的な出力パスが渡されます。
1. すべてのタスクが完了した後、ジョブのコミットフェーズでは、カスタムのパーティション場所に書き込まれたすべてのファイルの名前が、最終的な出力パスに順番に変更されます。
1. ステージングディレクトリがジョブのコミットフェーズの完了前に削除されます。
# EMRFS S3 向けに最適化されたコミッターとマルチパートアップロード
EMRFS S3 に最適化されたコミッターを使用するには、Amazon EMR のマルチパートアップロードを有効にする必要があります。マルチパートアップロードは、デフォルトで有効になっています。また、必要に応じて再度有効にすることができます。詳細については、「[Amazon EMR 管理ガイド](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-upload-s3.html#Config_Multipart)」の「*Amazon S3 用のマルチパートアップロードを設定する*」を参照してください。
EMRFS S3 向けに最適化されたコミッターは、マルチパートアップロードのトランザクションのような特性を利用して、タスクのコミット時にタスクの試行によって書き込まれたファイルのみがジョブの出力場所に表示されるようにします。この方法でマルチパートアップロードを使用することにより、コミッターはデフォルトの FileOutputCommitter アルゴリズムバージョン 2 におけるタスクコミットのパフォーマンスを向上させます。EMRFS S3 向けに最適化されたコミッターを使用するにあたっては、従来のマルチパートアップロードとのいくつかの重要な違いを考慮する必要があります。
+ マルチパートアップロードは常にファイルサイズに関係なく実行されます。これは EMRFS のデフォルトの動作とは異なり、マルチパートアップロードがトリガーされるファイルサイズは `fs.s3n.multipart.uploads.split.size` プロパティで制御されます。
+ タスクがコミットされるか中止されるまで、マルチパートアップロードは長期間不完全な状態になります。これは EMRFS のデフォルトの動作とは異なり、タスクで指定されたファイルの書き込みが終了するとマルチパートアップロードが完了します。
このような違いにより、タスクが実行されて Amazon S3 へのデータの書き込みが行われているときに Spark Executor JVM がクラッシュしたり強制終了されたりすると、不完全なマルチパートアップロードが残される可能性が高くなります。そのため、EMRFS S3 に最適化されたコミッターを使用するときは、失敗したマルチパートアップロードの管理に関するベストプラクティスに従ってください。詳細については、「*Amazon EMR 管理ガイド*」の Amazon S3 バケットの処理に関する[ベストプラクティス](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-upload-s3.html#emr-bucket-bestpractices)を参照してください。
# ジョブの調整に関する考慮事項
EMRFS S3 向けに最適化されたコミッターは、タスクがコミットされるか中止されるまで、タスクの試行によって書き込まれた各ファイルのために少量のメモリを消費します。ほとんどのジョブで消費されるメモリの量は無視できる程度です。多数のファイルを書き込む長時間のタスクを含むジョブの場合、コミッターが消費するメモリが多くなり、Spark エグゼキュターに割り当てれたメモリの調整が必要になることがあります。エグゼキュターのメモリは、`spark.executor.memory` プロパティを使用して調整できます。ガイドラインとして、100,000 個のファイルを書き込む 1 つのタスクでは、一般的に 100 MB のメモリを追加する必要があります。詳細については、Apache Spark Configuration ドキュメントの「[Application Properties](https://spark.apache.org/docs/latest/configuration.html#application-properties)」を参照してください。
# Amazon EMR 5.19.0 の EMRFS S3 向けに最適化されたコミッターを有効にする
Amazon EMR 5.19.0 を使用している場合は、クラスターの作成時、または Spark 内から (Amazon EMR を使用しているとき)、`spark.sql.parquet.fs.optimized.committer.optimization-enabled` プロパティを `true` に手動で設定できます。
## EMRFS S3 向けに最適化されたコミッターをクラスターの作成時に有効にする
`spark.sql.parquet.fs.optimized.committer.optimization-enabled` プロパティを `true` に設定するには、`spark-defaults` 設定分類を使用します。詳細については、「[アプリケーションの設定](emr-configure-apps.md)」を参照してください。
## EMRFS S3 向けに最適化されたコミッターを Spark から有効にする
`spark.sql.parquet.fs.optimized.committer.optimization-enabled` は、`SparkConf` でハードコーディングするか、Spark シェルまたは `spark-submit` および `spark-sql` ツールで `--conf` パラメータとして渡すか、`conf/spark-defaults.conf` によって `true` に設定できます。詳細については、Apache Spark ドキュメントの「[Spark Configuration](https://spark.apache.org/docs/latest/configuration.html)」を参照してください。
次の例は、spark-sql コマンドの実行中にコミッターを有効にする方法を示しています。
```
spark-sql \
--conf spark.sql.parquet.fs.optimized.committer.optimization-enabled=true \
-e "INSERT OVERWRITE TABLE target_table SELECT * FROM source_table;"
```
# EMRFS S3 向けに最適化されたコミットプロトコルを使用する
EMRFS S3 向けに最適化されたコミットプロトコルは代替 [FileCommitProtocol](https://spark.apache.org/docs/2.2.0//api/java/org/apache/spark/internal/io/FileCommitProtocol.html) 実装であり、EMRFS を使用するときの Amazon S3 への Spark の動的パーティション上書きを使用するファイル書き込みに対して最適化されています。このプロトコルは、Spark の動的パーティション上書きジョブのコミットフェーズ中の Amazon S3 での名前変更操作を回避することで、アプリケーションのパフォーマンスを向上させます。
[EMRFS S3-optimized committer](emr-spark-s3-optimized-committer.html) はまた、名前変更操作を回避することでパフォーマンスを向上させることにも注意してください。ただし、動的パーティション上書きの場合は機能しませんが、コミットプロトコルの改善は動的パーティション上書きの場合のみを対象としています。
コミットプロトコルは Amazon EMR リリース 5.30.0 以降で使用でき、6.2.0 以降ではデフォルトで有効になっています。リリース 5.31.0 以降、Amazon EMR では並列処理の改善が追加されました。プロトコルは、Spark、DataFrame、または Dataset を使用する Spark ジョブに使用されます。コミットプロトコルが使用されない状況があります。詳細については、「[EMRFS S3 向けに最適化されたコミットプロトコルの要件](emr-spark-committer-reqs.md)」を参照してください。
**Topics**
+ [EMRFS S3 向けに最適化されたコミットプロトコルの要件](emr-spark-commit-protocol-reqs.md)
+ [EMRFS S3 向けに最適化されたコミットプロトコルとマルチパートアップロード](emr-spark-commit-protocol-multipart.md)
+ [ジョブの調整に関する考慮事項](emr-spark-commit-protocol-tuning.md)
# EMRFS S3 向けに最適化されたコミットプロトコルの要件
以下の条件が満たされる場合に、EMRFS S3 向けに最適化されたコミットプロトコルが使用されます。
+ Spark、DataFrame、または Dataset によりパーティション化されたテーブルを書き込む Spark ジョブを実行する。
+ パーティション上書きモードが `dynamic` である Spark ジョブを実行する。
+ Amazon EMR でマルチパートアップロードが有効になっている。これがデフォルトです。詳細については、「[EMRFS S3 向けに最適化されたコミットプロトコルとマルチパートアップロード](emr-spark-commit-protocol-multipart.md)」を参照してください。
+ EMRFS のファイルシステムキャッシュは有効になっています。これがデフォルトです。`fs.s3.impl.disable.cache` 設定が `false` に設定されていることを確認します。
+ Spark のビルトインデータソースサポートが使用されています。組み込みデータソースサポートは以下の状況で使用されます。
+ ジョブでビルトインのデータソースまたはテーブルに書き込む場合。
+ ジョブで Hive メタストア Parquet テーブルに書き込む場合。これは `spark.sql.hive.convertInsertingPartitionedTable` と `spark.sql.hive.convertMetastoreParquet` が両方とも true に設定されている場合に発生します。これらはデフォルトの設定です。
+ ジョブで Hive メタストア ORC テーブルに書き込む場合。これは、`spark.sql.hive.convertInsertingPartitionedTable` と `spark.sql.hive.convertMetastoreOrc` が両方 `true` に設定されている場合に発生します。これらはデフォルトの設定です。
+ デフォルトのパーティションの場所 (`${table_location}/k1=v1/k2=v2/` など) に書き込む Spark ジョブオペレーションでコミットプロトコルが使用される。ジョブオペレーションによってカスタムのパーティション場所に書き込まれる場合、例えば、カスタムのパーティション場所が `ALTER TABLE SQL` コマンドを使用して設定されている場合、プロトコルは使用されません。
+ Spark で以下の値を使用する。
+ `spark.sql.sources.commitProtocolClass` を `org.apache.spark.sql.execution.datasources.SQLEmrOptimizedCommitProtocol` に設定する必要があります。これは、Amazon EMR リリース 5.30.0 以降および 6.2.0 以降のデフォルト設定です。
+ `partitionOverwriteMode` 書き込みオプションまたは `spark.sql.sources.partitionOverwriteMode` は `dynamic` に設定する必要があります。デフォルトの設定は `static` です。
**注記**
`partitionOverwriteMode` 書き込みオプションは Spark 2.4.0 で導入されました。Amazon EMR リリース 5.19.0 に含まれている Spark バージョン 2.3.2 では、`spark.sql.sources.partitionOverwriteMode` プロパティを設定します。
+ Spark ジョブが Hive メタストアの Parquet テーブルに上書きされる場合は、`spark.sql.hive.convertMetastoreParquet`、`spark.sql.hive.convertInsertingPartitionedTable` および `spark.sql.hive.convertMetastore.partitionOverwriteMode` を `true` に設定する必要があります。これらはデフォルトの設定です。
+ Spark ジョブが Hive メタストアの ORC テーブルに上書きされる場合は、`spark.sql.hive.convertMetastoreOrc`、`spark.sql.hive.convertInsertingPartitionedTable` および `spark.sql.hive.convertMetastore.partitionOverwriteMode` を `true` に設定する必要があります。これらはデフォルトの設定です。
**Example - 動的パーティション上書きモード**
この Scala の例では、最適化がトリガーされます。まず、`partitionOverwriteMode` プロパティを `dynamic` に設定します。これにより、データを書き込んでいるパーティションのみが上書きされます。次に、動的パーティション列が `partitionBy` で指定され、書き込みモードが `overwrite` に設定されます。
```
val dataset = spark.range(0, 10)
.withColumn("dt", expr("date_sub(current_date(), id)"))
dataset.write.mode("overwrite") // "overwrite" instead of "insert"
.option("partitionOverwriteMode", "dynamic") // "dynamic" instead of "static"
.partitionBy("dt") // partitioned data instead of unpartitioned data
.parquet("s3://amzn-s3-demo-bucket1/output") // "s3://" to use Amazon EMR file system, instead of "s3a://" or "hdfs://"
```
## EMRFS S3 向けに最適化されたコミットプロトコルが使用されない場合
一般的に、EMRFS S3 向けに最適化されたコミットプロトコルは、オープンソースのデフォルトの Spark コミットプロトコル `org.apache.spark.sql.execution.datasources.SQLHadoopMapReduceCommitProtocol` と同じように機能します。次の状況では、最適化は行われません。
****
| 状況 | コミットプロトコルが使用されない理由 |
| --- | --- |
| HDFS に書き込む場合 | コミットプロトコルは、EMRFS を使用した Amazon S3 への書き込みのみをサポートします。 |
| S3A ファイルシステムを使用する場合 | コミットプロトコルは EMRFS のみをサポートします。 |
| MapReduce または Spark の RDD API を使用する場合 | コミットプロトコルは、SparkSQL、DataFrame、またはデータセット API の使用のみをサポートします。 |
| 動的パーティションの上書きがトリガーされない場合 | コミットプロトコルは、動的なパーティション上書きの場合のみを最適化します。その他の場合については、「[EMRFS S3 向けに最適化されたコミッターを使用する](emr-spark-s3-optimized-committer.md)」を参照してください。 |
次の Scala の例は、EMRFS S3 向けに最適化されたコミットプロトコルが `SQLHadoopMapReduceCommitProtocol` に委任するその他の状況を示しています。
**Example - カスタムパーティションの場所を使用する動的パーティション上書きモード**
この例では、Scala プログラムは動的パーティション上書きモードで 2 つのパーティションを上書きします。1 つのパーティションはカスタムのパーティション場所を使用します。もう 1 つのパーティションはデフォルトのパーティション場所を使用します。EMRFS S3 向けに最適化されたコミットプロトコルは、デフォルトのパーティション場所を使用するパーティションのみを改善します。
```
val table = "dataset"
val inputView = "tempView"
val location = "s3://bucket/table"
spark.sql(s"""
CREATE TABLE $table (id bigint, dt date)
USING PARQUET PARTITIONED BY (dt)
LOCATION '$location'
""")
// Add a partition using a custom location
val customPartitionLocation = "s3://bucket/custom"
spark.sql(s"""
ALTER TABLE $table ADD PARTITION (dt='2019-01-28')
LOCATION '$customPartitionLocation'
""")
// Add another partition using default location
spark.sql(s"ALTER TABLE $table ADD PARTITION (dt='2019-01-29')")
def asDate(text: String) = lit(text).cast("date")
spark.range(0, 10)
.withColumn("dt",
when($"id" > 4, asDate("2019-01-28")).otherwise(asDate("2019-01-29")))
.createTempView(inputView)
// Set partition overwrite mode to 'dynamic'
spark.sql(s"SET spark.sql.sources.partitionOverwriteMode=dynamic")
spark.sql(s"INSERT OVERWRITE TABLE $table SELECT * FROM $inputView")
```
Scala コードは以下の Amazon S3 オブジェクトを作成します。
```
custom/part-00001-035a2a9c-4a09-4917-8819-e77134342402.c000.snappy.parquet
custom_$folder$
table/_SUCCESS
table/dt=2019-01-29/part-00000-035a2a9c-4a09-4917-8819-e77134342402.c000.snappy.parquet
table/dt=2019-01-29_$folder$
table_$folder$
```
以前の Spark バージョンでは、カスタムパーティションの場所に書き込むと、データが失われる可能性があります。この例では、パーティション `dt='2019-01-28'` が失われます。詳細については「[SPARK-35106](https://issues.apache.org/jira/browse/SPARK-35106)」を参照してください。Amazon EMR リリース 5.33.0 以降 (6.0.x および 6.1.x は除く) では、この問題は修正されています。
カスタムのパーティション場所に書き込むとき、先ほどの例と同様のコミットアルゴリズムが使用されます。先ほどの例と同様、このアルゴリズムでは名前が順番に変更されるため、パフォーマンスが低下する可能性があります。
Spark 2.4.0 のアルゴリズムは以下の手順に従います。
1. カスタムのパーティション場所に出力を書き込むとき、タスクでは最終的な出力場所に作成される Spark のステージングディレクトリにファイルを書き込みます。ファイルの名前には、ファイルの競合から保護するためのランダムな UUID が含まれます。タスクの試行によって各ファイルが最終的な出力パスと共に追跡されます。
1. タスクが正常に完了すると、ドライバーにファイルとそれらの最終的な出力パスが渡されます。
1. すべてのタスクが完了した後、ジョブのコミットフェーズでは、カスタムのパーティション場所に書き込まれたすべてのファイルの名前が、最終的な出力パスに順番に変更されます。
1. ステージングディレクトリがジョブのコミットフェーズの完了前に削除されます。
# EMRFS S3 向けに最適化されたコミットプロトコルとマルチパートアップロード
EMRFS S3 向けに最適化されたコミットプロトコルで動的パーティション上書きの最適化を使用するには、Amazon EMR でマルチパートアップロードを有効にする必要があります。マルチパートアップロードは、デフォルトで有効になっています。また、必要に応じて再度有効にすることができます。詳細については、「[Amazon EMR 管理ガイド](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-upload-s3.html#Config_Multipart)」の「*Amazon S3 用のマルチパートアップロードを設定する*」を参照してください。
動的パーティション上書きの間、EMRFS S3 向けに最適化されたコミットプロトコルは、マルチパートアップロードのトランザクションのような特性を利用して、ジョブのコミット時にタスクの試行によって書き込まれたファイルのみがジョブの出力場所に表示されるようにします。このようにマルチパートアップロードを使用することにより、コミットプロトコルはジョブコミットのパフォーマンスをデフォルトの `SQLHadoopMapReduceCommitProtocol` よりも向上させます。EMRFS S3 向けに最適化されたコミットプロトコルを使用するにあたっては、従来のマルチパートアップロードとのいくつかの重要な違いを考慮する必要があります。
+ マルチパートアップロードは常にファイルサイズに関係なく実行されます。これは EMRFS のデフォルトの動作とは異なり、マルチパートアップロードがトリガーされるファイルサイズは `fs.s3n.multipart.uploads.split.size` プロパティで制御されます。
+ タスクがコミットされるか中止されるまで、マルチパートアップロードは長期間不完全な状態になります。これは EMRFS のデフォルトの動作とは異なり、タスクで指定されたファイルの書き込みが終了するとマルチパートアップロードが完了します。
このような違いにより、タスクが実行されて Amazon S3 へのデータの書き込みが行われているときに Spark Executor JVM がクラッシュしたり強制終了されたりする、またはジョブが実行しているときに Spark Driver JVM がクラッシュしたり強制終了されたりすると、不完全なマルチパートアップロードが残される可能性が高くなります。そのため、EMRFS S3 に最適化されたコミットプロトコルを使用するときは、失敗したマルチパートアップロードの管理に関するベストプラクティスに従ってください。詳細については、「*Amazon EMR 管理ガイド*」の Amazon S3 バケットの処理に関する[ベストプラクティス](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-plan-upload-s3.html#emr-bucket-bestpractices)を参照してください。
# ジョブの調整に関する考慮事項
Spark エグゼキュターで、EMRFS S3 向けに最適化されたコミットプロトコルは、タスクがコミットされるか中止されるまで、タスクの試行によって書き込まれた各ファイルのために少量のメモリを消費します。ほとんどのジョブで消費されるメモリの量は無視できる程度です。
Spark ドライバーで、EMRFS S3 向けに最適化されたコミットプロトコルには、ジョブがコミットまたは中止されるまで、コミットされた各ファイルのメタデータ情報を保存するためのメモリが必要です。ほとんどのジョブでは、Spark ドライバーのデフォルトのメモリ設定はごくわずかです。
多数のファイルを書き込む長時間のタスクを含むジョブの場合、コミットプロトコルが消費するメモリが多くなり、Spark、特に Spark エグゼキュターに割り当てられたメモリの調整が必要になることがあります。Spark ドライバーの `spark.driver.memory` プロパティと Spark エグゼキュターの `spark.executor.memory` プロパティを使用してメモリを調整できます。ガイドラインとして、100,000 個のファイルを書き込む 1 つのタスクでは、一般的に 100 MB のメモリを追加する必要があります。詳細については、Apache Spark Configuration ドキュメントの「[Application Properties](https://spark.apache.org/docs/latest/configuration.html#application-properties)」を参照してください。
# EMRFS で Amazon S3 リクエストを再試行する
このトピックでは、EMRFS を使用して Amazon S3 にリクエストを行うときに使用できる再試行戦略について説明します。リクエストレートが上昇すると、S3 は新しいレートをサポートするためにスケーリングを試みます。このプロセス中、S3 はリクエストをスロットルし、`503 Slow Down` エラーを返す場合があります。S3 リクエストの成功率を高めるために、`emrfs-site` 設定でプロパティを設定し、再試行戦略を調整できます。
次の方法で再試行戦略を調整できます。
+ デフォルトのエクスポネンシャルバックオフ再試行戦略の再試行上限を増やします。
+ additive-increase/multiplicative-decrease (AIMD) 再試行戦略を有効にして設定します。AIMD は Amazon EMR リリース 6.4.0 以降でサポートされています。
## デフォルトのエクスポネンシャルバックオフ戦略を使用する
デフォルトでは、EMRFS はエクスポネンシャルバックオフ戦略を使用して Amazon S3 リクエストを再試行します。EMRFS のデフォルトの再試行上限は 15 です。S3 `503 Slow Down` エラーを回避するために、新しいクラスターを作成するときに、実行中のクラスターで、またはアプリケーションのランタイムで、再試行上限を増やすことができます。
再試行上限を増やすには、`emrfs-site` で `fs.s3.maxRetries` の値を変更する必要があります。以下の設定例では、`fs.s3.maxRetries` をカスタム値の 30 に設定します。
```
[
{
"Classification": "emrfs-site",
"Properties": {
"fs.s3.maxRetries": "30"
}
}
]
```
設定オブジェクトの処理の詳細については、「[アプリケーションの設定](emr-configure-apps.md)」を参照してください。
## AIMD 再試行戦略を使用する
Amazon EMR リリース 6.4.0 以降では、EMRFS は、additive-increase/multiplicative-decrease (AIMD) モデルに基づく代替再試行戦略をサポートします。AIMD 再試行戦略は、大規模な Amazon EMR クラスターで作業する場合に特に便利です。
AIMD は、最近の成功したリクエストに関するデータを使用してカスタムリクエストレートを計算します。この戦略では、スロットリングされるリクエストの数と、リクエストごとに必要な合計試行数が減少します。
AIMD 再試行戦略を有効にするには、以下の例のように、`emrfs-site` 設定で `fs.s3.aimd.enabled` プロパティを `true` に設定する必要があります。
```
[
{
"Classification": "emrfs-site",
"Properties": {
"fs.s3.aimd.enabled": "true"
}
}
]
```
設定オブジェクトの処理の詳細については、「[アプリケーションの設定](emr-configure-apps.md)」を参照してください。
## 詳細な AIMD 再試行設定
次の表にリストされているプロパティを設定して、AIMD 再試行戦略を使用するときの再試行の動作を詳細に指定できます。大部分のユースケースで、デフォルト値を使用することをお勧めします。
**高度な AIMD 再試行戦略のプロパティ**
| プロパティ | デフォルトの値 | 説明 |
| --- | --- | --- |
| fs.s3.aimd.increaseIncrement | 0.1 | 連続するリクエストが成功したときにリクエストレートが上昇する速さを制御します。 |
| fs.s3.aimd.reductionFactor | 2 | Amazon S3 が 503 レスポンスを返すときにリクエストレートが減少する速さを制御します。デフォルト係数の 2 は、リクエストレートを半分に削減します。 |
| fs.s3.aimd.minRate | 0.1 | リクエストで S3 による持続的なスロットリングが生じる場合のリクエストレートの下限を設定します。 |
| fs.s3.aimd.initialRate | 5500 | 初期リクエストレートを設定します。その後、これは、fs.s3.aimd.increaseIncrement と fs.s3.aimd.reductionFactor に指定する値に従って変わります。初期レートは GET リクエストにも使用され、PUT リクエストに比例してスケーリングされます (3500/5500)。 |
| fs.s3.aimd.adjustWindow | 2 | リクエストレートを調整する頻度を制御し、応答数で測定します。 |
| fs.s3.aimd.maxAttempts | 100 | リクエストを試行する最大試行回数を設定します。 |