Apache Sparkの最適化技術とパフォーマンスチューニング

Chandan Gaur|Big Data Engineering|05,Oct,2020

Apache Sparkとは何ですか？

Apacheは、2012年に、分散クラスタ上の読み取り専用分散データセットを持つResilient Distributed Dataset(APACHE SparkのRDD)財団を説明し、Apache Sparkと命名しました。 その後、Dataset APIを導入し、データのバッチおよび構造化ストリーミング用のDataframe Apiを導入しました。 この記事では、最良のApache Spark最適化手法を一覧表示します。

Apache Sparkは、より多くの計算とストリーム処理を実行するために開発された高速クラスタコンピューティングプラットフォームです。 Sparkは、複数のシステムの実行とサポートを必要とする従来のシステムと比較して、さまざまなワークロードを処理できます。 データ解析のパイプラインは生産に必要である異なった処理のタイプの組合せの火花によって促進される。 Apache Sparkは、YARNやそのスタンドアロンマネージャなどの外部クラスターマネージャで動作するように作成されます。

Apache Sparkの機能

Apache Sparkのいくつかの機能は次のとおりです:-

• ビッグデータアプリケーションを作成するための統合プラットフォーム。
• アクセス性が高いように設計されています。
• Sparkは独立して実行できます。 従ってそれは柔軟性を与える。

XenonStackは、モノのインターネット、監視、予防および予知保全のために、リアルタイムおよびストリームデータの取り込み、処理、およびデータストリームの分析を迅速かつ効率的に行うための分析サービスおよびソリューションを提供します。

記事から、ストリーミングとリアルタイム分析サービス

Apache Sparkの最適化がどのように機能するかを理解していますか？

Apache Spark最適化がどのように機能するかを理解するためには、まずそのアーキテクチャを理解する必要があり、以降のセクションでも同様に説明します。

Apache Sparkのアーキテクチャ

Sparkのランタイムアーキテクチャは三つの部分で構成されています-

Sparkドライバ(マスタープロセス)

Sparkドライバはプログ タスクスケジューラは、ドライバの一部であり、エグゼキュータにタスクを配布するのに役立ちます。

Spark Cluster Manager

クラスターマネージャーは、executorを起動することを可能にするSparkのコアであり、時にはドライバも起動することができます。 Spark Schedulerは、cluster manager自体のFIFO方法でSparkアプリケーションのアクションとジョブをスケジュールします。 Apache Airflowについても読む必要があります。

エグゼキュータ(スレーブプロセス)

エグゼキュータは、ジョブの個々のタスクが実行される個々のエンティティです。 Executorは、起動されるとsparkアプリケーションのライフサイクルまで常に実行されます。 失敗したエグゼキューターはsparkジョブの実行を停止しません。

RDD(Resilient Distributed Datasets)

RDDは、クラスターの分散ノード上の不変データセットの分散コレクションです。 RDDは1つまたは複数のパーティションに分割されます。 RDDは、データの局所性を活用するクラスターのさまざまなノード間での分布としてのsparkの中核です。 アプリケーション内で並列処理を実現するために、パーティションはそのための単位です。 再パーティション変換または合体変換は、パーティションの数を維持するのに役立ちます。 データアクセスはRDDシャッフルを利用して最適化されます。 Sparkはデータに近いため、Sparkを介してさまざまなノードにデータを送信し、必要に応じて必要なパーティションを作成します。

DAG(有向非巡回グラフ)

Sparkコンソールにコードを入力すると、sparkは演算子グラフを生成する傾向があります。 SPARK RDDにアクションがトリガーされると、SparkはそのグラフをDAGSchedulerに送信します。 次に、これらの演算子グラフをDAGScheduler内のタスクの段階に分割します。 すべてのステップには、受信データの複数のパーティションに基づくジョブが含まれています。 DAGSchedulerは、これらの個々の演算子グラフを一緒にパイプライン化します。 例えば、マップ演算子グラフは、単一のステージのスケジュールと、これらのステージはに渡されます。 クラスターマネージャのタスクスケジューラを実行します。 これは、スレーブ上でこれらのタスクを実行するための作業またはエグゼキュータのタスクです。

パーティションを効率的に使用した分散処理

クラスター上のエグゼキュータの数を増やすと、Sparkジョブの処理における並列性も向上します。 しかし、このためには、そのデータがパーティション化を介してそれらのエグゼキュータ間でどのように配布されるかにつ RDDは、これらのエグゼキュータ間でのデータシャッフルのトラフィックが無視できる場合に役立ちます。 ペアRDD（キーと値のペアを持つRDD）の分割をカスタマイズすることができます。 Sparkは、この場合には明示的な制御がないため、キーのセットが常に同じノードに一緒に表示されることを保証します。

