Hadoop和Spark性能监控工具Dr. Elephant的编

Dr. Elephant是linkedin开源的mapreduce和spark作业的性能监控及调优工具, 它开发了可插拔式的启发式规则来诊断mapreduce和spark作业运行中存在的问题, 并给出调优的解决方案. 它也可以通过与其他调度器 (如Azakaban) 结合来分析作业流的历史性能表现.
用户在上线新的hadoop和spark作业之前, 使用Dr. Elephant进行诊断及调优能够优化作业运行性能,提高集群资源使用效率.
本文主要介绍如何编译、部署Dr. Elephant, 以及使用中的常见问题.

下载并编译代码，生成部署需要的包

1. 从github上下载Dr. Elephant源码

git clone https://github.com/linkedin/dr-elephant.git

2. 本地安装play框架，在环境变量PATH添加play命令路径。play对Java版本要求，版本必须是Java 8以上。(从官网下载traditional play,不是activator)

   # play framework的环境变量配置
   export JAVA_HOME=/usr/jdk64/jdk1.8.0_60/
   export PLAY_HOME=/opt/play-2.2.6
   PATH=$PATH:$PLAY_HOME:$JAVA_HOME/bin
   # 能运行play命令即为配置成功
   

3. 修改源码中编译时用到的配置文件，app-conf/compile.conf，将hadoop_version和spark_version修改成对应的版本。（如果不修改，是否会遇到问题，我没有尝试）。

4. 运行源码中编译命令./compile.sh [./app-conf/compile.conf]，请将中括号中路径修改为相应文件的路径。编译过程会下载很多需要用到的jar包和文件，请耐心等待。我前两次编译都因为部分文件下载失败而导致编译失败。如果有问题，继续尝试编译，已经下载过的文件会自动跳过，就不用再次下载了。

5. 编译完成后，会有SUCCESS的提示。这时可以看到在源码文件夹中，多了一个目录dist，进入这个目录可以看到，里面有一个zip包dr-elephant.zip，解压缩这个zip包，生成一个文件夹dr-elephant。记住，这个文件夹需要被分发到需要部署Dr.Elephant的机器上，下面会用到它。

6. 源码中的app-conf文件夹也会在后面分发到需要部署Dr.Elephant的机器上

编译生成可执行文件，在本地即可完成。是不需要在本地安装Hadoop平台的。

集群环境要求

1. 确保Yarn资源管理器的正常运行以及历史任务服务的正常运行，对于使用Yarn管理Hadoop平台的集群环境来说，一般这个条件都能满足

2. 部署启动一个MySQL服务，MySQL的版本必须是5.5+。对MySQL没有特殊配置，分配好对这个DB有读写权限的账号即可。在MySQL中创建一个db名称drelephant（这个名称可以是任意的，后面可以看到，和elephant.conf配置文件中db名称一致即可）

   # 若操作系统是redhat7,没有mysql源,可以用mariadb代替
   yum install mariadb*  # 安装mariadb的全套
   systemctl start mariadb  # 启动mariadb
   # 安装成功后mariadb其他命令与mysql一样
   # 创建drelephant数据库
   mysql> create database drelephant;
   

3. 将准备工作中提到的需要用到的app-conf文件夹也拷贝到需要部署的机器上某个目录，假设这个目录是/a/app-conf/。修改app-conf文件夹中的文件elephant.conf文件，添加jvm_props,注意与文件里本来有的一行jvm_args不一样.

   jvm_props=" -Devolutionplugin=enabled -DapplyEvolutions.default=true
   

   并修改这个文件中的db_url、db_user、db_name和db_password配置参数，改为我们要使用的MySQL的连接参数。

4. 在这个机器上配置环境变量ELEPHANT_CONF_DIR，让这个环境变量指向从源码中拷贝过来的app-conf文件夹。其实这步不是必须的，后面可以看到，如果不设置这个变量ELEPHANT_CONF_DIR，在启动Dr.Elephant时，在命令行参数中第一个参数添加这个本地目录也是可以的。

部署启动

1. 在Hadoop集群中选择一台机器作为Dr.Elephant的部署机器，在这个机器上选择一个有权限访问HDFS的账号，作为Dr.Elephant的启动账号

2. 将前面步骤中提到的编译后生成的dist/dr-elephant*文件拷贝到这个机器上,如/opt/dr-elephant/

3. 修改app-conf/elephant.conf中第一行,端口号默认是8080,但是会与ambari冲突,所以改成port=8081

4. 进入bin目录，执行./start.sh [/a/app-conf]启动。如果没有执行“集群环境要求”中的第4步，这里需要手动输入/a/app-conf这个目录作为命令行参数

5. 启动后，验证是否执行成功。如果命令行输出提示“Dr.Elephant started”并不能证明它已经成功启动。可以按照以下步骤逐步验证：（1）ps aux | grep elephant，查看是否有这个进程，如果有，那么说明启动成功；（2）查看默认的log，dr.log，在安装包的首层目录下，如果日志中没有明显的异常抛出，说明启动正常；（3）登录用到的MySQL，进入使用的db，查看是否有表play_evolutions、yarn_app_heuristic_result、yarn_app_heuristic_result_details以及yarn_app_result，如果有这些表也说明启动正常；（4）浏览器请求 http://host:8081 ,（这个host是部署Dr.Elephant的机器的host），看看是否能看到Dr.Elephant UI的首页，如果可以，说明启动成功

6. Dr elephant运行的详细日志在安装包的上层目录,如/opt/logs/elephant/dr_elephant.log,比较难找.

使用中常见问题

作业信息获取

