Spark 应用监控告警-Graphite_exporter
阅读原文时间:2023年07月09日阅读:3

Spark 应用监控告警-Graphite_exporter


相关内容原文地址:

CSDN:瓜牛呱呱:让 Spark Streaming 程序在 YARN 集群上长时间运行(一)—— 故障容错、性能、安全

CSDN:瓜牛呱呱:让 Spark Streaming 程序在 YARN 集群上长时间运行(二)—— 日志、监控、Metrics

CSDN:瓜牛呱呱:让 Spark Streaming 程序在 YARN 集群上长时间运行(三)—— Grafana、优雅地停止(Graceful stop)、总结

CSDN:weixin_44772715:graphite_exporter收集spark on yarn metrics

简书:0o青团o0:Spark 应用监控告警和自动重启

灰信网(软件开发博客聚合):使用 PROMETHEUS 和 GRAFANA 监控 SPARK 应用

并发编程网:邓林:《Spark 官方文档》监控和工具


Spark监控和工具

监控Spark应用有很多种方式:web UI,metrics 以及外部工具。

每个SparkContext都会启动一个web UI,其默认端口为4040,并且这个web UI能展示很多有用的Spark应用相关信息。包括:

  • 一个stage和task的调度列表
  • 一个关于RDD大小以及内存占用的概览
  • 运行环境相关信息
  • 运行中的执行器相关信息

打开浏览器,输入 http://:4040 即可访问该web界面。如果有多个SparkContext在同时运行中,那么它们会从4040开始,按顺序依次绑定端口(4041,4042,等)。

注意,默认情况下,这些信息只有在Spark应用运行期内才可用。如果需要在Spark应用退出后仍然能在web UI上查看这些信息,则需要在应用启动前,将 spark.eventLog.enabled 设为 true。这项配置将会把Spark事件日志都记录到持久化存储中。

Spark独立部署时,其对应的集群管理器也有其对应的 web UI。如果Spark应用将其运行期事件日志保留下来了,那么独立部署集群管理器对应的web UI将会根据这些日志自动展示已经结束的Spark应用。

如果Spark是运行于Mesos或者YARN上的话,那么你需要开启Spark的history server,开启event log。开启history server需要如下指令:

./sbin/start-history-server.sh

如果使用file-system provider class(参考下面的 spark.history.provider),那么日志目录将会基于 spark.history.fs.logDirectory 配置项,并且在表达Spark应用的事件日志路径时,应该带上子目录。history server对应的web界面默认在这里 http://:18080。

环境变量

含义

SPARK_DAEMON_MEMORY

history server分配多少内存(默认: 1g)

SPARK_DAEMON_JAVA_OPTS

history server的 JVM参数(默认:none)

SPARK_PUBLIC_DNS

history server的外部访问地址,如果不配置,那么history server有可能会绑定server的内部地址,这可能会导致外部不能访问(默认:none)

SPARK_HISTORY_OPTS

history server配置项(默认:none):spark.history.*

度量信息除了可以在UI上查看之外,还可以以JSON格式访问。这能使开发人员很容易构建新的Spark可视化和监控工具。JSON格式的度量信息对运行中的Spark应用和history server中的历史作业均有效。其访问端点挂载在 /api/v1 路径下。例如,对于history server,一般你可以通过 http://:18080/api/v1 来访问,而对于运行中的应用,可以通过 http://localhost:4040/api/v1 来访问。

如果在YARN上运行,每个应用都由多个attempts,所以 [app-id] 实际上是 [app-id]/[attempt-id]。

这些API端点都有版本号,所以基于这些API开发程序就比较容易。Spark将保证:

  • 端点一旦添加进来,就不会删除
  • 某个端点支持的字段永不删除
  • 未来可能会增加新的端点
  • 已有端点可能会增加新的字段
  • 未来可能会增加新的API版本,但会使用不同的端点(如:api/v2 )。但新版本不保证向后兼容。
  • API版本可能会整个丢弃掉,但在丢弃前,一定会和新版本API共存至少一个小版本。

注意,在UI上检查运行中的应用时,虽然每次只能查看一个应用, 但applicatoins/[app-id] 这部分路径仍然是必须的。例如,你需要查看运行中应用的作业列表时,你需要输入 http://localhost:4040/api/v1/applications/[app-id]/jobs。虽然麻烦点,但这能保证两种模式下访问路径的一致性。

