Notes: Spark metrics

2022-07-19 13:54:48 浏览数 (1)

Below are some notes taken for future reference based on the brainstorm meeting last week, with company confidential information removed.

Background

The team use a home made workflow to manage the computation for the cost and profit, and there’s a lack of statistics for the jobs and input/output, usually SDE/oncall checks the data in Data Warehouse or TSV files on S3 manually. For EMR jobs, Spark UI and Ganglia are both powerful but when the clusters are terminated, all these valuable metrics data are gone.

Typical use cases:

  • Spark metrics: status / efficiency / executor / GC …
  • EMR cluster / instance metrics: CPU / memory / IO/ network …
  • Workflow: task time cost distribution for each running, time cost comparison between different runnings of a same pipeline and task dependency graph
  • Health status: critical pipeline statuses should be monitored and presented daily, including pipeline status itself and the input / output scale
  • Metrics lib integration: We need a dedicated component to gather metrics from code, this can cover the case that the metrics is tightly connected with the code logic. It’s more like the regular “log”, but cares more about the data on metrics side.
  • (Sev-2) Ticket linkage: display on going ticket information, with related job/components links appended, and component owner should be clearly displayed

I. Metrics for Spark jobs

Metrics when cluster is running – Ganglia and Spark UI, and we can even ssh to the instance to get any information. But the main problem happens when the cluster is terminated:

  1. Coda Hale Metrics Library (preferred)

In latest Spark documents there is an introduction guiding how to integrate the metrics lib. Spark has a configurable metrics system based on the Coda Hale Metrics Library (link1, link2). This allows users to report Spark metrics to a variety of sinks including HTTP, JMX, and CSV files:

  • configuration: $SPARK_HOME/conf/metrics.properties or spark.metrics.conf
  • supported instances: master / applications / worker / executor / driver
  • sinks: ConsoleSink / CSVSink / JmxSink / MetricsServlet / GraphiteSink / Slf4jSink and GangliaSink (needs custom build for license)

We can also implement an S3Sink to upload the metrics data to S3. Or, call MetricsServlet to get metrics data termly and transfer the data to a data visualization system. To get the data when a Spark job is running, besides checking the Spark UI, there’s also a group of RESTful API to make full use of, which is good for long term metrics persistence and visualization.

Example: Console Sink, simply output the metrics like DAG schedule to console:

代码语言：javascript复制
-- Gauges ----------------------------------------------------------------------
DAGScheduler.job.activeJobs
             value = 0
DAGScheduler.job.allJobs
             value = 0
DAGScheduler.stage.failedStages
             value = 0
DAGScheduler.stage.runningStages
             value = 0
DAGScheduler.stage.waitingStages
             value = 0
application_1456611008120_0001.driver.BlockManager.disk.diskSpaceUsed_MB
             value = 0
application_1456611008120_0001.driver.BlockManager.memory.maxMem_MB
             value = 14696
application_1456611008120_0001.driver.BlockManager.memory.memUsed_MB
             value = 0
application_1456611008120_0001.driver.BlockManager.memory.remainingMem_MB
             value = 14696
application_1456611008120_0001.driver.jvm.ConcurrentMarkSweep.count
             value = 1
application_1456611008120_0001.driver.jvm.ConcurrentMarkSweep.time
             value = 70
application_1456611008120_0001.driver.jvm.ParNew.count
             value = 3
application_1456611008120_0001.driver.jvm.ParNew.time
             value = 95
application_1456611008120_0001.driver.jvm.heap.committed
             value = 1037959168
application_1456611008120_0001.driver.jvm.heap.init
             value = 1073741824
application_1456611008120_0001.driver.jvm.heap.max
             value = 1037959168
application_1456611008120_0001.driver.jvm.heap.usage
             value = 0.08597782528570527
application_1456611008120_0001.driver.jvm.heap.used
             value = 89241472
application_1456611008120_0001.driver.jvm.non-heap.committed
             value = 65675264
application_1456611008120_0001.driver.jvm.non-heap.init
             value = 24313856
application_1456611008120_0001.driver.jvm.non-heap.max
             value = 587202560
application_1456611008120_0001.driver.jvm.non-heap.usage
             value = 0.1083536148071289
application_1456611008120_0001.driver.jvm.non-heap.used
             value = 63626104
application_1456611008120_0001.driver.jvm.pools.CMS-Old-Gen.committed
             value = 715849728
application_1456611008120_0001.driver.jvm.pools.CMS-Old-Gen.init
             value = 715849728
application_1456611008120_0001.driver.jvm.pools.CMS-Old-Gen.max
             value = 715849728
application_1456611008120_0001.driver.jvm.pools.CMS-Old-Gen.usage
             value = 0.012926198946603497
application_1456611008120_0001.driver.jvm.pools.CMS-Old-Gen.used
             value = 9253216
application_1456611008120_0001.driver.jvm.pools.CMS-Perm-Gen.committed
             value = 63119360
application_1456611008120_0001.driver.jvm.pools.CMS-Perm-Gen.init
             value = 21757952
application_1456611008120_0001.driver.jvm.pools.CMS-Perm-Gen.max
             value = 536870912
application_1456611008120_0001.driver.jvm.pools.CMS-Perm-Gen.usage
             value = 0.11523525416851044
application_1456611008120_0001.driver.jvm.pools.CMS-Perm-Gen.used
             value = 61869968
