監視とインストルメンテーション

• Webインターフェイス
  • 事後表示
    • 環境変数
    • ローリングイベントログファイルへのコンパクションの適用
    • Spark History Server の設定オプション
  • REST API
    • Executor タスクメトリクス
    • Executor メトリクス
    • API バージョンポリシー
• メトリクス
  • 利用可能なメトリクスプロバイダのリスト
    • コンポーネントインスタンス = Driver
    • コンポーネントインスタンス = Executor
    • ソース = JVMソース
    • コンポーネントインスタンス = applicationMaster
    • コンポーネントインスタンス = master
    • コンポーネントインスタンス = ApplicationSource
    • コンポーネントインスタンス = worker
    • コンポーネントインスタンス = shuffleService
• 高度なインストルメンテーション

Spark アプリケーションを監視するには、Web UI、メトリクス、外部インストルメンテーションなど、いくつかの方法があります。

Webインターフェイス

各 SparkContext は、アプリケーションに関する有用な情報を表示する Web UI を起動します。デフォルトではポート 4040 で起動し、これには以下が含まれます。

• スケジューラステージとタスクのリスト
• RDD サイズとメモリ使用量の概要
• 環境情報。
• 実行中の Executor に関する情報

Web ブラウザで http://<driver-node>:4040 を開くだけで、このインターフェイスにアクセスできます。同じホストで複数の SparkContext が実行されている場合、それらは 4040 から始まる連続したポート (4041、4042 など) にバインドされます。

デフォルトでは、この情報はアプリケーションの存続期間中のみ利用可能であることに注意してください。後で Web UI を表示するには、アプリケーションを開始する前に spark.eventLog.enabled を true に設定します。これにより、Spark は UI に表示される情報をエンコードした Spark イベントを永続ストレージにログ記録するように構成されます。

事後表示

アプリケーションのイベントログが存在する場合、Spark の履歴サーバーを通じて UI を構築することは依然として可能です。履歴サーバーを起動するには、以下を実行します。

./sbin/start-history-server.sh

これにより、デフォルトで http://<server-url>:18080 に Web インターフェイスが作成され、未完了および完了したアプリケーションと試行が一覧表示されます。

ファイルシステムプロバイダクラス (以下の spark.history.provider を参照) を使用する場合、ベースログディレクトリは spark.history.fs.logDirectory 設定オプションで指定する必要があります。これは、各アプリケーションのイベントログを表すサブディレクトリを含む必要があります。

Spark ジョブ自体は、イベントをログ記録し、それらを同じ共有可能ディレクトリにログ記録するように構成する必要があります。たとえば、サーバーが hdfs://namenode/shared/spark-logs のログディレクトリで構成されている場合、クライアント側のオプションは次のようになります。

spark.eventLog.enabled true
spark.eventLog.dir hdfs://namenode/shared/spark-logs

履歴サーバーは次のように構成できます。

環境変数

ローリングイベントログファイルへのコンパクションの適用

長時間実行されるアプリケーション (例: ストリーミング) は、非常に大きな単一イベントログファイルを生成する可能性があり、保守に多大なコストがかかり、Spark History Server での各更新ごとに大量のリソースを必要とします。

spark.eventLog.rolling.enabled と spark.eventLog.rolling.maxFileSize を有効にすると、単一の巨大なイベントログファイルではなく、ローリングイベントログファイルを使用できるようになります。これは、それ自体で一部のシナリオに役立つ可能性がありますが、ログの全体的なサイズを削減するのに役立つわけではありません。

Spark History Server は、Spark History Server で設定 spark.history.fs.eventLog.rolling.maxFilesToRetain を設定することにより、ローリングイベントログファイルをコンパクションしてログの全体的なサイズを削減できます。

詳細は後述しますが、コンパクションは損失を伴う操作であることに注意してください。コンパクションは、UI に表示されなくなる一部のイベントを破棄します。オプションを有効にする前に、どのイベントが破棄されるかを確認することをお勧めします。