Spark的度量子系统是可配置的,其功能是基于Coda Hale Metrics Library开发的。这套度量子系统允许用户以多种形式的汇报槽(sink)汇报Spark度量信息,包括:HTTP、JMX和CSV文件等。其对应的配置文件路径为:${SPARK_HOME}/conf/metrics.properties。当然,你可以通过spark.metrics.conf 这个Spark属性来自定义配置文件路径(详见configuration property)。Spark的各个组件都有其对应的度量实例,且这些度量实例之间是解耦的。这些度量实例中,你都可以配置一系列不同的汇报槽来汇报度量信息。以下是目前支持的度量实例:

  • master: 对应Spark独立部署时的master进程。
  • applications: master进程中的一个组件,专门汇报各个Spark应用的度量信息。
  • worker: 对应Spark独立部署时的worker进程。
  • executor: 对应Spark执行器。
  • driver: 对应Spark驱动器进程(即创建SparkContext对象的进程)。

每个度量实例可以汇报给0~n个槽。以下是目前 org.apache.spark.metrics.sink 包中包含的几种汇报槽(sink):

  • ConsoleSink:将度量信息打印到控制台。
  • CSVSink: 以特定的间隔,将度量信息输出到CSV文件。
  • JmxSink: 将度量信息注册到JMX控制台。
  • MetricsServlet: 在已有的Spark UI中增加一个servlet,对外提供JSON格式的度量数据。
  • GraphiteSink: 将度量数据发到Graphite 节点。
  • Slf4jSink: 将度量数据发送给slf4j 打成日志。

Spark同样也支持Ganglia,但因为license限制的原因没有包含在默认的发布包中:

  • GangliaSink: 将度量信息发送给一个Ganglia节点或者多播组。

度量系统配置文件语法可以参考这个配置文件示例:${SPARK_HOME}/conf/metrics.properties.template

以下是几个可以用以分析Spark性能的外部工具:

  • 集群整体监控工具,如:Ganglia,可以提供集群整体的使用率和资源瓶颈视图。比如,Ganglia的仪表盘可以迅速揭示出整个集群的工作负载是否达到磁盘、网络或CPU限制。
  • 操作系统分析工具,如:dstat, iostat, 以及 iotop ,可以提供单个节点上细粒度的分析剖面。
  • JVM工具可以帮助你分析JVM虚拟机,如:jstack可以提供调用栈信息,jmap可以转储堆内存数据,jstat可以汇报时序统计信息,jconsole可以直观的探索各种JVM属性,这对于熟悉JVM内部机制非常有用。

一、下载graphite_exporter

Graphite 来收集度量标准,Grafana 则用于构建仪表板,首先,需要配置 Spark 以将 metrics 报告到 Graphite。

prometheus 提供了一个插件(graphite_exporter),可以将 Graphite metrics 进行转化并写入 Prometheus (本文的方式)。

先去https://prometheus.io/download/下载graphite_exporter。

将下载下来的graphite_exporter上传至服务器:

rz + graphite_exporter-.....tar.gz

解压:

tar -zxvf graphite_exporter....tar.gz

重命名:

mv graphite_exporter....  graphite_exporter

进入到graphite_exporter,在该路面下创建graphite_exporter_mapping文件:

vi graphite_exporter_mapping

添加如下内容

mappings:
- match: '*.*.executor.filesystem.*.*'
  name: spark_app_filesystem_usage
  labels:
    application: $1
    executor_id: $2
    fs_type: $3
    qty: $4
- match: '*.*.jvm.*.*'
  name: spark_app_jvm_memory_usage
  labels:
    application: $1
    executor_id: $2
    mem_type: $3
    qty: $4
- match: '*.*.executor.jvmGCTime.count'
  name: spark_app_jvm_gcTime_count
  labels:
    application: $1
    executor_id: $2
- match: '*.*.jvm.pools.*.*'
  name: spark_app_jvm_memory_pools
  labels:
    application: $1
    executor_id: $2
    mem_type: $3
    qty: $4
- match: '*.*.executor.threadpool.*'
  name: spark_app_executor_tasks
  labels:
    application: $1
    executor_id: $2
    qty: $3
- match: '*.*.BlockManager.*.*'
  name: spark_app_block_manager
  labels:
    application: $1
    executor_id: $2
    type: $3
    qty: $4
- match: '*.*.DAGScheduler.*.*'
  name: spark_app_dag_scheduler
  labels:
    application: $1
    executor_id: $2
    type: $3
    qty: $4
