Apache Spark チューニング戦略の理解と適用

この記事を読む動機。

• 私自身の経験は、混乱の瞬間と冷静な分析の瞬間にありました。
• さらに深く掘り下げるために探したもの。
• Spark が最適化にどのように機能するかについて学んだこと。
• 最適化のための Databricks の「プラス」とは何ですか。
• チューニングやリファクタリングの必要性を回避できる優れたプラクティス。

導入

私はリレーショナル データベースと、その後は Spark のような分散システムと常によく関わってきました。最初は、複雑なクエリの設定、管理、そして主に DBMS のパフォーマンス スクリプトのまとめ方について、DBMS をさらに詳しく調べました。 Spark をさらに使用し、その後 Databricks を使用するようになったとき、最初は構築する必要があったシナリオのパフォーマンスの問題はありませんでしたが、ビッグデータ領域が実際にビッグデータになるにつれて、ルーチンでパフォーマンスの問題が発生し始め、それが毎回 30% ずつ増加しました。今週、私は Spark が「内部で」どのように動作するかを調べることになりました。主な理由は、DBMS がどのように動作するかをすでに知っていたためであり、これはここで説明するいくつかの概念を理解するのに役立ちました。

Apache Spark コンポーネントの簡単な説明

この記事ではパフォーマンス分析のシナリオ、テクニック、ベスト プラクティスに焦点を当てたいので、簡潔にまとめましょう。

スパークコア:

このコンポーネントは Spark の基礎であり、メモリ管理、タスク、障害回復、I/O 管理を担当します。つまり、RDD を操作します。したがって、彼はクラスターの大部分を占める男です。

執行者:

このコンポーネントは、Spark エコシステム (クラスター) の実際のワーカーであり、ディスク上、メモリ上、またはその両方上にある書き込みまたは読み取り命令 (タスク) を受け取るコンポーネントです (これが登場する理由は後で説明します)プレイ）。パフォーマンスシナリオ）。

労働者：

ワーカーは、分散コンピューティングにすでに精通している人にとって文字通りそのものであり、クラスターのノードであるため、上で述べたエグゼキュータを「ホスト」するものであり、各ワーカーには 1 つ以上のエグゼキュータを含めることができます。実行者がアシスタントであり、作業者が倉庫作業者であるかのように、実行者に割り当てられたリソースを管理する責任があります。彼が直属の倉庫管理者だったらどうしますか?

クラスターマネージャー:

これはマネージャーです。彼はワーカーのリソース (メモリと CPU) を管理します。彼は、各アプリケーションに何人のエグゼキュータが必要か、どのくらいのリソースが割り当てられるかを決定します。彼は、自分の ' によって送信されたタスクを管理します。これについては後ほど説明します。これはより高い責任を負う立場であるため、クラスターの状態を監視して障害から回復し、必要に応じてタスクを再分散します。 (注: クラスター マネージャーにはいくつかの種類があります: Yarn、mesos、kubernetes、そして最も単純なものはスタンドアロンです)。

スパークコンテキスト:

そうですね、これはボスまたはゲートウェイです。ゲートウェイと言うのは、どの Spark アプリケーションもそこを通過するためです。アプリケーションがクラスター (ワーカーやエグゼキューター) と対話できるようにするもので、これを許可および管理します。このようにして、構成、エグゼキュータの数、メモリなどのリソースの観点からアプリケーション全体を管理します。タスクがどのように実行されているかを知る必要がありますか?ここの上司と話してください。

具体的に説明すると、

ここで、パフォーマンス、チューニング、スピード、スピード、そしてさまざまな位置から聞こえるすべてについて話しましょう。

リレーショナル バンキング側と協力し、主にアプリケーションのプロシージャ、関数、またはクエリでパフォーマンスの問題が発生したとき、次の側面を分析しました。

1. このスクリプトはいつ実行されますか?
2. リソースやテーブルのロックを争っている人はいますか?
3. すべてがスムーズで、サーバー リソースをブロックする人もいないので、とても良いです...
4. それではスクリプトを見てみましょう。そのロジックはパフォーマンス的ですか?言い換えれば、一緒に読み書きすることを考えた人、またはそれを 1 行ずつ考えた人 (プログラミング中毒) は、必要のない列、サブクエリを使用した巨大なクエリ、CTE などを参照しすぎていませんか?これらすべての点を変更し (リファクタリング)、応答速度とサーバー リソースの使用量の両方をテストしました。 Apache Spark について説明するときに、なぜこれについて説明する必要があるのでしょうか?つまり...これは Spark にも当てはまり、ある意味ではさらに複雑になりますが、そこまではいきません。
5. 最後に、脚本がよければ「石の道」、つまり推定実行計画と実際の実行計画を分析すると思います。このことから、DBMS がその統計 (ヒストグラム) を使って何を行っていたのか、その情報がどのパスをたどると想定していたのか、そして実際はどのようなパスをたどったのかを理解することができました。さらに、クエリ内の追加フィルター、よりパフォーマンスの高い JOIN、さらにはインデックスや一時テーブルの作成などのポイントを特定できます。