コンパクションが発生すると、履歴サーバーはアプリケーションで利用可能なすべてのイベントログファイルを一覧表示し、保持される最小インデックスを持つファイルよりも小さいインデックスを持つイベントログファイルをコンパクションの対象と見なします。たとえば、アプリケーション A に 5 つのイベントログファイルがあり、spark.history.fs.eventLog.rolling.maxFilesToRetain が 2 に設定されている場合、最初の 3 つのログファイルがコンパクションのために選択されます。

対象を選択したら、それらを分析して除外できるイベントを特定し、除外が決定されたイベントを破棄して 1 つのコンパクトなファイルに書き換えます。

コンパクションは、古いデータを示すイベントを除外しようとします。現在、除外されるイベントの候補は次のとおりです。

• 完了したジョブおよび関連するステージ/タスクのイベント
• 終了した Executor のイベント
• 完了した SQL 実行および関連するジョブ/ステージ/タスクのイベント

書き換えが完了すると、元のログファイルは最善の努力で削除されます。履歴サーバーは元のログファイルを削除できない場合がありますが、履歴サーバーの操作には影響しません。

履歴サーバーは、コンパクション中に削減されるスペースが少ないと判断した場合、古いイベントログファイルをコンパクトしない場合があることに注意してください。ストリーミングクエリの場合、通常、各マイクロバッチが 1 つ以上のジョブをトリガーしてすぐに完了するため、コンパクションが実行されると予想されますが、バッチクエリでは多くの場合、コンパクションは実行されません。

また、これは Spark 3.0 で導入された新機能であり、完全に安定していない可能性があることに注意してください。状況によっては、コンパクションが予期しないよりも多くのイベントを除外する可能性があり、履歴サーバーでアプリケーションの UI に問題が発生する可能性があります。注意して使用してください。

Spark History Server の設定オプション

Spark History Server のセキュリティオプションについては、セキュリティページで詳しく説明しています。

これらの UI では、テーブルはヘッダーをクリックすることで並べ替え可能であり、遅いタスク、データスキューなどを簡単に特定できることに注意してください。

注意

1. 履歴サーバーは、完了した Spark ジョブと未完了の Spark ジョブの両方を表示します。アプリケーションが障害後に複数回試行した場合、失敗した試行、進行中の未完了の試行、または最終的な成功した試行が表示されます。

2. 未完了のアプリケーションは断続的にのみ更新されます。更新間の時間は、変更されたファイルをチェックする間隔 (spark.history.fs.update.interval) によって定義されます。大規模なクラスタでは、更新間隔が大きな値に設定されている場合があります。実行中のアプリケーションを表示する実際の方法は、その独自の Web UI を表示することです。

3. 完了したと自身を登録せずに終了したアプリケーションは、実行されていなくても未完了として一覧表示されます。これは、アプリケーションがクラッシュした場合に発生する可能性があります。

4. Spark ジョブの完了を通知する 1 つの方法は、Spark Context を明示的に停止すること (sc.stop())、または Python では with SparkContext() as sc: 構文を使用して Spark Context のセットアップとティアダウンを処理することです。

REST API

UI でメトリクスを表示するのに加えて、JSON としても利用できます。これにより、開発者は Spark 用の新しいビジュアライゼーションと監視ツールを簡単に作成できます。JSON は、実行中のアプリケーションと履歴サーバーの両方で利用できます。エンドポイントは /api/v1 にマウントされています。たとえば、履歴サーバーの場合、通常は http://<server-url>:18080/api/v1 でアクセスでき、実行中のアプリケーションの場合は https://:4040/api/v1 でアクセスできます。

API では、アプリケーションはアプリケーション ID [app-id] で参照されます。YARN で実行する場合、各アプリケーションには複数の試行がありますが、試行 ID はクラスタモードのアプリケーションにのみ存在し、クライアントモードのアプリケーションには存在しません。YARN クラスタモードのアプリケーションは、その [attempt-id] で識別できます。以下の API では、YARN クラスタモードで実行する場合、[app-id] は実際には [base-app-id]/[attempt-id] になります。ここで、[base-app-id] は YARN アプリケーション ID です。

取得できるジョブとステージの数は、スタンドアロン Spark UI と同じ保持メカニズムによって制限されます。"spark.ui.retainedJobs" はジョブのガベージコレクションをトリガーする閾値を定義し、spark.ui.retainedStages はステージの閾値を定義します。ガベージコレクションは再生時に行われることに注意してください。これらの値を増やして履歴サーバーを再起動することで、より多くのエントリを取得できます。