- match: '*.*.CodeGenerator.*.*'
  name: spark_app_code_generator
  labels:
    application: $1
    executor_id: $2
    type: $3
    qty: $4
- match: '*.*.HiveExternalCatalog.*.*'
  name: spark_app_hive_external_catalog
  labels:
    application: $1
    executor_id: $2
    type: $3
    qty: $4
- match: '*.*.*.StreamingMetrics.*.*'
  name: spark_app_streaming_metrics
  labels:
    application: $1
    executor_id: $2
    app_name: $3
    type: $4
    qty: $5


./graphite_exporter --graphite.mapping-config=graphite_exporter_mapping

Spark应用监控解决方案–使用Prometheus和Grafana监控Spark应用——

graphite_exporter_mapping格式:

mappings:
- match: '*.*.executor.filesystem.*.*'
  name: filesystem_usage
  labels:
    application: $1
    executor_id: $2
    fs_type: $3
    qty: $4

- match: '*.*.executor.threadpool.*'
  name: executor_tasks
  labels:
    application: $1
    executor_id: $2
    qty: $3

- match: '*.*.executor.jvmGCTime.count'
  name: jvm_gcTime_count
  labels:
    application: $1
    executor_id: $2

- match: '*.*.executor.*.*'
  name: executor_info
  labels:
    application: $1
    executor_id: $2
    type: $3
    qty: $4

- match: '*.*.jvm.*.*'
  name: jvm_memory_usage
  labels:
    application: $1
    executor_id: $2
    mem_type: $3
    qty: $4

- match: '*.*.jvm.pools.*.*'
  name: jvm_memory_pools
  labels:
    application: $1
    executor_id: $2
    mem_type: $3
    qty: $4

- match: '*.*.BlockManager.*.*'
  name: block_manager
  labels:
    application: $1
    executor_id: $2
    type: $3
    qty: $4

- match: '*.driver.DAGScheduler.*.*'
  name: DAG_scheduler
  labels:
    application: $1
    type: $2
    qty: $3

- match: '*.driver.*.*.*.*'
  name: task_info
  labels:
    application: $1
    task: $2
    type1: $3
    type2: $4
    qty: $5

二、Spark配置Graphite metrics

Spark 是自带 Graphite Sink 的,只需要配置一下metrics.properties;

进入到spark安装目录下,进入到conf目录下,找到metrics.properties

cd /spark/conf/
vi metrics.properties

在最下面更新如下内容:

*.source.jvm.class=org.apache.spark.metrics.source.JvmSource
master.source.jvm.class=org.apache.spark.metrics.source.JvmSource
worker.source.jvm.class=org.apache.spark.metrics.source.JvmSource
driver.source.jvm.class=org.apache.spark.metrics.source.JvmSource
executor.source.jvm.class=org.apache.spark.metrics.source.JvmSource

*.sink.graphite.class=org.apache.spark.metrics.sink.GraphiteSink
*.sink.graphite.protocol=tcp
*.sink.graphite.host=spark-ip
*.sink.graphite.port=9109
*.sink.graphite.period=1
*.sink.graphite.unit=seconds

graphite_exporter 接收数据端口为9109

Prometheus接收数据端口开发说明

三、启动Spark程序

启动spark程序时,需要加上–files /usr/etc/spark/conf/metrics.properties参数。

例如:

nohup ./spark-submit --class StreamingInput --master spark://master:7077 --supervise --num-executors 3 --total-executor-cores 3 --executor-memory 2g --files /usr/etc/spark/conf/metrics.properties /root/StreamingInput.jar > /root/logs/StreamingInput.log.txt 2>&1 &

将应用程序的 metric 固定为静态的应用程序名称:

--conf spark.metrics.namespace=my_application_name

四、配置Prometheus

scrape_configs:
  - job_name: 'spark'
    static_configs:
    - targets: ['spark-ip:9108']

重启Prometheus。

curl spark-ip:9108/metrics

若成功,则会收到如下数据:

# HELP go_gc_duration_seconds A summary of the GC invocation durations.
# TYPE go_gc_duration_seconds summary
go_gc_duration_seconds{quantile="0"} 3.5844e-05
go_gc_duration_seconds{quantile="0.25"} 7.1282e-05
go_gc_duration_seconds{quantile="0.5"} 9.674e-05
go_gc_duration_seconds{quantile="0.75"} 0.0001363

