S3Aコネクタ使用時のパフォーマンス最大化

fadvise sequential

ファイル全体を読み取ります。場合によっては、短い前方シークを伴います。

ドキュメント全体が単一のHTTPリクエストでリクエストされます。リードアヘッド範囲内の前方シークは、中間データをスキップすることでサポートされます。

これにより、最大のシーケンシャルスループットが実現されますが、後方シークは非常にコストがかかります。

ファイル (DistCP、コピー操作) を一括で読み取るアプリケーションは、gzip圧縮された .gz ファイルからデータを読み取るアプリケーションと同様に、シーケンシャルアクセスを使用する必要があります。「ノーマル」fadviseポリシーはシーケンシャルIOモードで開始するため、このポリシーを明示的にリクエストする必要はめったにありません。

fadvise random

ランダムIO、特にHadoop PositionedReadable 操作向けに最適化されています。ただし、seek(offset); read(byte_buffer) もメリットがあります。

ファイル全体を要求するのではなく、HTTPリクエストの範囲は、read 操作で必要なデータの長さに設定されます (必要に応じて、setReadahead() で設定されたリードアヘッド値に切り上げられます)。

既存のHTTPリクエストを閉じるコストを削減することにより、これは、一連の PositionedReadable.read() および PositionedReadable.readFully() 呼び出しを介してバイナリファイルにアクセスするファイルIOに非常に効率的です。ファイルのシーケンシャル読み取りは、ファイル全体を読み取るために多数のHTTPリクエストを行う必要があるため、コストがかかります。各GET操作の間には遅延があります。

ランダムIOは、シークが頻繁に行われるIOに最適です。

• データは PositionedReadable APIを使用して読み取られます。
• 長距離 (数MB) の前方シーク
• 後方シークは前方シークと同様に可能性が高い。
• シングル文字の read() 呼び出しまたは小さな read(buffer) 呼び出しはほとんどまたはまったく使用しない。
• S3データストアに近い場所で実行されるアプリケーション。つまり、S3インスタンスと同じデータセンター内のEC2 VMです。

目的のfadviseポリシーは、ファイルシステムインスタンスが作成されるときに、構成オプション fs.s3a.experimental.input.fadvise で設定する必要があります。つまり、ファイルごとの読み取りではなく、ファイルシステムごとにのみ設定できます。

<property>
  <name>fs.s3a.experimental.input.fadvise</name>
  <value>random</value>
  <description>
  Policy for reading files.
  Values: 'random', 'sequential' or 'normal'
   </description>
</property>

HDFS-2744、fadviseを許可するようにFSDataInputStreamを拡張する は、入力ストリームにfadviseポリシーを設定するためのパブリックAPIを追加することを提案しています。実装されると、これが入力IOポリシーの構成に使用されるサポートされているメカニズムになります。

fadvise normal (デフォルト)

normal ポリシーは、sequential モードでファイルの読み取りを開始しますが、呼び出し元がストリーム内で後方にシークすると、シーケンシャルから random に切り替わります。

このポリシーは、基本的に、ファイルの最後にシークし、インデックスデータを読み込み、次に後方にシークして列を選択的に読み込む列指向ストレージ形式 (例: Apache ORCおよびApache Parquet) の最初の読み取りパターンを認識します。最初のシークはランダムポリシーと比較してコストがかかる場合がありますが、全体的なプロセスは、random ポリシーでファイルを順次読み取るか、sequential ポリシーで列指向データを読み取るよりもはるかにコストが低くなります。

スレッドと接続プールのサイズ

単一のユーザーとして、単一のバケットと対話する各S3Aクライアントは、アップロードおよびコピー操作に使用されるスレッドのプールに加えて、オープンなHTTP 1.1接続の専用プールを持っています。デフォルトのプールサイズは、パフォーマンスとメモリ/スレッドの使用量のバランスを取るように意図されています。

プロパティを設定することにより、並列IO（特にアップロード）のためのより大きな（再利用された）HTTP接続とスレッドのプールを持つことができます。

多くのIOを実行するプロセス（DistCp、Sparkワーカーなど）には、より大きな値を使用することをお勧めします。

<property>
  <name>fs.s3a.threads.max</name>
  <value>20</value>
</property>
<property>
  <name>fs.s3a.connection.maximum</name>
  <value>20</value>
</property>

ただし、多くの並列クエリを実行するプロセスは、各クエリが異なるS3バケットのセットで動作している場合、または異なるユーザーの代わりに動作している場合、大量のリソースを消費する可能性があることに注意してください。