Executor タスクメトリクス

REST API は、タスク実行の粒度で Spark Executor によって収集されたタスクメトリクスの値を提供します。メトリクスは、パフォーマンスのトラブルシューティングとワークロードの特性評価に使用できます。利用可能なメトリクスのリストと簡単な説明

Executor メトリクス

Executor レベルのメトリクスは、Executor 自体のパフォーマンスメトリクス (JVM ヒープメモリ、GC 情報など) を説明するために、Heartbeat の一部として各 Executor からドライバに送信されます。Executor メトリクス値とその測定されたメモリピーク値は、REST API を介して JSON 形式と Prometheus 形式で公開されます。JSON エンドポイントは /applications/[app-id]/executors で公開され、Prometheus エンドポイントは /metrics/executors/prometheus で公開されます。さらに、Executor メモリメトリクスの集計されたステージごとのピーク値は、spark.eventLog.logStageExecutorMetrics が true の場合にイベントログに書き込まれます。Executor メモリメトリクスは、Dropwizard メトリクスライブラリに基づいた Spark メトリクスシステムを介して公開される場合もあります。利用可能なメトリクスのリストと簡単な説明

RSS と Vmem の計算は proc(5) に基づいています。

API バージョンポリシー

これらのエンドポイントは、アプリケーションを構築しやすくするために、強力にバージョン管理されています。特に、Spark は以下を保証します。

• エンドポイントは、1 つのバージョンから削除されることはありません
• 特定の API エンドポイントの個々のフィールドが削除されることはありません
• 新しいエンドポイントが追加される可能性があります
• 既存のエンドポイントに新しいフィールドが追加される可能性があります
• API の新しいバージョンは、将来、別のエンドポイント (例: api/v2) として追加される可能性があります。新しいバージョンは、後方互換性が保証されている必要はありません。
• API バージョンは、新しい API バージョンと共存してから少なくとも 1 つのマイナーリリース後に削除される可能性があります。

実行中のアプリケーションの UI を確認する場合でも、applications/[app-id] の部分は、アプリケーションが 1 つしかない場合でも必要です。たとえば、実行中のアプリのジョブリストを表示するには、https://:4040/api/v1/applications/[app-id]/jobs にアクセスします。これは、両方のモードでパスを一貫させるためです。

メトリクス

Spark は、Dropwizard Metrics Library に基づいた設定可能なメトリクスシステムを備えています。これにより、ユーザーは Spark メトリクスを HTTP、JMX、CSV ファイルなど、さまざまなシンクにレポートできます。メトリクスは、Spark コードベースに埋め込まれたソースによって生成されます。これらは、特定の活動と Spark コンポーネントのインストルメンテーションを提供します。メトリクスシステムは、Spark が $SPARK_HOME/conf/metrics.properties に存在すると想定する設定ファイルを通じて構成されます。カスタムファイル場所は、spark.metrics.conf 設定プロパティを介して指定できます。設定ファイルを使用する代わりに、spark.metrics.conf. というプレフィックスを持つ一連の設定パラメータを使用できます。デフォルトでは、ドライバまたは Executor メトリクスに使用されるルート名前空間は spark.app.id の値です。しかし、多くの場合、ユーザーはドライバと Executor のアプリ全体でメトリクスを追跡できるようにしたいと考えており、これはアプリの各呼び出しで変更されるため、アプリケーション ID (つまり spark.app.id) では困難です。このようなユースケースでは、spark.metrics.namespace 設定プロパティを使用して、メトリクスレポート用のカスタム名前空間を指定できます。たとえば、ユーザーがメトリクス名前空間をアプリケーションの名前に設定したい場合、spark.metrics.namespace プロパティを ${spark.app.name} のような値に設定できます。この値は、Spark によって適切に展開され、メトリクスシステムのルート名前空間として使用されます。ドライバと Executor 以外のメトリクスは、spark.app.id でプレフィックスが付けられることはありません。また、spark.metrics.namespace プロパティもそのようなメトリクスには影響しません。