Apache Sparkセキュリティは、共有シークレットを介した認証を支援します。 Spark認証は、認証を構成するための構成パラメータです。 Spark通信のプロトコルが共有シークレットを使用して認証を行っているかどうかをチェックするパラメータです。

記事から、Apache Spark Security

Sparkアプリケーションを書くときに避けるべき間違い

reduceByKeyまたはgroupByKey

groupByKeyとreduceByKeyの両方が同じ答えを生成しますが、結果を生成す reduceByKeyは、Sparkでは、データをシャッフルする前に各パーティションの共有キーと出力を結合するため、大規模なデータセットに最適です。 反対側では、groupByKeyはすべてのキーと値のペアをシャッフルします。 GroupByKeyは、不要なシャッフルとネットワーク上のデータの転送を引き起こします。

シャッフルブロックの必要なサイズを維持

デフォルトでは、Spark shuffleブロックは2GBを超えることはできません。 より良い使用方法は、パーティションを増やし、シャッフルブロックサイズを小さくするパーティションあたりの容量を-128MBに減らすことです。 通常のアプリケーションでは、再分割または合体を使用できます。 大きなパーティションは、2GBの制限のためにプロセスが遅くなり、いくつかのパーティションは、ジョブを拡張し、並列処理を達成することはできません。

ファイル形式と区切り文字

各データ関連の仕様に適切なファイル形式を選択することは頭痛の種です。 取り込みタイプ、中間タイプ、および最終出力タイプのデータ形式を賢明に選択する必要があります。 また、AvroはParquetよりもバイナリデータに最適化されているため、メディアデータを格納するためにAVRO file formatを使用できるなど、いくつかの方法で各タイプのデー Parquetは、高度に圧縮されているため、メタデータ情報を格納するために使用できます。

小さなデータファイル

ブロードキャストは、小さなデータファイルやデータセットをメモリブロックにロードし、データをシャッフルするオーバーヘッドを少なくして、より大規模なデータセットと結合できるようにする技術です。 例えば、小さなデータファイルをn個のブロックに格納することができ、大きなデータファイルをこれらのブロック間で並列に分散させることができ、将来的には大きなデータファイルをこれらのデータブロックに結合することができます。

ジョブステージの監視なし

DAGは、タスクのさまざまなステージをグラフ形式で記述するSparkで使用されるデータ構造です。 ほとんどの開発者はコードを記述して実行しますが、ジョブタスクの監視は不可欠です。 この監視は、DAGを管理し、ステージを削減することによって最適に達成されます。 20段階のジョブは、3-4段階のジョブと比較して延長されます。

ByKey、再分割、またはシャッフルをトリガするその他の操作

ほとんどの場合、データシャッフルと同じくらいシャッフルを避ける必要があります。 GroupByKeyは貴重な資産になる可能性がありますが、その必要性を最初に説明する必要があります。

強化学習

強化学習は、より良い機械学習環境を得るだけでなく、より良い方法で意思決定を処理する概念です。 遷移モデルと報酬モデルがデータセット上に正しく構築され、エージェントが結果を推定するのに十分な能力がある場合、sparkで深い強化学習を適用する必

Apache Spark最適化の要因とテクニック

Apache Spark最適化の最高の機能の一つは、メモリ内のデータ計算に役立ちます。 これらのspark最適化計算のボトルネックは、CPU、メモリ、またはクラスター内の任意のリソースです。 このような場合には、データをシリアル化し、メモリを削減する必要が生じる可能性がある。 Spark最適化のためのこれらの要因は、適切に使用される場合、次のことができます–

• 長時間実行されるジョブプロセスを排除
• 修正実行エンジン
• リソースを管理することにより、パフォーマンス時間を向上させます

13 Apache Spark最適化のための簡単なテクニック

アキュムレータは、連想演算と可換演算によってのみ追加できるエグゼキュータへのグローバル変数です。 したがって、それは並行して効率的である可能性があります。 アキュムレータは、カウンタ(Map Reduceと同じ)やAPI呼び出しの追跡などの別のタスクを実装するために使用できます。

デフォルトでは、Sparkは数値アキュムレータをサポートしていますが、プログラマは新しい型のサポートを追加する利点があります。 Sparkは、各タスクの更新がアキュムレータ変数に一度だけ適用されるようにします。 ジョブステージが再実行された場合、これらは複数回適用される可能性があるため、変換中に、ユーザーは各タスクの更新を認識する必要があります。

Hiveバケットパフォーマンス