大規模なデータアップロードの場合、ブロックサイズを調整します: fs.s3a.block.size

データをアップロードする場合、fs.s3a.block.sizeオプションで設定されたブロックでアップロードされます。デフォルト値は「32M」で32メガバイトです。

より大きな値が使用されると、アップロードが開始される前により多くのデータがバッファリングされます。

<property>
  <name>fs.s3a.block.size</name>
  <value>128M</value>
</property>

これは、S3にデータをアップロードするために作成されるPUT/POSTリクエストが少なくなり、S3がクライアントをスロットリングする可能性が低くなることを意味します。

場合によっては: メモリに書き込みデータをバッファリングする

大きなファイルをアップロードする場合、ブロックはディスクに保存され、アップロードのためにキューに入れられ、複数のスレッドが異なるブロックを並行してアップロードします。

fs.s3a.fast.upload.bufferオプションをbytebufferに設定するか、オンヒープストレージの場合はarrayに設定することで、ブロックをメモリにバッファリングできます。

1. メモリ内IOに切り替えることでディスクIOが削減され、S3ストアへの帯域幅が非常に高く、ディスクIOがボトルネックになる場合は高速化できます。これは、サーバーへの帯域幅が非常に高いオンプレミスS3互換オブジェクトストアを使用する場合に、具体的なメリットをもたらす可能性があります。

メモリにバッファリングすると、メモリ不足になりやすくなります。オプションfs.s3a.fast.upload.active.blocks"は、S3に書き込む単一の出力ストリームが一度にキューに入れることができるアクティブなブロック数を調整するために存在します。

バッファリングされた各ブロックのサイズはfs.s3a.block.sizeの値によって決定されるため、ブロックサイズが大きいほど、メモリ不足になる可能性が高くなります。

DistCP: チューニングするオプション

• -numListstatusThreads <threads> : デフォルト（1）よりも高い値に設定します。
• -bandwidth <mb> : ワーカーごとのアップロード帯域幅を制限するために使用します
• -m <maps> : マッパーの数、したがってS3バケットへの負荷を制限します。

-mオプションを使用してマップを追加しても、パフォーマンスが向上する保証はありません。スロットリングの量が増加するだけの場合があります。マップごとの帯域幅が高い、より少ないマップの方が効率的な場合があります。

DistCP: 回避するオプション

DistCpの-atomicオプションは、データをディレクトリにコピーし、コピーが実行される場所に名前を変更します。これはパフォーマンスを低下させます。

• -atomicオプションを使用しないでください。
• -append操作はS3でサポートされていません。避けてください。
• -p S3にはPOSIXスタイルのアクセス許可モデルがありません。これは失敗します。

DistCP: チューニングするパラメータ

1. 前述のように、fs.s3a.threads.maxとfs.s3a.connection.maximumに大きな値を使用します。

2. バケットがsequentialまたはnormalのfadviseシークポリシーを使用していることを確認してください。つまり、fs.s3a.experimental.input.fadviseがrandomに設定されていないことを確認してください。

3. -numListstatusThreadsをより高い数値に設定して、リストを並行して実行します。fs.s3a.connection.maximumが使用される値以上であることを確認してください。

4. -deleteを使用する場合は、fs.trash.intervalを0に設定して、削除されたオブジェクトがごみ箱ディレクトリにコピーされないようにします。

メモリ内にバッファリングするためにfs.s3a.fast.upload.bufferを切り替えないでください。1つのdistcpマッパーがメモリ不足になると失敗し、ジョブ全体が失敗するリスクがあります。デフォルト値のdiskを維持する方が安全です。

潜在的に有用なのは、より大きなブロックでアップロードすることです。これはHTTP接続の使用という点でより効率的であり、S3バケット/シャードに対するIOPレートを下げます。

<property>
  <name>fs.s3a.threads.max</name>
  <value>20</value>
</property>

<property>
  <name>fs.s3a.connection.maximum</name>
  <value>30</value>
  <descriptiom>
   Make greater than both fs.s3a.threads.max and -numListstatusThreads
   </descriptiom>
</property>

<property>
  <name>fs.s3a.experimental.input.fadvise</name>
  <value>normal</value>
</property>

<property>
  <name>fs.s3a.block.size</name>
  <value>128M</value>
</property>

<property>
  <name>fs.s3a.fast.upload.buffer</name>
  <value>disk</value>
</property>