Spark のメトリクスは、Spark コンポーネントに対応するさまざまな *インスタンス* に分離されています。各インスタンス内で、メトリクスがレポートされる一連のシンクを構成できます。現在サポートされているインスタンスは次のとおりです。

• master: Spark スタンドアロンマスタープロセス。
• applications: マスター内の、さまざまなアプリケーションをレポートするコンポーネント。
• worker: Spark スタンドアロンワーカープロセス。
• executor: Spark Executor。
• driver: Spark ドライバプロセス (SparkContext が作成されるプロセス)。
• shuffleService: Spark シャッフルサービス。
• applicationMaster: YARN で実行する場合の Spark ApplicationMaster。

各インスタンスは、0 個以上の *シンク* にレポートできます。シンクは org.apache.spark.metrics.sink パッケージに含まれています。

• ConsoleSink: メトリクス情報をコンソールにログ記録します。
• CSVSink: メトリクスデータを定期的に CSV ファイルにエクスポートします。
• JmxSink: JMX コンソールで表示するためのメトリクスを登録します。
• MetricsServlet: 既存の Spark UI 内にサーブレットを追加し、メトリクスデータを JSON データとして提供します。
• PrometheusServlet: (実験的) 既存の Spark UI 内にサーブレットを追加し、メトリクスデータを Prometheus 形式で提供します。
• GraphiteSink: メトリクスを Graphite ノードに送信します。
• Slf4jSink: メトリクスをログエントリとして slf4j に送信します。
• StatsdSink: メトリクスを StatsD ノードに送信します。

Prometheus Servlet は、Metrics Servlet と REST API によって公開される JSON データを時系列形式でミラーリングします。以下は、同等の Prometheus Servlet エンドポイントです。

Spark は、ライセンス上の制限によりデフォルトビルドに含まれていない Ganglia シンクもサポートしています。

• GangliaSink: メトリクスを Ganglia ノードまたはマルチキャストグループに送信します。

GangliaSink をインストールするには、Spark のカスタムビルドを実行する必要があります。このライブラリを埋め込むと、LGPL ライセンスのコードが Spark パッケージに含まれることに注意してください。sbt ユーザーは、ビルド前に SPARK_GANGLIA_LGPL 環境変数を設定します。Maven ユーザーは、-Pspark-ganglia-lgpl プロファイルを有効にします。クラスタの Spark ビルドを変更するだけでなく、ユーザーアプリケーションは spark-ganglia-lgpl アーティファクトにリンクする必要があります。

メトリクス設定ファイルの構文と、各シンクで利用可能なパラメータは、例の設定ファイル $SPARK_HOME/conf/metrics.properties.template で定義されています。

メトリクス設定ファイルではなく Spark 設定パラメータを使用する場合、関連するパラメータ名はプレフィックス spark.metrics.conf. に設定詳細が続くことで構成されます。つまり、パラメータは spark.metrics.conf.[instance|*].sink.[sink_name].[parameter_name] の形式を取ります。この例では、Graphite シンクの Spark 設定パラメータのリストを示します。

"spark.metrics.conf.*.sink.graphite.class"="org.apache.spark.metrics.sink.GraphiteSink"
"spark.metrics.conf.*.sink.graphite.host"="graphiteEndPoint_hostName>"
"spark.metrics.conf.*.sink.graphite.port"=<graphite_listening_port>
"spark.metrics.conf.*.sink.graphite.period"=10
"spark.metrics.conf.*.sink.graphite.unit"=seconds
"spark.metrics.conf.*.sink.graphite.prefix"="optional_prefix"
"spark.metrics.conf.*.sink.graphite.regex"="optional_regex_to_send_matching_metrics"

Spark メトリクス設定のデフォルト値は次のとおりです。

"*.sink.servlet.class" = "org.apache.spark.metrics.sink.MetricsServlet"
"*.sink.servlet.path" = "/metrics/json"
"master.sink.servlet.path" = "/metrics/master/json"
"applications.sink.servlet.path" = "/metrics/applications/json"

追加のソースは、メトリクス設定ファイルまたは設定パラメータ spark.metrics.conf.[component_name].source.jvm.class=[source_name] を使用して構成できます。現在、JVM ソースは唯一利用可能なオプションソースです。たとえば、次の設定パラメータは JVM ソースをアクティブにします: "spark.metrics.conf.*.source.jvm.class"="org.apache.spark.metrics.source.JvmSource"