バケットは、バケットでバケットの数を指定するため、固定数のファイルで結果をバケット化します。 Hiveはフィールドを取得し、ハッシュを計算し、その特定のバケットにレコードを割り当てます。 バケット化は、フィールドの基数が高く、データ処理が大きく、レコードがすべてのバケットに均等に分散されている場合により安定しますが、パーティショニングフィールドの基数が低い場合にはパーティショニングが機能します。

バケット化により、ファイルの並べ替えのオーバーヘッドが軽減されます。 たとえば、同じ数のバケットを持つ2つのテーブルを結合する場合、sparkは既にソートされているバケットのキーとしてデータを直接結合します。 バケットファイルの数は、複数のバケットに複数のパーティションとして計算できます。

述語プッシュダウン最適化

述語プッシュダウンは、必要なデータのみを処理する手法です。 Where条件でフィルターを定義することで、述語をSparkSQLに適用できます。 Explainコマンドを使用してクエリを実行すると、クエリ処理段階を確認できます。 クエリプランにPushedFilterが含まれている場合は、すべての述語がTrueまたはFalseを返すため、必要なデータのみを選択するようにクエリが最適化されます。

クエリプランにPushedFilterが見つからない場合は、where条件をキャストすることをお勧めします。 述語プッシュダウンは、照会中にSparkSQLが読み取るファイルとパーティションの数を制限するため、インメモリ分析を開始するディスクI/Oを削減します。 述語プッシュダウンを使用してバケット内のデータを照会すると、シャッフルが少なくても高速に結果が得られます。

Apache Arrowを使用したゼロデータのシリアル化/逆シリアル化

Apache Arrowは、分析クエリエンジンのメモリ内ランタイムフォーマットとして使用されます。 Arrowは、共有メモリを介したデータのシリアル化/逆シリアル化ゼロシャッフルを提供します。 Arrow flightは、ネットワーク経由で大きなデータセットを送信します。 Arrowには、ディスク上のデータへのゼロコピーランダムアクセスを可能にするarrowファイル形式があります。 Arrowには、すべてのsparkアプリケーション用の標準データアクセスレイヤーがあります。

すべてのデータが存在し、矢印固有の形式で共通の場所を持つため、データをシャッフルするためのSerDe操作のオーバーヘッドを削減します。

G1GCコレクションを使用したガベージコレクションのチューニング

ガベージコレクタをチューニングする場合は、まずG1GCを使用してSparkアプ G1ガベージコレクタは、Sparkで一般的に見られる成長するヒープを完全に処理します。 G1を使用すると、より高いスループットとより低い遅延の両方を提供するために必要なオプションが少なくなります。 さまざまなアプリケーションの予測不可能な特性と動作を制御するには、生成されたログに従ってGCチューニングを習得する必要があります。

その前に、ストリーミングやリアルタイム分析ソリューションなどの他の最適化手法をプログラムのロジックとコードに適用する必要があります。 ほとんどの場合、G1GCはSparkアプリケーションで頻繁に使用されるプロセス間の一時停止時間を最適化するのに役立ち、より信頼性の高いシステムで

メモリ管理とチューニング

シャッフル、ソートなどの計算には実行メモリが使用され、キャッシュ目的には内部データも伝播するストレージメモリが使 ジョブがキャッシュを使用していない場合があるため、実行中に領域外エラーが発生する場合があります。 キャッシュされたジョブは、実行要件によってデータを削除することができない場合、常により少ない記憶領域を適用します。 また、Apache Sparkを使用したリアルタイムストリーミングアプリケーションを行うことができます。

sparkを設定できます。メモリ。Spark実行メモリに使用されるJVMヒープ領域の量を決定するための分数。 一般的には、60%がデフォルトです。 Executorメモリは、JVMガベージコレクションの遅延につながる可能性があるため、できるだけ少なく保つ必要があります。 この事実は、単一のJVMインスタンス上で複数のタスクが実行される可能性があるため、小さなエグゼキュータにも適用できます。

データの局所性

Apache Sparkでは、データの局所性と呼ばれる処理されたデータの近くに実行コードを配置することによって、処理タスクが最適化されます。 データが利用できないため、処理タスクがデータを取得する前に待機する必要がある場合があります。 しかし、ときに火花の時間。地域性。つまり、任意のノードにラックするノードにローカルです。

ディスク間のデータ転送は非常にコストがかかるため、ほとんどの操作はデータが存在する場所で実行する必要があります。 これは、ロードするのに役立ちますが、Apache Sparkのテスト駆動開発と一緒にデータ量を必要としました。

コロケーションされた結合を使用

コロケーションされた結合は、再配布とブロードキャストの決定を行います。 ブロードキャストのより良い利用を達成するために，複数のメモリブロックに配置される小さなデータセットを定義することができる。 2つのデータセットに結合を適用している間、sparkは最初に両方のデータセットのデータをキーでソートし、それらをマージします。