<property>
  <name>fs.trash.interval</name>
  <value>0</value>
</property>

スロットリングを抑えるためのヒント

distcpコピーのような大規模な操作で多くのスロットリング応答が見られる場合は、バケットを操作しようとするプロセスの数を減らしてください（distcpの場合：-mオプションを使用してマッパーの数を減らします）。

ファイルリストを読み書きしている場合は、単純なソートされた順序で処理されないようにリストをランダム化すると、特定のS3データのシャードへの負荷を減らし、それによってスループットを向上させることができます。

S3バケットは、同時に接続しているすべてのクライアントからのリクエストによってスロットリングされます。異なるアプリケーションとジョブは互いに干渉する可能性があります。トラブルシューティングを行う際は、そのことを考慮してください。データを異なるバケットに分割すると、ここで負荷を分離するのに役立つ場合があります。

SSE-KMSで暗号化されたデータを使用している場合は、これも適用されます。これらはS3の数値よりも厳格です。これらの制限に達していると思われる場合は、制限を増やすことができる場合があります。 KMSレート制限ドキュメントを参照してください。

rename()

データを一時ファイルにアップロードし、rename() を使用して最終パスにコミットするアルゴリズムは避けてください。HDFS では、これは高速なコミット操作を提供します。S3、Wasb、その他のオブジェクトストアでは、書き込みをクローズするまでファイルが表示されないことを知った上で、直接宛先に書き込むことができます。書き込み自体はアトミックです。

rename() 操作は、ソースが見つからない場合、false を返すことがあります。これは API の弱点です。rename を呼び出す前にチェックを行い、新しい rename() 呼び出しが公開された場合は、それに切り替えることを検討してください。

delete(path, recursive)

delete(path, recursive) は、パスが存在しない場合は何も処理しないことに注意してください。そのため、呼び出す前にパスが存在するかどうかをチェックする必要はありません。

delete() は、クリーンアップ操作としてよく使用されます。オブジェクトストアでは、これは遅く、呼び出し元が即時応答を期待する場合、問題を引き起こす可能性があります。たとえば、スレッドが長時間ブロックされ、他の活性チェックが失敗し始める可能性があります。これらのバックグラウンドクリーンアップ操作を実行するために、executor スレッドを生成することを検討してください。

OpenSSL アクセラレーション

実験的な機能

HADOOP-16050 および HADOOP-16346 の時点では、fs.s3a.ssl.channel.mode を default または openssl のいずれかに設定して、HTTPS リクエストのネイティブ OpenSSL アクセラレーションを有効にできます。OpenSSL は、ネイティブコードを使用して SSL および TLS プロトコルを実装します。HTTPS 経由で大量のデータを読み取るユーザーの場合、OpenSSL は JSSE よりも大幅なパフォーマンス上の利点を提供できます。

S3A は、WildFly OpenSSL ライブラリを使用して、OpenSSL を Java JSSE API にバインドします。このライブラリを使用すると、S3A は OpenSSL を使用して透過的にデータを読み取ることができます。wildfly-openssl ライブラリは S3A のオプションのランタイム依存関係であり、Java JSSE を OpenSSL にバインドするためのネイティブライブラリが含まれています。

WildFly OpenSSL は OpenSSL 自体をロードする必要があります。これは、システムプロパティ org.wildfly.openssl.path を使用して実行できます。たとえば、HADOOP_OPTS="-Dorg.wildfly.openssl.path=<OpenSSL ライブラリへのパス> ${HADOOP_OPTS}" のようにします。詳細については、WildFly OpenSSL のドキュメントを参照してください。

fs.s3a.ssl.channel.mode が default に設定されている場合、S3A は WildFly ライブラリを使用して OpenSSL ライブラリのロードを試みます。失敗した場合、default_jsse の動作にフォールバックします。

fs.s3a.ssl.channel.mode が openssl に設定されている場合、S3A は WildFly を使用して OpenSSL ライブラリのロードを試みます。失敗した場合、例外がスローされ、S3A の初期化は失敗します。

fs.s3a.ssl.channel.mode の構成

fs.s3a.ssl.channel.mode は次のように構成できます。