利用可能なメトリクスプロバイダのリスト

Spark で使用されるメトリクスは、ゲージ、カウンター、ヒストグラム、メーター、タイマーなど、複数のタイプがあります。詳細は Dropwizard ライブラリのドキュメントを参照してください。次のコンポーネントとメトリクスのリストは、利用可能なメトリクスの名前と詳細を、コンポーネントインスタンスとソース名前空間ごとにグループ化して報告します。Spark インストルメンテーションで最も一般的に使用されるメトリクスのタイプは、ゲージとカウンターです。カウンターは .count サフィックスがあることで認識できます。タイマー、メーター、ヒストグラムはリストに注釈付けされています。リストの残りの要素はゲージタイプのメトリクスです。メトリクスの大多数は、親コンポーネントインスタンスが構成されるとすぐにアクティブになります。一部のメトリクスは、追加の設定パラメータを介して有効にする必要もあります。詳細はリストに報告されています。

コンポーネントインスタンス = Driver

これは、最も多くのインストルメント化されたメトリクスを持つコンポーネントです。

• namespace=BlockManager
  • disk.diskSpaceUsed_MB
  • memory.maxMem_MB
  • memory.maxOffHeapMem_MB
  • memory.maxOnHeapMem_MB
  • memory.memUsed_MB
  • memory.offHeapMemUsed_MB
  • memory.onHeapMemUsed_MB
  • memory.remainingMem_MB
  • memory.remainingOffHeapMem_MB
  • memory.remainingOnHeapMem_MB
• namespace=HiveExternalCatalog
  • 注意: これらのメトリクスは、設定パラメータ: spark.metrics.staticSources.enabled (デフォルトは true) に条件付きです。
  • fileCacheHits.count
  • filesDiscovered.count
  • hiveClientCalls.count
  • parallelListingJobCount.count
  • partitionsFetched.count
• namespace=CodeGenerator
  • 注意: これらのメトリクスは、設定パラメータ: spark.metrics.staticSources.enabled (デフォルトは true) に条件付きです。
  • compilationTime (histogram)
  • generatedClassSize (histogram)
  • generatedMethodSize (histogram)
  • sourceCodeSize (histogram)
• namespace=DAGScheduler
  • job.activeJobs
  • job.allJobs
  • messageProcessingTime (timer)
  • stage.failedStages
  • stage.runningStages
  • stage.waitingStages
• namespace=LiveListenerBus
  • listenerProcessingTime.org.apache.spark.HeartbeatReceiver (timer)
  • listenerProcessingTime.org.apache.spark.scheduler.EventLoggingListener (timer)
  • listenerProcessingTime.org.apache.spark.status.AppStatusListener (timer)
  • numEventsPosted.count
  • queue.appStatus.listenerProcessingTime (timer)
  • queue.appStatus.numDroppedEvents.count
  • queue.appStatus.size
  • queue.eventLog.listenerProcessingTime (timer)
  • queue.eventLog.numDroppedEvents.count
  • queue.eventLog.size
  • queue.executorManagement.listenerProcessingTime (timer)
• namespace=appStatus (すべてのメトリクスタイプ=counter)
  • 注意: Spark 3.0 で導入されました。設定パラメータ: spark.metrics.appStatusSource.enabled (デフォルトは true) に条件付き
  • stages.failedStages.count
  • stages.skippedStages.count
  • stages.completedStages.count
  • tasks.blackListedExecutors.count // 非推奨、excludedExecutors を使用
  • tasks.excludedExecutors.count
  • tasks.completedTasks.count
  • tasks.failedTasks.count
  • tasks.killedTasks.count
  • tasks.skippedTasks.count
  • tasks.unblackListedExecutors.count // 非推奨、unexcludedExecutors を使用
  • tasks.unexcludedExecutors.count
  • jobs.succeededJobs
  • jobs.failedJobs
  • jobDuration
• namespace=AccumulatorSource
  • 注意: アキュムレータをメトリクスシステムにアタッチするための、ユーザー提供のソース。
  • DoubleAccumulatorSource
  • LongAccumulatorSource