1. mapreduce作业有两种可选的fetcher,在FetcherConf.xml中配置.
   MapReduceFetcherHadoop2是从yarn history server获取作业信息,如
   http://localhost:19888/ws/v1/history/mapreduce/jobs/job_1477464172237_0069

   MapReduceFSFetcherHadoop2是通过读日志文件来获取作业信息,每个作业对应.jhist和.xml两个文件

   # *.xml文件里面记录的是相应作业运行时候的完整参数配置
   hdfs dfs -cat /mr-history/done/2016/11/01/000000/job_1477464172237_0052_conf.xml
   # *.jhist文件里存放的是具体Hadoop作业运行的详细信息
   hdfs dfs -cat /mr-history/done/2016/11/01/000000/job_1477464172237_0052-1477984365827-ocdp-QuasiMonteCarlo-1477984395977-4-1-SUCCEEDED-default-1477984370174.jhist
   

2. spark作业只有一种fetcher,即SparkFSFetcher,通过读日志文件来获取信息

监控和优化spark作业

1. 首先要保证spark history server在正常运行,如果它正常运行的话,每运行一个spark作业,可以在hdfs:///spark-history/下看到对应的日志文件,如application_1478153889331_0019,dr-elephant就是从这个日志获取数据的.

   cd /usr/hdp/2.4.0.0-169/spark/sbin/
   ./start-history-server.sh 
   

2. 如果还有问题,先把spark history server的进程kill掉,再重新起.

3. 要注意dr-elephant是否有/spark-history/目录的读写权限

4. 提交spark作业时,要使用yarn-client模式(--master yarn-client),产生的日志文件名为application_1478153889331_0021,如果用yarn-cluster模式,产生的日志文件名为application_1478153889331_0012_appattempt_1478153889331_0012_000001,而dr-elephant能识别的文件名为前者.

   ./bin/spark-submit \
     --class org.apache.spark.examples.SparkPi \
     --master yarn-client \
     --num-executors 3 \
     --driver-memory 512m \
     --executor-memory 2g \
     --executor-cores 1 \
     /usr/hdp/2.4.0.0-169/spark/lib/spark-examples-1.6.0.2.4.0.0-169-hadoop2.7.1.2.4.0.0-169.jar \
     10
   

5. FetcherConf.xml里spark的params默认是注释掉的,注意需要去掉注释. event_log_dir定义了spark作业历史日志的路径,要与spark history server的配置一致,我的是/spark-history,注意这里指是hdfs上的文件.spark_log_ext定义了spark作业日志文件的后缀,如果把spark.eventLog.compress设为true,则后缀为.snappy,表示压缩文件.相反如果把spark.eventLog.compress设为false,则没有后缀,spark_log_ext应为空.

   <fetcher>
       <applicationtype>spark</applicationtype>
       <classname>org.apache.spark.deploy.history.SparkFSFetcher</classname>
       <params>
         <event_log_size_limit_in_mb>100</event_log_size_limit_in_mb>
         <event_log_dir>/spark-history</event_log_dir>
         <spark_log_ext>.snappy</spark_log_ext>
         #the values specified in namenode_addresses will be used for obtaining spark logs. The cluster configuration will be ignored.
   <!--      <namenode_addresses>address1,address2</namenode_addresses>
             -->
       </params>
     </fetcher>
   

   spark配置

   spark.eventLog.compress
   默认值：false 
   是否压缩记录Spark事件，前提spark.eventLog.enabled为true，默认使用的是snappy
   

   以spark.history开头的需要配置在spark-env.sh中的SPARK_HISTORY_OPTS，以spark.eventLog开头的配置在spark-defaults.conf.

   如果ambari中搜不到这项配置,可以在自定义spark-defaults中手工增加属性.

Job history和 Flow history功能

搜索框要输入的job id/url可以是azkaban里定义的job name(如pi),dr.elephant会从mysql的drelephant库里的yarn_app_result表中搜索字段job_def_id='pi'对应的作业application id,然后根据数据库里的具体信息画出图表.

    MariaDB [drelephant]> show tables;
    +-----------------------------------+
    | Tables_in_drelephant              |
    +-----------------------------------+
    | play_evolutions                   |
    | yarn_app_heuristic_result         |
    | yarn_app_heuristic_result_details |
    | yarn_app_result                   |
    +-----------------------------------+
    4 rows in set (0.00 sec)
    MariaDB [drelephant]> select id,name,username,job_type,scheduler,job_name,job_exec_id,job_def_id,flow_def_id,flow_exec_id  from yarn_app_result where job_def_id !="";
    +--------------------------------+-----------------+----------+------------+-----------+----------+-------------+------------+-------------+--------------+
    | id                             | name            | username | job_type   | scheduler | job_name | job_exec_id | job_def_id | flow_def_id | flow_exec_id |
    +--------------------------------+-----------------+----------+------------+-----------+----------+-------------+------------+-------------+--------------+
    | application_1477464172237_0067 | QuasiMonteCarlo | test     | HadoopJava | azkaban   | pi       | 4           | pi         | example     | 8            |
    | application_1477464172237_0071 | QuasiMonteCarlo | test     | HadoopJava | azkaban   | pi       | 3           | pi         | example     | 9            |
    | application_1478153889331_0007 | QuasiMonteCarlo | test     | HadoopJava | azkaban   | pi       | 1           | pi         | example     | 10           |
    | application_1478153889331_0008 | QuasiMonteCarlo | test     | HadoopJava | azkaban   | pi       | 2           | pi         | example     | 11           |
    +--------------------------------+-----------------+----------+------------+-----------+----------+-------------+------------+-------------+--------------+
    4 rows in set (0.00 sec)

参考资料

1. dr.elephant github
2. 知乎dr. elephant专栏