そうですね、以上だと思いますが、これらの点と Apache Spark の共通点は何でしょうか?

• スクリプトは分散セット操作用に設計されていません (Spark には「プラス」の難しさがあると言いました笑)。
• 単純な Spark ジョブが、すべてのリソースを消費している別の実行ジョブと同じクラスター内で実行されている場合 (またはそうでない場合)、特定のルーチンが実行されている時間。 (ここで有名な DBMS ロックの一種を見てください)。
• そして最後に、はい、Apache Spark には実行プランがあり、より正確に言うと、次の段階があります。

1. 論理的な計画
2. 物理面
3. 実行戦略。
4. 推定コストが表示される場合があります。

名前にもかかわらず、それぞれが何であるかを要約すると、すでにアイデアを得ることができます:

論理計画:
元のクエリを一連の論理演算として表します。これはクエリの抽象的な形式であり、実際にどのように実行されるかは考慮されていません。フィルタリング、選択、結合、集計などの実行される操作に関する情報と、間違った「小さなこと」も含まれます (笑)。

物理面:
Spark が実際にクエリを実行する方法について詳しく説明します。これには、操作の順序と、使用されるアルゴリズム (DBMS アルゴリズムなど) が含まれます。これには、データがどのように分割され、実行者間で分散されるかについての詳細が含まれる場合があります。

実行戦略:
物理プレーンは、操作やデータ サイズに応じて、「ブロードキャスト結合」や「シャッフル ハッシュ結合」など、Spark が使用できるさまざまな実行戦略を表示できます。実行計画の主なアルゴリズムについても説明します。落ち着いてください...

推定費用:
常に表示されるわけではありませんが、一部のプランにはプランのさまざまな部分のコスト見積もりが含まれており、処理のどの部分に最もコストがかかるかを理解するのに役立ちます。

Apache Spark 実行計画を表示する方法

explain() コマンドを使用すると、テキスト形式の「ルート」フォームがあり、単純なフィルターとデータフレームを示す以下のような出力が得られます。

== 物理的な計画 ==
*(2) フィルタ (値 > 1)
- *(2) プロジェクト [名前#0、値#1]
- *(1) 既存のRDD[名前#0、値#1]をスキャン

そして客観的に言えば、セルの実行、ジョブ、クラスターのいずれであっても、Databricks の Spark UI を通じてインターフェイス経由で分析できます。 Apache Spark では、これはデフォルト ポート 4040 の IP です。

Spark UI はいくつかの便利なセクションに分かれています:

• ジョブ: アプリケーションで実行されたすべてのジョブのリストを表示します。各ジョブはコード内のアクションに対応します。

• ステージ: 各ジョブを構成するステージを表示します。ステージは、並行して実行できる作業の細分化です。

• タスク: タスクの実行時間やステータスに関する情報など、各ステージ内の個々のタスクの詳細を示します。

• ストレージ: RDD (Resilient Distributed Datasets) のメモリとストレージの使用状況に関する情報を提供します。

• 環境: Spark 構成やシステム変数などのランタイム環境プロパティを表示します。

• エグゼキュータ: メモリ使用量、ディスク使用量、パフォーマンス統計など、アプリケーション用に作成されたエグゼキュータに関する情報を表示します。

ここでは私は階層的でした、いいですか？これは、物事が機能する順序です。

画面に画像を載せたい!!

Spark アルゴリズムとチューニング違反者を知る方法:

まず、Spark UI インターフェイスと実行プラン (論理プランまたは物理プラン) の両方で示される主なアルゴリズムについて説明します。

注: ここでのデータセットは Spark テーブルと同じであることに注意してください ;)

1.最も有名なスキャンから始めましょう:

• FileScan: 入力ファイルからデータを読み取ります。寄木細工、ORC、CSV、JSON などのさまざまなファイル形式に最適化できます。

2.参加 (これにより B.O が得られます):

• ブロードキャスト ハッシュ結合: データセットの 1 つがクラスター内のすべてのノードに送信できるほど小さい場合に使用され、シャッフルを回避します (これについては後ほど詳しく説明しますが、簡単に言うと、データ シャッフル操作です)最終結合)。

• シャッフル ハッシュ結合: 両方のデータセット (必要に応じてテーブル) がシャッフルされ、対応するキーが同じパーティション内に存在します。データセットが大きく、他のノードに送信できない場合に使用されます。