• namespace=spark.streaming
  • 注意: これは Spark Structured Streaming にのみ適用されます。設定パラメータ: spark.sql.streaming.metricsEnabled=true (デフォルトは false) に条件付き
  • eventTime-watermark
  • inputRate-total
  • latency
  • processingRate-total
  • states-rowsTotal
  • states-usedBytes
• namespace=JVMCPU
  • jvmCpuTime
• namespace=executor
  • 注意: これらのメトリクスは、ローカルモードでのみドライバで利用できます。
  • この名前空間で利用可能なメトリクスの完全なリストは、Executor コンポーネントインスタンスに対応するエントリで確認できます。
• namespace=ExecutorMetrics
  • 注意: これらのメトリクスは、設定パラメータ: spark.metrics.executorMetricsSource.enabled (デフォルトは true) に条件付きです。
  • このソースにはメモリ関連のメトリクスが含まれています。この名前空間で利用可能なメトリクスの完全なリストは、Executor コンポーネントインスタンスに対応するエントリで確認できます。
• namespace=ExecutorAllocationManager
  • 注意: これらのメトリクスは、動的割り当てを使用している場合にのみ発行されます。設定パラメータ spark.dynamicAllocation.enabled (デフォルトは false) に条件付き。
  • executors.numberExecutorsToAdd
  • executors.numberExecutorsPendingToRemove
  • executors.numberAllExecutors
  • executors.numberTargetExecutors
  • executors.numberMaxNeededExecutors
  • executors.numberDecommissioningExecutors
  • executors.numberExecutorsGracefullyDecommissioned.count
  • executors.numberExecutorsDecommissionUnfinished.count
  • executors.numberExecutorsExitedUnexpectedly.count
  • executors.numberExecutorsKilledByDriver.count
• namespace=plugin.<Plugin Class Name>
  • オプションの名前空間。この名前空間のメトリクスは、ユーザー提供のコードによって定義され、Spark プラグイン API を使用して構成されます。「高度なインストルメンテーション」セクションで、カスタムプラグインを Spark にロードする方法を参照してください。

コンポーネントインスタンス = Executor

これらのメトリクスは Spark Executor によって公開されます。

• namespace=executor (メトリクスは counter または gauge タイプ)
  • 注記
    • spark.executor.metrics.fileSystemSchemes (デフォルト: file,hdfs) は、公開されるファイルシステムメトリクスを決定します。
  • bytesRead.count
  • bytesWritten.count
  • cpuTime.count
  • deserializeCpuTime.count
  • deserializeTime.count
  • diskBytesSpilled.count
  • filesystem.file.largeRead_ops
  • filesystem.file.read_bytes
  • filesystem.file.read_ops
  • filesystem.file.write_bytes
  • filesystem.file.write_ops
  • filesystem.hdfs.largeRead_ops
  • filesystem.hdfs.read_bytes
  • filesystem.hdfs.read_ops
  • filesystem.hdfs.write_bytes
  • filesystem.hdfs.write_ops
  • jvmGCTime.count
  • memoryBytesSpilled.count
  • recordsRead.count
  • recordsWritten.count
  • resultSerializationTime.count
  • resultSize.count
  • runTime.count
  • shuffleBytesWritten.count
  • shuffleFetchWaitTime.count
  • shuffleLocalBlocksFetched.count
  • shuffleLocalBytesRead.count
  • shuffleRecordsRead.count
  • shuffleRecordsWritten.count
  • shuffleRemoteBlocksFetched.count
  • shuffleRemoteBytesRead.count
  • shuffleRemoteBytesReadToDisk.count
  • shuffleTotalBytesRead.count
  • shuffleWriteTime.count
  • プッシュベースシャッフル関連メトリクス
    • shuffleCorruptMergedBlockChunks
    • shuffleMergedFetchFallbackCount
    • shuffleMergedRemoteBlocksFetched
    • shuffleMergedLocalBlocksFetched
    • shuffleMergedRemoteChunksFetched
    • shuffleMergedLocalChunksFetched
    • shuffleMergedRemoteBytesRead
    • shuffleMergedLocalBytesRead
    • shuffleRemoteReqsDuration
    • shuffleMergedRemoteReqsDuration
  • succeededTasks.count
  • threadpool.activeTasks
  • threadpool.completeTasks
  • threadpool.currentPool_size
  • threadpool.maxPool_size
  • threadpool.startedTasks