.................................此处略过很多行

spark_app_jvm_memory_usage{application="application_1556625797222_0045",executor_id="driver",mem_type="total",qty="max"} 1.050673151e+09
spark_app_jvm_memory_usage{application="application_1556625797222_0045",executor_id="driver",mem_type="total",qty="used"} 1.114036792e+09

五、Metrics分析

Spark 会从 Driver 和 executors 生成大量指标。 如果我选择最重要的一个,那将是最后收到的 batch 的信息。 当StreamingMetrics.streaming.lastReceivedBatch_records == 0时,这可能意味着 Spark Streaming 作业已停止或失败。

下面列出了其他重要的 metrics:

  • 当总延迟大于批处理间隔时,处理延迟将会增加。

    driver.StreamingMetrics.streaming.lastCompletedBatch_totalDelay

  • 当运行的 task 数低于 number of executors * number of cores 时,YARN 分配的资源未充分利用。

    executor.threadpool.activeTasks

  • RDD的缓存使用了多少内存。

    driver.BlockManager.memory.memUsed_MB

  • 当 RDD 缓存没有足够的内存时,有多少数据 spilled 到磁盘。 你应该增加executor 内存或更改 spark.memory.fraction 的Spark参数以避免性能下降。

    driver.BlockManager.disk.diskSpaceUsed_MB

  • Spark Driver 上的 JVM 内存利用率是多少

    driver.jvm.heap.used
    driver.jvm.non-heap.used
    driver.jvm.pools.G1-Old-Gen.used
    driver.jvm.pools.G1-Eden-Space.used
    driver.jvm.pools.G1-Survivor-Space.used

  • 在Spark Driver上的 GC 花费了多少时间

    driver.jvm.G1-Old-Generation.time
    driver.jvm.G1-Young-Generation.time

  • Spark executors 上的 JVM 内存利用率是多少

    [0-9].jvm.heap.used [0-9].jvm.non-heap.used
    [0-9].jvm.pools.G1-Old-Gen.used [0-9].jvm.pools.G1-Survivor-Space.used
    [0-9]*.jvm.pools.G1-Eden-Space.used

  • 在Spark executors上的 GC 花费了多少时间

    [0-9].jvm.G1-Old-Generation.time [0-9].jvm.G1-Young-Generation.time

六、Grafana模板

在Grafana官网,没有找到合适的dashboard,貌似得自己手写了。需要自己研究手写搞一套了。。

DASHBOARD 配置:

将 application label 加入 Variables 用于筛选不同的应用:

配置相应的图表:

七、进程监控失败重启和告警模块

监控yarn上指定的Spark应用是否存在,不存在则发出告警。

使用Python脚本查看yarn状态,指定监控应用,应用中断则通过webhook发送报警信息到钉钉群,并且自动重启。

#!/usr/bin/python3.5
# -*- coding: utf-8 -*-
import os
import json
import requests

'''
Yarn应用监控:当配置的应用名不在yarn applicaition -list时,钉钉告警
'''

def yarn_list(applicatin_list):
    yarn_application_list = os.popen('yarn application -list').read()
    result = ""
    for appName in applicatin_list:
        if appName in yarn_application_list:
            print("应用:%s 正常!" % appName)
        else:
            result += ("告警--应用:%s 中断!" % appName)
            if "应用名1" == appName:
                os.system('重启命令')

    return result

def dingding_robot(data):
    # 机器人的webhooK 获取地址参考:https://open-doc.dingtalk.com/microapp/serverapi2/qf2nxq
    webhook = "https://oapi.dingtalk.com/robot/send?access_token" \
              "=你的token "
    headers = {'content-type': 'application/json'}  # 请求头
    r = requests.post(webhook, headers=headers, data=json.dumps(data))
    r.encoding = 'utf-8'
    return r.text

if __name__ == '__main__':
    applicatin_list = ["应用名1", "应用名2", "应用名3"]
    output = yarn_list(applicatin_list)
    print(output)

    if len(output) > 0:
        # 请求参数 可以写入配置文件中
        data = {
            "msgtype": "text",
            "text": {
                "content": output
            },
            "at": {
                "atMobiles": [
                    "xxxxxxx"
                ],
                "isAtAll": False
            }
        }
        res = dingding_robot(data)
        print(res)  # 打印请求结果
    else:
        print("一切正常!")