application_1456611008120_0001.driver.jvm.pools.Code-Cache.committed
             value = 2555904
application_1456611008120_0001.driver.jvm.pools.Code-Cache.init
             value = 2555904
application_1456611008120_0001.driver.jvm.pools.Code-Cache.max
             value = 50331648
application_1456611008120_0001.driver.jvm.pools.Code-Cache.usage
             value = 0.035198211669921875
application_1456611008120_0001.driver.jvm.pools.Code-Cache.used
             value = 1771584
application_1456611008120_0001.driver.jvm.pools.Par-Eden-Space.committed
             value = 286326784
application_1456611008120_0001.driver.jvm.pools.Par-Eden-Space.init
             value = 286326784
application_1456611008120_0001.driver.jvm.pools.Par-Eden-Space.max
             value = 286326784
application_1456611008120_0001.driver.jvm.pools.Par-Eden-Space.usage
             value = 0.23214205486274034
application_1456611008120_0001.driver.jvm.pools.Par-Eden-Space.used
             value = 66468488
application_1456611008120_0001.driver.jvm.pools.Par-Survivor-Space.committed
             value = 35782656
application_1456611008120_0001.driver.jvm.pools.Par-Survivor-Space.init
             value = 35782656
application_1456611008120_0001.driver.jvm.pools.Par-Survivor-Space.max
             value = 35782656
application_1456611008120_0001.driver.jvm.pools.Par-Survivor-Space.usage
             value = 0.3778301979595925
application_1456611008120_0001.driver.jvm.pools.Par-Survivor-Space.used
             value = 13519768
application_1456611008120_0001.driver.jvm.total.committed
             value = 1103634432
application_1456611008120_0001.driver.jvm.total.init
             value = 1098055680
application_1456611008120_0001.driver.jvm.total.max
             value = 1625161728
application_1456611008120_0001.driver.jvm.total.used
             value = 152898864

-- Timers ----------------------------------------------------------------------
DAGScheduler.messageProcessingTime
             count = 4
         mean rate = 0.13 calls/second
     1-minute rate = 0.05 calls/second
     5-minute rate = 0.01 calls/second
    15-minute rate = 0.00 calls/second
               min = 0.06 milliseconds
               max = 6.99 milliseconds
              mean = 1.78 milliseconds
            stddev = 2.98 milliseconds
            median = 0.07 milliseconds
              75% <= 0.08 milliseconds
              95% <= 6.99 milliseconds
              98% <= 6.99 milliseconds
              99% <= 6.99 milliseconds
            99.9% <= 6.99 milliseconds

Load the metrics config:

  1. Create a metrics config file: /tmp/metrics.properties based on the template in bootstrap step.
  2. Configure to load the file when starting Spark:
  • command line: appending –conf “spark.metrics.conf=/tmp/metrics.properties” doesn’t work, didn’t try -Dspark.metrics.conf=/tmp/metrics.properties
  • an alternative is –files=/path/to/metrics.properties –conf spark.metrics.conf=metrics.properties, have’t try that
  • in code: .set(“spark.metrics.conf”, “/tmp/metrics.properties”), this works

Metrics config file example:

代码语言：javascript复制
*.sink.console.class=org.apache.spark.metrics.sink.ConsoleSink
*.sink.console.period=20
*.sink.console.unit=seconds
master.sink.console.period=15
master.sink.console.unit=seconds

*.sink.csv.class=org.apache.spark.metrics.sink.CsvSink
*.sink.csv.period=1
*.sink.csv.unit=minutes
*.sink.csv.directory=/tmp/metrics/csv
worker.sink.csv.period=1
worker.sink.csv.unit=minutes

*.sink.slf4j.class=org.apache.spark.metrics.sink.Slf4jSink
*.sink.slf4j.period=1
*.sink.slf4j.unit=minutes

Coda Hale Metrics Library also supports Graphite Sink (link1, link2), which stores numeric time-series data and render graphs of them on demand.

Graphite consists of 3 software components:

  • carbon – a Twisted daemon that listens for time-series data, which receives the data published by Graphite Sink on EMR
  • whisper – a simple database library for storing time-series data (similar in design to RRD)
  • graphite webapp – A Django webapp that renders graphs on-demand using Cairo
  1. Sync Existing Metrics Data to S3 (not recommended)
  • step 1, write some bootstrap scripts to sync metrics data from EMR cluster to S3 incrementally and timely
  • step 2, after that, a possible way to restore these metrics in the same visualization is having a tool (environment) to download it from S3 and start the Spark History Server on it, or we can make some simple tool to get and analyze the metrics on S3

Other metrics data:

  • OS profiling tools such as dstat, iostat, and iotop can provide fine-grained profiling on individual nodes.
  • JVM utilities such as jstack for providing stack traces, jmap for creating heap-dumps, jstat for reporting time-series statistics and jconsole.

II. EMR cluster/instance Metrics

What metrics we already have even clusters are terminated?

  1. EMR cluster monitor
  • Cluster status – idle/running/failed
  • Map/Reduce – map task running/remaining …
  • Node status – core/task/data nodes …
  • IO – S3/HDFS r & w
  1. CloudWatch

link

Basic monitor and alert system built based on SNS is already itegrated.

文章未经特殊标明皆为本人原创，未经许可不得用于任何商业用途，转载请保持完整性并注明来源链接 《四火的唠叨》

emr spark 面向对象编程 数据结构 linux

0 人点赞