yarn web ui接口数据的获取的探索和实现思路

需求背景

     公司一项目需对spark程序进行性能分析，单纯从时间的角度去看并不能看出实际的执行性能，就想通过yarn对每个application的资源记录来综合评估。

分析yarn的源码发现，yarn的ui在获取application的数据都是通过RMWebServiceProtocol 这个接口来实现的，开始有了一个思路就是：

根据 RMWebServiceProtocol这个协议生成一个客户端去访问RM 的服务端，调里面的的一些方法可以拿到具体的信息。

后来发现yarn的web ui 项目中RMWebServiceProtocol的实现类RMWebServices中已经有一些对外暴露的接口，在配置类RMWSConsts中有详细的信息：

/** Path for {@code RMWebServiceProtocol}. */
  public static final String RM_WEB_SERVICE_PATH = "/ws/v1/cluster";

  /** Path for {@code RMWebServiceProtocol#getClusterInfo}. */
  public static final String INFO = "/info";

  /** Path for {@code RMWebServiceProtocol#getClusterUserInfo}. */
  public static final String CLUSTER_USER_INFO = "/userinfo";

  /** Path for {@code RMWebServiceProtocol#getClusterMetricsInfo}. */
  public static final String METRICS = "/metrics";

  /** Path for {@code RMWebServiceProtocol#getSchedulerInfo}. */
  public static final String SCHEDULER = "/scheduler";

  /** Path for {@code RMWebServices#updateSchedulerConfiguration}. */
  public static final String SCHEDULER_CONF = "/scheduler-conf";

  /** Path for {@code RMWebServices#formatSchedulerConfiguration}. */
  public static final String FORMAT_SCHEDULER_CONF = "/scheduler-conf/format";

  /** Path for {@code RMWebServices#getSchedulerConfigurationVersion}. */
  public static final String SCHEDULER_CONF_VERSION = "/scheduler-conf/version";

  /** Path for {@code RMWebServiceProtocol#dumpSchedulerLogs}. */
  public static final String SCHEDULER_LOGS = "/scheduler/logs";

  /**
   * Path for {@code RMWebServiceProtocol#validateAndGetSchedulerConfiguration}.
   */
  public static final String SCHEDULER_CONF_VALIDATE
          = "/scheduler-conf/validate";

  /** Path for {@code RMWebServiceProtocol#getNodes}. */
  public static final String NODES = "/nodes";

  /** Path for {@code RMWebServiceProtocol#getNode}. */
  public static final String NODES_NODEID = "/nodes/{nodeId}";

  /**
   * Path for {@code RMWebServiceProtocol#getApps} and
   * {@code RMWebServiceProtocol#getApp}.
   */
  public static final String APPS = "/apps";
  /** Path for {@code RMWebServiceProtocol#getApp}. */
  public static final String APPS_APPID = "/apps/{appid}";

地址就是：http://{yarnhostname}:8088/ws/v1/cluster

跟上上面配置类RMWSConsts中 的具体地址，比如：

获取所有apps的信息就用是：http://{yarnhostname}:8088/ws/v1/cluster/apps

获取各个资源池的运行情况：http://{yarnhostname}:8088/ws/v1/cluster/scheduler

访问这些接口地址会发现返回的都是类json格式的数据，可以直接通过HTTPCLIENT或则通过PY的urllib2包都可以去拿到这些数据。这样就有了新的实现方式。

下面是用python简单实现的代码，最终数据写入到了oracle，也可以写入到其他数据库

import urllib2
import json
import time
import sys
import os
import cx_Oracle

os.environ["NLS_LANG"] = ".AL32UTF8" 
db = cx_Oracle.connect('数据库连接信息')
cursor = db.cursor()

reload(sys)
sys.setdefaultencoding( "utf-8" )
url="http://[yarnhostname]:8088/ws/v1/cluster/apps?"
response = urllib2.urlopen(url)
result_dict=json.loads(response. read())
print("applicationId"+"|"+"user"+"|"+"pool"+"|"+"name"+"|"+"state"+"|"+"startTime"+"|"+"launchTime"+"|"+"endTime"+"|"+"allocatedMemorySeconds"+"|"+"allocatedVcoreSeconds"+"|"+"allocatedMB"+"|"+"allocatedVCores"+"|"+"runningContainers")+"|"+"collectTime"+"|"+"elapsedTime"
for app in result_dict["apps"]["app"]: 
    print app["id"]+"|"+app["user"]+"|"+app["queue"]+"|"+app["name"].replace('\n', '').replace('\r', '')+"|"+app["state"]+"|"+time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(float(app["startedTime"]/1000)))+"|"+time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(float(app["launchTime"]/1000)))+"|"+time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(float(app["finishedTime"]/1000)))+"|"+str(app["memorySeconds"])+"|"+str(app["vcoreSeconds"])+"|"+str(app["allocatedMB"])+"|"+str(app["allocatedVCores"])+"|"+str(app["runningContainers"])+"|"+time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())+"|"+str(app["elapsedTime"])
    sql_content = (
"""insert into YARN_META_COL(
APPLICATIONID,
JOB_USER,
RES_POOL,
JOB_NAME,
JOB_STATE,
STARTTIME,
LAUNCHTIME,
ENDTIME,
ALLOCATEDMEMORYSECONDS,
ALLOCATEDVCORESECONDS,
ALLOCATEDMB,
ALLOCATEDVCORES,
RUNNINGCONTAINERS,
COLLECTTIME,
ELAPSEDTIME
)
	values 
(
'%s',
'%s',
'%s',
'%s',
'%s',
to_date('%s','yyyy-mm-dd hh24:mi:ss'),
to_date('%s','yyyy-mm-dd hh24:mi:ss'),
to_date('%s','yyyy-mm-dd hh24:mi:ss'),
'%s',
'%s',
'%s',
'%s',
'%s',
'%s',
to_number('%s')
) """ 
% (app["id"],
app["user"],
app["queue"],
app["name"].replace('\n', '').replace('\r', '').replace("'", '"'),
app["state"],
time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(float(app["startedTime"]/1000))),
time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(float(app["launchTime"]/1000))),
time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(float(app["finishedTime"]/1000))),
str(app["memorySeconds"]),
str(app["vcoreSeconds"]),
str(app["allocatedMB"]),
str(app["allocatedVCores"]),
str(app["runningContainers"]),
time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
str(app["elapsedTime"])
)
)
#    print(sql_content)
    cursor.execute(sql_content)

db.commit()
cursor.close()
db.close()