• 並べ替え結合結合: 結合する前に両方のデータセットを並べ替える必要があります。これは、すでにパーティション化され順序付けされている大規模なデータセットの場合に効率的です。つまり、パーティション化され順序付けされた列によって結合が行われます (例: df.write.partitionBy("coluna1").sortBy("coluna2").parquet(")パス /to/save/partitioned")

3.集計 (合計、カウント、グループ化など):

• HashAggregate: ハッシュ テーブルを使用してデータを集約します。メモリに収まる大きなデータセットの場合は効率的です。

• ソート集計。データを並べ替えてから集計します。データがメモリに収まらない場合に使用します。

4.シャッフル (この男に注意してください):

• シャッフル: 結合や集計など、再編成が必要な操作のためにパーティション間でデータを再分散します。これは、I/O とネットワークの点で高価な操作です。

5.交換:

• パーティション間のデータの分散を変更します。これを使用して、クラスター ノード間でワークロードのバランスをとることができます。 (バランスをとってシャッフルから逃れる戦略)

6.プロジェクト:

• DataFrame または RDD から列のサブセットを選択します。

7.フィルター:

• 条件を適用してデータ行をフィルターします。

8.並べ替え:

• 1 つ以上の列に基づいてデータを並べ替えます。

上記のアルゴリズムはすべて、前に述べたように Explain() コマンドを通じて確認できます。

実際のシャッフル問題のシナリオ:

1. Join および GroupBy 操作
join() や groupByKey() などの操作は、パーティション間でデータを再分散するシャッフルをトリガーすることがよくあります。これにより、次のような結果が生じる可能性があります。
ディスク I/O 使用率が高い: Shuffle は多くの中間ファイルを生成するため、実行プログラムのローカル ディスクが飽和状態になる可能性があります。
高いネットワーク負荷: 必要な接続数 (マッパーの数とリデューサーの数を掛けたもの) に応じて、エグゼキューター間で転送されるデータの量が膨大になる可能性があります

• 識別: Spark UI の [ステージ] タブで、[シャッフル読み取りサイズ/レコード] および [シャッフル スピル (ディスク)] の値を確認します。これらの指標の量が多い場合は、潜在的な問題があることを示しています。

1. パーティションの不均衡 (データ スキュー) データがパーティション間で不均等に分散されている場合、一部のタスクに他のタスクよりもはるかに時間がかかり、全体的なパフォーマンスが低下する可能性があります。識別は同じです。Spark UI に移動し、時間がかかっているセクションを参照してジョブに移動し (ここで、後述する良い実践のポイントを示します)、スタックしたステージを確認します (実行中ですが、実行されていません)。進行中）、シャッフル メトリクスを確認します。一般に、メモリ内のボリュームが多く、リフレッシュするとディスク上にボリュームが増え始めます。これは、この不均衡がメモリに到達し、ディスクへの書き込みが開始されたことを示しています。明らかにディスクが遅くなっているため、座って、このシナリオを許すと泣く。

緩和

• シャッフル関連の問題を軽減するには: シャッフルを引き起こす操作を減らす: 可能な限り、 groupByKey() を使用し、reduceByKey() を優先します。の数を調整します パーティション:spark.sql.shuffle.partitions を使用して、パーティションの数を調整します。 シャッフル操作中のパーティション。 ブロードキャスト結合などのテクニックを使用します。大規模なセットを結合するには、 データのセットが小さいため、不必要なシャッフルが回避されます。

Spark UI でメトリクスをシャッフルする:

シャッフルの仕組みとコストがかかる理由:

最後に、そしておそらく最も重要なこと - 良い実践例:

1. Databricks、Jupyter Notebook、Google Colab の人気が高いため、大部分はノートブックで動作します。したがって、各クエリまたは変換を個別のセルに分割すると、どの部分がパフォーマンス上の問題であるかを特定しやすくなります。ひとまとめにしてみるとジョブがいくつかあってどの段階なのかわかりにくいです

2. 上書きの代わりにマージを使用します。手間がかかることは承知していますが、データレイク内のテーブル全体を再度「ダンプ」上書きするよりもマージの方が使用する I/O が少ないため、より論理的でパフォーマンスが向上します。

3. 特に複数の操作で中間データを再利用する場合は、cache() またはpersist() を使用してメモリに中間データを保存します。これにより、再計算時間が短縮され、パフォーマンスが向上します。

4. ご存知ない方のために付け加えておきますが、Spark は JVM 上で実行されるため、ネイティブでは Java ですが、有名な UDF (ユーザー定義関数) を Python で作成すると、Spark に一種の「ブラック ボックス」が残され、自動最適化。可能な限り、パフォーマンスが最適化された組み込みの Spark SQL 関数を使用してください。

そうですね、思っていたことはすべて書けたと思います。記事を書くのはシナリオを思い出すのに役立つので好きです。実際にいくつかの公開データを使用して、このすべてを示すビデオを録画するつもりです。おそらく Kaggle で入手できるので、データ、人工知能、ソフトウェア開発の世界に関連するすべての情報を入手するために LinkedIn でフォローしてください

--> https://www.linkedin.com/in/airton-lira-junior-6b81a661

LinkedIn で私をフォローし、後押ししてください。私はフィードバックが好きで、知識の共有を改善することにも完全にオープンです。

ここまで読んでくださった方、おめでとうございます!!!すべてのパフォーマンスの問題が克服されることを願っています。次の記事では Databricks の利点について説明します。LinkedIn で私をフォローして確認してください。ありがとうございます!!

以上がApache Spark チューニング戦略の理解と適用の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。