• namespace=ExecutorMetrics
  • 注記
    • これらのメトリクスは、設定パラメータ: spark.metrics.executorMetricsSource.enabled (デフォルト値は true) に条件付きです。
    • ExecutorMetrics は、Executor およびドライバのためにスケジュールされたハートビートプロセスの一部として定期的に更新されます: spark.executor.heartbeatInterval (デフォルト値は 10 秒)。
    • Executor メモリメトリクス用のオプションの高速ポーリングメカニズムが利用可能です。これは、設定パラメータ spark.executor.metrics.pollingInterval を使用してポーリング間隔 (ミリ秒単位) を設定することでアクティブ化できます。
  • JVMHeapMemory
  • JVMOffHeapMemory
  • OnHeapExecutionMemory
  • OnHeapStorageMemory
  • OnHeapUnifiedMemory
  • OffHeapExecutionMemory
  • OffHeapStorageMemory
  • OffHeapUnifiedMemory
  • DirectPoolMemory
  • MappedPoolMemory
  • MinorGCCount
  • MinorGCTime
  • MajorGCCount
  • MajorGCTime
  • 「ProcessTree*」メトリクスカウンター
    • ProcessTreeJVMVMemory
    • ProcessTreeJVMRSSMemory
    • ProcessTreePythonVMemory
    • ProcessTreePythonRSSMemory
    • ProcessTreeOtherVMemory
    • ProcessTreeOtherRSSMemory
    • 注意: 「ProcessTree*」メトリクスは、特定の条件下でのみ収集されます。条件は、/proc ファイルシステムが存在し、spark.executor.processTreeMetrics.enabled=true であることの論理 AND です。「ProcessTree*」メトリクスは、これらの条件が満たされない場合は 0 を返します。
• namespace=JVMCPU
  • jvmCpuTime
• namespace=NettyBlockTransfer
  • shuffle-client.usedDirectMemory
  • shuffle-client.usedHeapMemory
  • shuffle-server.usedDirectMemory
  • shuffle-server.usedHeapMemory
• namespace=HiveExternalCatalog
  • 注意: これらのメトリクスは、設定パラメータ: spark.metrics.staticSources.enabled (デフォルトは true) に条件付きです。
  • fileCacheHits.count
  • filesDiscovered.count
  • hiveClientCalls.count
  • parallelListingJobCount.count
  • partitionsFetched.count
• namespace=CodeGenerator
  • 注意: これらのメトリクスは、設定パラメータ: spark.metrics.staticSources.enabled (デフォルトは true) に条件付きです。
  • compilationTime (histogram)
  • generatedClassSize (histogram)
  • generatedMethodSize (histogram)
  • sourceCodeSize (histogram)
• namespace=plugin.<Plugin Class Name>
  • オプションの名前空間。この名前空間のメトリクスは、ユーザー提供のコードによって定義され、Spark プラグイン API を使用して構成されます。「高度なインストルメンテーション」セクションで、カスタムプラグインを Spark にロードする方法を参照してください。

ソース = JVMソース

注記

• 関連する metrics.properties ファイルのエントリまたは設定パラメータ: spark.metrics.conf.*.source.jvm.class=org.apache.spark.metrics.source.JvmSource を設定して、このソースをアクティブ化します。
• これらのメトリクスは、設定パラメータ: spark.metrics.staticSources.enabled (デフォルトは true) に条件付きです。
• このソースは、ドライバおよび Executor インスタンスで利用可能であり、他のインスタンスでも利用可能です。
• このソースは、Dropwizard/Codahale Metric Sets for JVM instrumentation を使用して JVM メトリクスに関する情報を提供し、特に BufferPoolMetricSet、GarbageCollectorMetricSet、MemoryUsageGaugeSet のメトリクスセットを提供します。

コンポーネントインスタンス = applicationMaster

注意: YARN で実行する場合に適用されます。

• numContainersPendingAllocate
• numExecutorsFailed
• numExecutorsRunning
• numLocalityAwareTasks
• numReleasedContainers

コンポーネントインスタンス = master