<property>
  <name>fs.s3a.ssl.channel.mode</name>
  <value>default_jsse</value>
  <description>
    If secure connections to S3 are enabled, configures the SSL
    implementation used to encrypt connections to S3. Supported values are:
    "default_jsse", "default_jsse_with_gcm", "default", and "openssl".
    "default_jsse" uses the Java Secure Socket Extension package (JSSE).
    However, when running on Java 8, the GCM cipher is removed from the list
    of enabled ciphers. This is due to performance issues with GCM in Java 8.
    "default_jsse_with_gcm" uses the JSSE with the default list of cipher
    suites. "default_jsse_with_gcm" is equivalent to the behavior prior to
    this feature being introduced. "default" attempts to use OpenSSL rather
    than the JSSE for SSL encryption, if OpenSSL libraries cannot be loaded,
    it falls back to the "default_jsse" behavior. "openssl" attempts to use
    OpenSSL as well, but fails if OpenSSL libraries cannot be loaded.
  </description>
</property>

fs.s3a.ssl.channel.mode でサポートされている値

命名規則は、HADOOP-15669 の ABFS サポートとの下位互換性を維持するために設定されています。

SSL の最適化が進むにつれて、fs.s3a.ssl.channel.mode に他のオプションが将来追加される可能性があります。

WildFly クラスパスの要件

OpenSSL アクセラレーションを機能させるには、互換性のあるバージョンの wildfly JAR がクラスパス上にある必要があります。これは、オプションであるため、公開された hadoop-aws モジュールの依存関係で明示的に宣言されていません。

wildfly JAR が見つからない場合、ネットワークアクセラレーションは常に JVM にフォールバックします。

注意: 過去に wildfly JAR と openSSL リリースの間に互換性の問題が発生しています。バージョン 1.0.4.Final は openssl 1.1.1 と互換性がありませんでした。さらに複雑なのは、hadoop-azure-datalake で使用されている古いバージョンの azure-data-lake-store-sdk JAR には、1.0.4.Final クラスのシェーディングされていないコピーが含まれており、より新しいバージョンが明示的にクラスパスに配置されている場合でもバインディングの問題が発生していました。

バケット存在チェックの無効化

S3A Filesystem インスタンスが作成および初期化されると、クライアントは指定されたバケットが有効かどうかをチェックします。これは時間がかかる場合があります。fs.s3a.bucket.probe を次のように構成することで、バケット検証を無視できます。

<property>
  <name>fs.s3a.bucket.probe</name>
  <value>0</value>
</property>

注意: バケットが存在しない場合、この問題はファイルシステムで操作が実行されたときに表面化します。UnknownStoreException スタックトレースが表示されます。

並列 FileSystem 作成操作のレート制限

アプリケーションは通常、FileSystem.get() または Path.getFileSystem() を介して、共有キャッシュからファイルシステムを要求します。キャッシュ FileSystem.CACHE は、ユーザーごとに、指定された URI のファイルシステムのインスタンスを 1 つキャッシュします。s3a://landsat-pds/ などの URI のキャッシュされた FS に対する FileSystem.get のすべての呼び出しは、その 1 つのインスタンスを返します。

FileSystem インスタンスは、キャッシュに対してオンデマンドで作成され、インスタンスを要求する各スレッドで実行されます。これは、同期ブロックの外で行われます。タスクが初期化された FileSystem インスタンスを持っていると、同期ブロックでそれをキャッシュに追加します。キャッシュがその URI のインスタンスをすでに持っていることが判明した場合、キャッシュされたコピーを元に戻し、作成したばかりの FS インスタンスを閉じます。

FileSystem の初期化に時間がかかり、多くのスレッドが同じ S3 バケットの FileSystem インスタンスを並行して取得しようとしている場合、スレッドの 1 つを除くすべてが無駄な作業を行い、共有オブジェクトで意図せずにロックの競合を引き起こす可能性があります。

並行して作成できる FS インスタンスの数を制限するためにセマフォを使用するオプション fs.creation.parallel.count があります。

これを低い数値に設定すると、異なるオブジェクトストア/分散ファイルシステムに対して同時に作成できる FileSystem クライアントの数が制限される代わりに、無駄な作業の量が減ります。

たとえば、値が 4 の場合、s3a://landsat-pds/ バケットのコネクタの無駄なインスタンス化の数に上限が設定されます。

<property>
  <name>fs.creation.parallel.count</name>
  <value>4</value>
</property>

また、4 つのスレッドがこのようなコネクタを作成している場合、他のバケットのコネクタを作成しようとするすべてのスレッドもブロックされることになります。

多くのスレッドが同じ初期化に時間がかかるオブジェクトストアと同時に対話しようとする可能性のあるアプリケーションを実行する場合は、これを試してみることを検討してください。