しかし、それらを結合する前に、またはそれらのデータフレームをApache Arrow Architectureで作成するときに、並べ替えパーティションキーを適用することもできます。 これにより、sortへの不要な関数呼び出しがないため、クエリの実行時が最適化されます。

Sparkでのキャッシュ

GPUを使用したApache Sparkでのキャッシュは、何度も何度もデータが必要な場合にApache Sparkの最適化に最適な手法です。 しかし、データをキャッシュすることは常に許容されません。

次のいずれかの場合には、cache()RDDとDataFramesを使用する必要があります–

• 機械学習アルゴリズムのような反復ループがある場合。
• 単一のジョブまたはタスクでRDDに複数回アクセスされた場合。
• RDD区画を再度生成するコストが高い場合。

Cache()とpersist(StorageLevel.MEMORY_ONLY)は、お互いの代わりに使用することができます。 メモリから追い出されるすべてのRDDパーティションは、依然として非常に高価なソースから再度ビルドする必要があります。 最善の解決策の1つは、persist（ストレージレベル。MEMORY_AND_DISK_ONLY)は、Rddのパーティションをワーカーのローカルディスクにこぼすことになります。 この場合は、比較的高速なワーカーのローカルドライブからデータを取得する必要があります。

Executor Size

メモリの多いexecutorを実行すると、ガベージコレクションが過度に遅延することがよくあります。 Executorあたりのコア数をexecutorあたりの5つのタスク以下に保つ必要があります。 小さすぎるexecutorsは、単一のJVM上で複数のジョブを実行するという点で便利ではありませんでした。

たとえば、ブロードキャスト変数は、各エグゼキュータに対して正確に一度複製する必要があります。

Sparkウィンドウ関数

ウィンドウ関数は、テーブルの入力行を計算できるフレームを定義します。 個々の行レベルで。 各行は明確なフレームワークを持つことができます。 ウィンドウを使用すると、データフレーム内のデータのウィンドウを定義できます。 同じデータフレーム内の複数の行を比較できます。

以前のデータとのデータ依存の問題を解決する特定の間隔にウィンドウ時間を設定することができます。 Apache Beamでのシャッフルは、ウィンドウ間隔のデータを保持しているため、以前に処理されたデータでは少なくなります。

透かし技法

透かし技法はApache Spark最適化において有用な技法であり、設計上システムを制約し、実行中にシステムが爆発するのを防ぐのに役立 Watermarkは二つの引数を取ります–

• イベント時間の列と
• 遅延データを処理するために必要な時間を指定するしきい値時間

Apache Arrowアーキテクチャのクエリは、データがその規定されたしきい値 完全モードは最初にすべてのデータを結果のテーブルに保持するため、完全モードを透かしで並べて使用できることを覚えておく必要があります。

データのシリアル化

Apache Spark optimizationは、分析などのいくつかのユースケースやデータの移動だけのために処理する必要があるデータに対して動作します。 このようなデータまたは分析の移動は、データがよりシリアル化された形式である場合には、適切に実行できます。 Apache Sparkは、ソースまたは宛先で必要なデータ形式を効果的に管理するためのデータシリアル化をサポートしています。 デフォルトでは、Apache SparkはJavaシリアル化を使用しますが、Kryoシリアル化もサポートしています。

デフォルトでは、SparkはJavaのObjectOutputStreamを使用してデータをシリアル化します。 実装はjavaを介して行うことができます。イオ直列化可能クラス。 オブジェクトをバイトのストリームにエンコードします。 それは軽量の持続を適用範囲が広い提供し。 しかし、それが使用される各クラスの巨大な直列化された形式につながるので、それは遅くなります。 Sparkは、Javaよりもコンパクトであるため、Javaのシリアル化よりもほぼ10倍高速なオブジェクトのシリアル化のためのKryo Serialization library（v4）をサポートしています。

: 包括的なアプローチ

オープンソースの分散コンピューティングエンジンであるApache Sparkは、現在、インメモリバッチ処理のための最も一般的なフレームワークであり、リアルタイムストリーミングもサポートしている。 高度なクエリオプティマイザと実行エンジンにより、Apache Sparkの最適化技術は、大規模なデータセットを非常に効率的に処理および分析できます。 しかし、Apache Spark Joinの最適化手法を慎重にチューニングせずに実行すると、パフォーマンスが低下する可能性があります。 Apache Sparkアプリケーションのパワーを活用したい場合は、Managed Apache Sparkサービスをチェックしてください。

:

ビッグデータ開発、ビッグデータソリューション、ストリーミングデータ分析、

シェア：