監視および計測器

• Web インタフェース
  • 後で見る
    • 環境変数
    • ローリングイベントログファイルに圧縮を適用
    • Spark 履歴サーバの設定オプション
  • REST API
    • Executorのタスクのメトリクス
    • Executor のメトリクス
    • API バージョニング ポリシー
• マトリックス
  • 利用可能なメトリクスプロバイダのリスト
    • Component instance = Driver
    • Component instance = Executor
    • Source = JVM Source
    • Component instance = applicationMaster
    • Component instance = mesos_cluster
    • Component instance = master
    • Component instance = ApplicationSource
    • Component instance = worker
    • Component instance = shuffleService
• 上級の計測器

Sparkアプリケーションを監視するいくつかの方法があります: web UI、マトリクス、および実験的な計測器。

Web インタフェース

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

• スケジューラのステージおよびタスクのリスト
• RDDサイズおよびメモリの使用量の概要
• 環境情報
• 実行中のexecutorの情報

単にwebブラウザでhttp://<driver-node>:4040を開くことで、このインタフェースにアクセスできます。同じホスト上で複数のSparkContextが実行中の場合、それらは4040から始まる連続したポートに割り当てられるでしょう(4041, 4042など)。

デフォルトではこの情報はアプリケーションの持続期間のみ利用可能です。後でwebUIを見るためには、アプリケーションの開始前にspark.eventLog.enabled をtrueに設定します。これはUIに表示された情報を符号化したSparkイベントを永続的なストレージに記録するように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

履歴サーバは以下のようにして設定することができます:

環境変数

ローリングイベントログファイルに圧縮を適用

実行時間の長いアプリケーション(例えばストリーミング)は、1つの巨大なエベントログファイルになる可能性があり、維持にコストが掛かり、また Spark履歴サーバで更新ごとに再生するために大量のリソースが必要になります。

spark.eventLog.rolling.enabled と spark.eventLog.rolling.maxFileSize を有効にすると、幾つかのシナリオで役立つが、それでもログの全体的なサイズを減らすのには役に立たないような、1つの巨大なイベントログファイルではなく、ローリングイベントログファイルを使用できるようになります。

Spark履歴サーバは、Spark履歴サーバ上のspark.history.fs.eventLog.rolling.maxFilesToRetain設定をすることで、ローリングイベントファイルに圧縮を適用してログの全体的なサイズを削減することができます。

詳細は以下で説明しますが、圧縮は損失のある操作であることに先に注意してください。圧縮により UI に表示されなくなるイベントが破棄されます - オプションを有効にする前に破棄されるイベントを確認することをお勧めします。

圧縮が発生すると、履歴サーバはアプリケーションで利用可能な全てのイベントログファイルを列挙し、圧縮のターゲットとして保持される最小のインデックスを持つファイルよりもインデックスが少ないイベントログファイルになるように考慮します。例えば、アプリケーション A に5つのイベントログがあり、spark.history.fs.eventLog.rolling.maxFilesToRetain が 2 に設定された場合、最初の3つのログファイルが圧縮対象として選択されます。

ターゲットを選択すると、それらを分析して除外できるイベントを特定し、除外することに決定したイベントを破棄して1つの圧縮ファイルに書きなおします。

圧縮は古いデータを指すイベントを除外しようとします。現在、除外するイベントの候補を以下に示します:

• 終了したジョブのイベントと、関連するステージ/タスク イベント
• 終了した executor のイベント
• 終了した SQL 実行のイベントと、関連するジョブ/ステージ/タスク イベント。

書き換えが完了すると、元のログファイルはベストエフォートの方式で削除されます。履歴サーバは元のファイルを削除できない場合がありますが、それは履歴サーバの操作には影響しません。

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

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

Spark 履歴サーバの設定オプション

Spark Historyサーバのためのセキュリティ オプションはセキュリティのページの中でより詳細にカバーされます。

これら全てのUI内でテーブルはヘッダーをクリックすることでソート可能です。遅いタスク、データのゆがみなどを識別するのを容易にします。

注意

1. 履歴サーバは完了および未完了のSparkジョブを表示します。障害の後でアプリケーションが複数回の試行をする場合、外に出ていく全ての未完了の試行あるいは最後に成功した試行と共に、失敗した試行が表示されるでしょう。

2. 未完了のアプリケーションは断続的に更新されるだけです。更新の間の時間は変更されたファイルのチェックの間隔によって定義されます(spark.history.fs.update.interval)。大きなクラスターほど更新の間隔は大きな値に設定されるかも知れません。実行中のアプリケーションを見る方法は実際には自身のweb UIを見ることです。