注意: Spark スタンドアロンのマスターとして実行する場合に適用されます。

• workers
• aliveWorkers
• apps
• waitingApps

コンポーネントインスタンス = ApplicationSource

注意: Spark スタンドアロンのマスターとして実行する場合に適用されます。

• status
• runtime_ms
• cores

コンポーネントインスタンス = worker

注意: Spark スタンドアロンのワーカーとして実行する場合に適用されます。

• executors
• coresUsed
• memUsed_MB
• coresFree
• memFree_MB

コンポーネントインスタンス = shuffleService

注意: シャッフルサービスに適用されます。

• blockTransferRate (meter) - ブロック転送のレート
• blockTransferMessageRate (meter) - ブロック転送メッセージのレート。つまり、バッチフェッチが有効になっている場合、これはブロック数ではなくバッチ数を表します。
• blockTransferRateBytes (meter)
• blockTransferAvgSize_1min (gauge - 1分移動平均)
• numActiveConnections.count
• numRegisteredConnections.count
• numCaughtExceptions.count
• openBlockRequestLatencyMillis (timer)
• registerExecutorRequestLatencyMillis (timer)
• fetchMergedBlocksMetaLatencyMillis (timer)
• finalizeShuffleMergeLatencyMillis (timer)
• registeredExecutorsSize
• shuffle-server.usedDirectMemory

• shuffle-server.usedHeapMemory

• 注意: 以下のメトリクスは、サーバー側設定 spark.shuffle.push.server.mergedShuffleFileManagerImpl が Push-Based Shuffle のために org.apache.spark.network.shuffle.MergedShuffleFileManager に設定されている場合に適用されます。
• blockBytesWritten - プッシュされたブロックデータがファイルに書き込まれたサイズ (バイト単位)
• blockAppendCollisions - シャッフルプッシュブロックがシャッフルサービスで衝突した回数。同じ削減パーティションの別のブロックが書き込まれていた場合。
• lateBlockPushes - 特定のシャッフルマージが完了した後にシャッフルサービスで受信されたシャッフルプッシュブロックの数
• deferredBlocks - メモリにバッファリングされている現在の遅延ブロック部分の数
• deferredBlockBytes - メモリにバッファリングされている現在の遅延ブロック部分のサイズ
• staleBlockPushes - 古いシャッフルブロックプッシュ要求の数
• ignoredBlockBytes - ESS に転送されたが無視されたプッシュブロックデータのサイズ。プッシュブロックデータは、次の場合に無視されたと見なされます: 1. シャッフルが完了した後に受信された場合。2. プッシュ要求が重複ブロックのものである場合。3. ESS がブロックを書き込めなかった場合。

高度なインストルメンテーション

Spark ジョブのパフォーマンスをプロファイリングするのに役立ついくつかの外部ツールがあります。

• Ganglia のようなクラスタ全体の監視ツールは、全体的なクラスタの使用状況とリソースのボトルネックに関する洞察を提供できます。たとえば、Ganglia ダッシュボードは、特定のワークロードがディスクバウンド、ネットワークバウンド、または CPU バウンドであるかを迅速に明らかにできます。
• dstat、iostat、iotop のような OS プロファイリングツールは、個々のノードで詳細なプロファイリングを提供できます。
• スタックトレースを提供する jstack、ヒープダンプを作成するための jmap、時系列統計を報告するための jstat、およびさまざまな JVM プロパティを視覚的に探索するための jconsole のような JVM ユーティリティは、JVM の内部に精通しているユーザーにとって役立ちます。

Spark はプラグイン API も提供しているため、カスタムインストルメンテーションコードを Spark アプリケーションに追加できます。Spark にプラグインをロードするために利用できる 2 つの設定キーがあります。

• spark.plugins
• spark.plugins.defaultList

どちらも、org.apache.spark.api.plugin.SparkPlugin インターフェイスを実装するクラス名のカンマ区切りリストを受け取ります。2 つの名前が存在するのは、1 つのリストを Spark のデフォルト設定ファイルに配置できるようにするためです。これにより、ユーザーは設定ファイルのリストを上書きすることなく、コマンドラインから簡単に他のプラグインを追加できます。重複するプラグインは無視されます。