3. 完了したと自身を登録しないで終了したアプリケーションは、未完了としてリスト化されるでしょう - それらがもう実行していないとしてもです。これはアプリケーションがクラッシュした場合に起こりえます。

4. Sparkジョブの完了の信号を送る1つの方法はSparkのContextを明示的に終了することです (sc.stop())。あるいは、Pythonではwith SparkContext() as sc: を使ってSparkのContextのsteupとtear downを処理するために構築します。

REST API

UI内でマトリックスを見ることに加えて、それらはJSONとして利用可能です。これにより開発者は新しい表示方法およびSparkのための監視ツールを簡単に作成することができます。JSONは実行中のアプリケーションおよび履歴サーバ内の両方で利用可能です。エンドポイントは /api/v1にマウントされます。例えば、履歴サーバの場合、それらは一般的にhttp://<server-url>:18080/api/v1でアクセスすることができ、実行中のアプリケーションの場合http://localhost:4040/api/v1で利用可能です。

APIの中では、アプリケーションはアプリケーションID [app-id]によって参照されます。YARN上で実行中の場合、各アプリケーションは複数の試行を持つかも知れませんが、クライアントモードでのアプリケーションと異なり、クラスタモードでのアプリケーションの試行IDです。YARNクラスタモードのアプリケーションはそれらの [attempt-id]によって識別することができます。以下でリスト化されているAPIでは、[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 レべルのメトリクスは Heartbeat の一部として各 executor からドライバに送信され、JVM ヒープメモリや GC 情報などの executor 自身のパフォーマンスメトリクスを説明します。executor メトリクスの値と executor ごとの測定されたメモリピーク値は、REST API を使って JSON 形式および Prometheus 形式で公開されます。JSON のエンドポイントは、以下で公開されます: /applications/[app-id]/executors。そして、Prometheusのエンドポイントは以下で公開されます: /metrics/executors/prometheus。The Prometheus endpoint is conditional to a configuration parameter: spark.ui.prometheus.enabled=true (the default is false). さらに、spark.eventLog.logStageExecutorMetrics が true の場合、executor メモリメトリクスの集計されたステージごとのピーク値がイベントログに書き込まれます。
executorのメモリメトリクスもDropwizard metrics libraryに基づくSparkメトリクスシステムを介して公開されます。簡単な説明と利用可能なメトリクスのリスト:

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

API バージョニング ポリシー

これらのエンドポイントはその上でアプリケーションを開発を容易にするために熱心にバージョン付けされています。特に、Sparkは以下を保証します:

• エンドポイントは一つのバージョンからは削除されることはないでしょう
• 各フィールドは全ての指定されたエンドポイントにおいて削除されないでしょう
• 新しいエンドポイントが追加されるかも知れません
• 新しいフィールドが既存のエンドポイントに追加されるかも知れません
• APIの新しいバージョンは将来別個のエンドポイント(例えば、api/v2)に追加されるかも知れません。新しいバージョンは後方互換性が必要とされません。
• APIのバージョンは削除されるかも知れませんが、新しいapiバージョンの少なくとも一つの既存のマイナーバージョンがあります。

実行中のアプリケーションのUIを検証する時でも、applications/[app-id]の分割はたった一つのアプリケーションだけが利用可能だとしてもまだ必要です。例えば、実行中のアプリケーションのジョブのリストを見るには、http://localhost:4040/api/v1/applications/[app-id]/jobsに行きます。これは両方のモードにおいてパスを矛盾が無い状態にします。

マトリックス

SparkはDropwizard Metrics Libraryに基づいた設定可能なマトリックスシステムを持っています。これにより、ユーザはSparkマトリックスにHTTP, JMX およびCSV ファイルを含む様々なsinkを伝えることができます。メトリクスは 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コンポーネントに対応する異なる インスタンス に分離されます。各インスタンスの中で、マトリックが伝える一連のsinkを設定することができます。現在のところ以下のインスタンスがサポートされています:

• master: Spark スタンドアローンマスタープロセス。
• applications: 様々なアプリケーション上で報告するマスター内のコンポーネント。
• worker: Spark スタンドアローン ワーカープロセス。
• executor: Sparkの executor。
• driver: Spark ドライバープロセス (SparkContextが生成されるプロセス)。
• shuffleService: Sparkのシャッフルサービス。
• applicationMaster: YARN上で実行する時のSparkアプリケーションマスタ。
• mesos_cluster: Mesos で実行中の場合、Spark クラスタスケジューラ。

各インスタンスは0個以上のsinksに対して報告することができます。sinkはorg.apache.spark.metrics.sink パッケージに含まれています。

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

Sparkはライセンスの制限のためにデフォルトのビルドに含まれていないGanglia sinkもサポートします:

• GangliaSink: マトリックスをGangliaノードあるいはマルチキャストグループに送信する。

GangliaSinkをインストールするには、Sparkのカスタムビルドを実行する必要があるでしょう。このライブラリを組み込むことでSparkパッケージにLGPLでライセンスされたコードが含まれることに注意してください。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 サフィックスを持つものとして認識されます。タイマー、メータおよびヒストグラムはリストで注釈を付けられ、リストの残りの要素はゲージタイプのメトリクスです。メトリクスの大部分は親コンポーネントインスタンスが設定されるとすぐにアクティブになります。一部のメトリクスは追加の設定パラメータを介して有効にする必要もあります。詳細はリストの中で報告されます。

Component instance = Driver

This is the component with the largest amount of instrumented metrics

• 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
  • note: these metrics are conditional to a configuration parameter: spark.metrics.staticSources.enabled (default is true)
  • fileCacheHits.count
  • filesDiscovered.count
  • hiveClientCalls.count
  • parallelListingJobCount.count
  • partitionsFetched.count
• namespace=CodeGenerator
  • note: these metrics are conditional to a configuration parameter: spark.metrics.staticSources.enabled (default is 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 (all metrics of type=counter)
  • note: Introduced in Spark 3.0. Conditional to a configuration parameter:
    spark.metrics.appStatusSource.enabled (default is false)
  • stages.failedStages.count
  • stages.skippedStages.count
  • stages.completedStages.count
  • tasks.blackListedExecutors.count // deprecated use excludedExecutors instead
  • tasks.excludedExecutors.count
  • tasks.completedTasks.count
  • tasks.failedTasks.count
  • tasks.killedTasks.count
  • tasks.skippedTasks.count
  • tasks.unblackListedExecutors.count // deprecated use unexcludedExecutors instead
  • tasks.unexcludedExecutors.count
  • jobs.succeededJobs
  • jobs.failedJobs
  • jobDuration
• namespace=AccumulatorSource
  • note: User-configurable sources to attach accumulators to metric system
  • DoubleAccumulatorSource
  • LongAccumulatorSource
• namespace=spark.streaming
  • 注意: これはSpark 構造化ストリーミングのみに適用されます。Conditional to a configuration parameter: spark.sql.streaming.metricsEnabled=true (default is false)
  • eventTime-watermark
  • inputRate-total
  • latency
  • processingRate-total
  • states-rowsTotal
  • states-usedBytes
• namespace=JVMCPU
  • jvmCpuTime
• namespace=executor
  • 注意: これらのメトリクスはローカルモードのドライバのみで利用可能です。
  • この名前空間で利用可能なメトリクスの完全なリストは、executorコンポーネントインスタンスの対応するエントリで見つかります。
• namespace=ExecutorMetrics
  • note: these metrics are conditional to a configuration parameter: spark.metrics.executorMetricsSource.enabled (default is true)
  • このソースはメモリに関係するメトリクスを含みます。この名前空間で利用可能なメトリクスの完全なリストは、executorコンポーネントインスタンスの対応するエントリで見つかります。
• namespace=ExecutorAllocationManager
  • 注意: これらのメトリクスは動的割り当てを使っている場合のみ発行されます。Conditional to a configuration parameter spark.dynamicAllocation.enabled (default is false)
  • executors.numberExecutorsToAdd
  • executors.numberExecutorsPendingToRemove
  • executors.numberAllExecutors
  • executors.numberTargetExecutors
  • executors.numberMaxNeededExecutors
  • executors.numberExecutorsGracefullyDecommissioned.count
  • executors.numberExecutorsDecommissionUnfinished.count
  • executors.numberExecutorsExitedUnexpectedly.count
  • executors.numberExecutorsKilledByDriver.count
• namespace=plugin.<Plugin Class Name>
  • Optional namespace(s). この名前空間内のメトリクスはユーザ定義のコードで定義され、SparkプラグインAPIを使って設定されます。独自のプラグインをSparkにロードする方法については、以下の“Advanced Instrumentation”を見てください。

Component instance = Executor

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

• namespace=executor (metrics are of type counter or gauge)
  • notes:
    • spark.executor.metrics.fileSystemSchemes (default: file,hdfs) determines the exposed file system metrics.
  • 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
  • succeededTasks.count
  • threadpool.activeTasks
  • threadpool.completeTasks
  • threadpool.currentPool_size
  • threadpool.maxPool_size
  • threadpool.startedTasks
• namespace=ExecutorMetrics
  • notes:
    • These metrics are conditional to a configuration parameter: spark.metrics.executorMetricsSource.enabled (default value is true)
    • ExecutorMetrics are updated as part of heartbeat processes scheduled for the executors and for the driver at regular intervals: spark.executor.heartbeatInterval (default value is 10 seconds)
    • An optional faster polling mechanism is available for executor memory metrics, it can be activated by setting a polling interval (in milliseconds) using the configuration parameter spark.executor.metrics.pollingInterval
  • JVMHeapMemory
  • JVMOffHeapMemory
  • OnHeapExecutionMemory
  • OnHeapStorageMemory
  • OnHeapUnifiedMemory
  • OffHeapExecutionMemory
  • OffHeapStorageMemory
  • OffHeapUnifiedMemory
  • DirectPoolMemory
  • MappedPoolMemory
  • MinorGCCount
  • MinorGCTime
  • MajorGCCount
  • MajorGCTime
  • “ProcessTree*” metric counters:
    • ProcessTreeJVMVMemory
    • ProcessTreeJVMRSSMemory
    • ProcessTreePythonVMemory
    • ProcessTreePythonRSSMemory
    • ProcessTreeOtherVMemory
    • ProcessTreeOtherRSSMemory
    • note: “ProcessTree” metrics are collected only under certain conditions. The conditions are the logical AND of the following: /proc filesystem exists, spark.executor.processTreeMetrics.enabled=true. “ProcessTree” metrics report 0 when those conditions are not met.
• namespace=JVMCPU
  • jvmCpuTime
• namespace=NettyBlockTransfer
  • shuffle-client.usedDirectMemory
  • shuffle-client.usedHeapMemory
  • shuffle-server.usedDirectMemory
  • shuffle-server.usedHeapMemory
• namespace=HiveExternalCatalog
  • note: these metrics are conditional to a configuration parameter: spark.metrics.staticSources.enabled (default is true)
  • fileCacheHits.count
  • filesDiscovered.count
  • hiveClientCalls.count
  • parallelListingJobCount.count
  • partitionsFetched.count
• namespace=CodeGenerator
  • note: these metrics are conditional to a configuration parameter: spark.metrics.staticSources.enabled (default is true)
  • compilationTime (histogram)
  • generatedClassSize (histogram)
  • generatedMethodSize (histogram)
  • sourceCodeSize (histogram)
• namespace=plugin.<Plugin Class Name>
  • Optional namespace(s). この名前空間内のメトリクスはユーザ定義のコードで定義され、SparkプラグインAPIを使って設定されます。独自のプラグインをSparkにロードする方法については、以下の“Advanced Instrumentation”を見てください。

Source = JVM Source

注意:

• Activate this source by setting the relevant metrics.properties file entry or the configuration parameter:spark.metrics.conf.*.source.jvm.class=org.apache.spark.metrics.source.JvmSource
• These metrics are conditional to a configuration parameter: spark.metrics.staticSources.enabled (default is true)
• This source is available for driver and executor instances and is also available for other instances.
• This source provides information on JVM metrics using the Dropwizard/Codahale Metric Sets for JVM instrumentation and in particular the metric sets BufferPoolMetricSet, GarbageCollectorMetricSet and MemoryUsageGaugeSet.

Component instance = applicationMaster

Note: applies when running on YARN

• numContainersPendingAllocate
• numExecutorsFailed
• numExecutorsRunning
• numLocalityAwareTasks
• numReleasedContainers

Component instance = mesos_cluster

Note: applies when running on mesos

• waitingDrivers
• launchedDrivers
• retryDrivers

Component instance = master

Note: applies when running in Spark standalone as master

• ワーカー
• aliveWorkers
• apps
• waitingApps

Component instance = ApplicationSource

Note: applies when running in Spark standalone as master

• 状態
• runtime_ms
• cores

Component instance = worker

Note: applies when running in Spark standalone as worker

• executors
• coresUsed
• memUsed_MB
• coresFree
• memFree_MB

Component instance = shuffleService

Note: applies to the shuffle service

• blockTransferRate (meter) - rate of blocks being transferred
• blockTransferMessageRate (meter) - rate of block transfer messages, i.e. if batch fetches are enabled, this represents number of batches rather than number of blocks
• blockTransferRateBytes (meter)
• blockTransferAvgTime_1min (gauge - 1-minute moving average)
• numActiveConnections.count
• numRegisteredConnections.count
• numCaughtExceptions.count
• openBlockRequestLatencyMillis (histogram)
• registerExecutorRequestLatencyMillis (histogram)
• registeredExecutorsSize
• shuffle-server.usedDirectMemory
• shuffle-server.usedHeapMemory

上級の計測器

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 のデフォルト設定ファイルに配置できるため、ユーザは設定ファイルのリストを上書きせずにコマンドラインから他のプラグインを簡単に追加することができます。重複するプラグインは無視されます。