models.py 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570
  1. #!/usr/bin/env python
  2. # Licensed to Cloudera, Inc. under one
  3. # or more contributor license agreements. See the NOTICE file
  4. # distributed with this work for additional information
  5. # regarding copyright ownership. Cloudera, Inc. licenses this file
  6. # to you under the Apache License, Version 2.0 (the
  7. # "License"); you may not use this file except in compliance
  8. # with the License. You may obtain a copy of the License at
  9. #
  10. # http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. # Unless required by applicable law or agreed to in writing, software
  13. # distributed under the License is distributed on an "AS IS" BASIS,
  14. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. # See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. import datetime
  18. import logging
  19. import lxml.html
  20. import re
  21. import urllib2
  22. from urlparse import urlparse, urlunparse
  23. from desktop.lib.view_util import format_duration_in_millis
  24. from desktop.lib import i18n
  25. from hadoop import job_tracker
  26. from hadoop import confparse
  27. from hadoop.api.jobtracker.ttypes import JobNotFoundException
  28. import hadoop.api.jobtracker.ttypes as ttypes
  29. from desktop.lib.exceptions_renderable import PopupException
  30. from django.utils.translation import ugettext as _
  31. LOGGER = logging.getLogger(__name__)
  32. class JobLinkage(object):
  33. """
  34. A thin representation of a job, without much of the details.
  35. Its purpose is to wrap a JobID to allow us to get further
  36. information from Hadoop, without instantiating a full Job object
  37. (which requires talking to Hadoop).
  38. """
  39. def __init__(self, jobtracker, jobid):
  40. """
  41. JobLinkage(jobtracker, jobid) -> JobLinkage
  42. The jobid is the jobid string (not the thrift jobid)
  43. """
  44. self._jobtracker = jobtracker
  45. self.jobId = jobid
  46. self.jobId_short = "_".join(jobid.split("_")[-2:])
  47. self.is_mr2 = False
  48. def get_task(self, task_id):
  49. """Retrieve a TaskInProgress from hadoop."""
  50. ttask = self._jobtracker.get_task(
  51. self._jobtracker.thriftjobid_from_string(self.jobId),
  52. self._jobtracker.thrifttaskid_from_string(task_id))
  53. return Task(ttask, self._jobtracker)
  54. class Job(JobLinkage):
  55. """
  56. Creates a Job instance pulled from the job tracker Thrift interface.
  57. """
  58. def __getitem__(self, item):
  59. """
  60. For backwards-compatibility, resolve job["foo"] as job.foo
  61. """
  62. return getattr(self, item)
  63. @staticmethod
  64. def from_id(jt, jobid, is_finished=False):
  65. """
  66. Returns a Job instance given a job tracker interface and an id. The job tracker interface is typically
  67. located in request.jt.
  68. """
  69. try:
  70. thriftjob = jt.get_job(jt.thriftjobid_from_string(jobid))
  71. except JobNotFoundException:
  72. try:
  73. thriftjob = jt.get_retired_job(jt.thriftjobid_from_string(jobid))
  74. except JobNotFoundException, e:
  75. raise PopupException(_("Could not find job with id %(jobid)s.") % {'jobid': jobid}, detail=e)
  76. return Job(jt, thriftjob)
  77. @staticmethod
  78. def from_thriftjob(jt, thriftjob):
  79. """
  80. Returns a Job instance given a job tracker interface and a thriftjob object returned from that job tracker interface.
  81. The job tracker interface is typically located in request.jt
  82. """
  83. return Job(jt, thriftjob)
  84. def __init__(self, jt, thriftJob):
  85. """
  86. Returns a Job instance given a job tracker interface and a thriftjob object returned from that
  87. job tracker interface. The job tracker interface is typically located in request.jt
  88. """
  89. JobLinkage.__init__(self, jt, thriftJob.jobID.asString)
  90. self.jt = jt
  91. self.job = thriftJob
  92. self.tasks = []
  93. if self.job.tasks is not None:
  94. self.tasks = TaskList.from_thriftTaskList(self.job.tasks, jt)
  95. self.task_map = dict( (task.taskId, task) for task in self.tasks )
  96. self._counters = None
  97. self._conf_keys = None
  98. self._full_job_conf = None
  99. self._init_attributes()
  100. self.is_retired = hasattr(thriftJob, 'is_retired')
  101. self.is_mr2 = False
  102. @property
  103. def counters(self):
  104. if self.is_retired:
  105. self._counters = {}
  106. elif self._counters is None:
  107. rollups = self.jt.get_job_counter_rollups(self.job.jobID)
  108. # We get back a structure with counter lists for maps, reduces, and total
  109. # and we need to invert this
  110. def aggregate_counters(ctrs_from_jt, key, target):
  111. for group in ctrs_from_jt.groups:
  112. if group.name not in target:
  113. target[group.name] = {
  114. 'name': group.name,
  115. 'displayName': group.displayName,
  116. 'counters': {}
  117. }
  118. agg_counters = target[group.name]['counters']
  119. for counter in group.counters.itervalues():
  120. if counter.name not in agg_counters:
  121. agg_counters[counter.name] = {
  122. 'name': counter.name,
  123. 'displayName': counter.displayName,
  124. }
  125. agg_counters[counter.name][key] = counter.value
  126. self._counters = {}
  127. aggregate_counters(rollups.mapCounters, "map", self._counters)
  128. aggregate_counters(rollups.reduceCounters, "reduce", self._counters)
  129. aggregate_counters(rollups.jobCounters, "total", self._counters)
  130. return self._counters
  131. @property
  132. def conf_keys(self):
  133. if self._conf_keys is None:
  134. self._initialize_conf_keys()
  135. return self._conf_keys
  136. @property
  137. def full_job_conf(self):
  138. if self._full_job_conf is None:
  139. self._initialize_conf_keys()
  140. return self._full_job_conf
  141. def _init_attributes(self):
  142. self.queueName = i18n.smart_unicode(self.job.profile.queueName)
  143. self.jobName = i18n.smart_unicode(self.job.profile.name)
  144. self.user = i18n.smart_unicode(self.job.profile.user)
  145. self.mapProgress = self.job.status.mapProgress
  146. self.reduceProgress = self.job.status.reduceProgress
  147. self.setupProgress = self.job.status.setupProgress
  148. self.cleanupProgress = self.job.status.cleanupProgress
  149. if self.job.desiredMaps == 0:
  150. maps_percent_complete = 0
  151. else:
  152. maps_percent_complete = int(round(float(self.job.finishedMaps)/self.job.desiredMaps*100))
  153. self.desiredMaps = self.job.desiredMaps
  154. if self.job.desiredReduces == 0:
  155. reduces_percent_complete = 0
  156. else:
  157. reduces_percent_complete = int(round(float(self.job.finishedReduces)/self.job.desiredReduces*100))
  158. self.desiredReduces = self.job.desiredReduces
  159. self.maps_percent_complete = maps_percent_complete
  160. self.finishedMaps = self.job.finishedMaps
  161. self.finishedReduces = self.job.finishedReduces
  162. self.reduces_percent_complete = reduces_percent_complete
  163. self.startTimeMs = self.job.startTime
  164. self.startTimeFormatted = format_unixtime_ms(self.job.startTime)
  165. self.launchTimeMs = self.job.launchTime
  166. self.launchTimeFormatted = format_unixtime_ms(self.job.launchTime)
  167. self.finishTimeMs = self.job.finishTime
  168. self.finishTimeFormatted = format_unixtime_ms(self.job.finishTime)
  169. self.status = self.job.status.runStateAsString
  170. self.priority = self.job.priorityAsString
  171. self.jobFile = self.job.profile.jobFile
  172. finishTime = self.job.finishTime
  173. if finishTime == 0:
  174. finishTime = datetime.datetime.now()
  175. else:
  176. finishTime = datetime.datetime.fromtimestamp(finishTime/1000)
  177. self.duration = finishTime - datetime.datetime.fromtimestamp(self.job.startTime/1000)
  178. diff = int(finishTime.strftime("%s"))*1000 - self.startTimeMs
  179. self.durationFormatted = format_duration_in_millis(diff)
  180. self.durationInMillis = diff
  181. def kill(self):
  182. self.jt.kill_job(self.job.jobID)
  183. def get_task(self, id):
  184. try:
  185. return self.task_map[id]
  186. except:
  187. return JobLinkage.get_task(self, id)
  188. def filter_tasks(self, task_types=None, task_states=None, task_text=None):
  189. """
  190. Filters the tasks of the job.
  191. Pass in task_type and task_state as sets; None for "all".
  192. task_text is used to search in the state, mostRecentState, and the ID.
  193. """
  194. assert task_types is None or job_tracker.VALID_TASK_TYPES.issuperset(task_types)
  195. assert task_states is None or job_tracker.VALID_TASK_STATES.issuperset(task_states)
  196. def is_good_match(t):
  197. if task_types is not None:
  198. if t.task.taskID.taskTypeAsString.lower() not in task_types:
  199. return False
  200. if task_states is not None:
  201. if t.state.lower() not in task_states:
  202. return False
  203. if task_text is not None:
  204. tt_lower = task_text.lower()
  205. if tt_lower not in t.state.lower() and tt_lower not in t.mostRecentState.lower() and tt_lower not in t.task.taskID.asString.lower():
  206. return False
  207. return True
  208. return [ t for t in self.tasks if is_good_match(t) ]
  209. def _initialize_conf_keys(self):
  210. if self.is_retired:
  211. self._conf_keys = {}
  212. self._full_job_conf = {}
  213. else:
  214. conf_keys = [
  215. 'mapred.mapper.class',
  216. 'mapred.reducer.class',
  217. 'mapred.input.format.class',
  218. 'mapred.output.format.class',
  219. 'mapred.input.dir',
  220. 'mapred.output.dir',
  221. ]
  222. jobconf = get_jobconf(self.jt, self.jobId)
  223. self._full_job_conf = jobconf
  224. self._conf_keys = {}
  225. for k, v in jobconf.iteritems():
  226. if k in conf_keys:
  227. self._conf_keys[dots_to_camel_case(k)] = v
  228. class TaskList(object):
  229. @staticmethod
  230. def select(jt, jobid, task_types, task_states, text, count, offset):
  231. """
  232. select(jt, jobid, task_types, task_states, text, count, offset) -> TaskList
  233. Retrieve a TaskList from Hadoop according to the given criteria.
  234. task_types is a set of job_tracker.VALID_TASK_TYPES. A value to None means everything.
  235. task_states is a set of job_tracker.VALID_TASK_STATES. A value to None means everything.
  236. """
  237. assert task_types is None or job_tracker.VALID_TASK_TYPES.issuperset(task_types)
  238. assert task_states is None or job_tracker.VALID_TASK_STATES.issuperset(task_states)
  239. if task_types is None:
  240. task_types = job_tracker.VALID_TASK_TYPES
  241. if task_states is None:
  242. task_states = job_tracker.VALID_TASK_STATES
  243. tjobid = jt.thriftjobid_from_string(jobid)
  244. thrift_list = jt.get_task_list(tjobid, task_types, task_states, text, count, offset)
  245. return TaskList.from_thriftTaskList(thrift_list, jt)
  246. @staticmethod
  247. def from_thriftTaskList(thrift_task_list, jobtracker):
  248. """TaskList.from_thriftTaskList(thrift_task_list, jobtracker) -> TaskList
  249. """
  250. if thrift_task_list is None:
  251. return None
  252. return TaskList(thrift_task_list, jobtracker)
  253. def __init__(self, tasklist, jobtracker):
  254. self.__tasklist = tasklist # The thrift task list
  255. self.__jt = jobtracker
  256. self.__init_attributes()
  257. def __init_attributes(self):
  258. self.__tasksSoFar = [ Task(t, self.__jt) for t in self.__tasklist.tasks ]
  259. self.__nTotalTasks = self.__tasklist.numTotalTasks
  260. def __iter__(self):
  261. return self.__tasksSoFar.__iter__()
  262. def __len__(self):
  263. return len(self.__tasksSoFar)
  264. def __getitem__(self, key):
  265. return self.__tasksSoFar[key]
  266. @property
  267. def tasks(self):
  268. return self.__tasksSoFar
  269. @property
  270. def numTotalTasks(self):
  271. return self.__nTotalTasks
  272. class Task(object):
  273. def __getitem__(self, item):
  274. """
  275. For backwards-compatibility, resolve job["foo"] as job.foo
  276. """
  277. return getattr(self, item)
  278. def __init__(self, task, jt):
  279. self.task = task
  280. self.jt = jt
  281. self._init_attributes()
  282. self.attempt_map = {}
  283. for id, attempt in self.task.taskStatuses.iteritems():
  284. ta = TaskAttempt(attempt, task=self)
  285. self.attempt_map[id] = ta
  286. @property
  287. def attempts(self):
  288. return self.attempt_map.values()
  289. def _init_attributes(self):
  290. self.taskType = self.task.taskID.taskTypeAsString
  291. self.taskId = self.task.taskID.asString
  292. self.taskId_short = "_".join(self.taskId.split("_")[-2:])
  293. self.startTimeMs = self.task.startTime
  294. self.startTimeFormatted = format_unixtime_ms(self.task.startTime)
  295. self.execStartTimeMs = self.task.execStartTime
  296. self.execStartTimeFormatted = format_unixtime_ms(self.task.execStartTime)
  297. self.execFinishTimeMs = self.task.execFinishTime
  298. self.execFinishTimeFormatted = format_unixtime_ms(self.task.execFinishTime)
  299. self.state = self.task.state
  300. assert self.state in job_tracker.VALID_TASK_STATES
  301. self.progress = self.task.progress
  302. self.taskId = self.task.taskID.asString
  303. self.jobId = self.task.taskID.jobID.asString
  304. self.taskAttemptIds = self.task.taskStatuses.keys()
  305. self.mostRecentState = self.task.mostRecentState
  306. self.diagnosticMap = self.task.taskDiagnosticData
  307. self.counters = self.task.counters
  308. self.failed = self.task.failed
  309. self.complete = self.task.complete
  310. self.is_mr2 = False
  311. def get_attempt(self, id):
  312. """
  313. Returns a TaskAttempt for a given id.
  314. """
  315. return self.attempt_map[id]
  316. class TaskAttempt(object):
  317. def __getitem__(self, item):
  318. """
  319. For backwards-compatibility, resolve task["foo"] as task.foo.
  320. """
  321. return getattr(self, item)
  322. def __init__(self, task_attempt, task):
  323. assert task_attempt is not None
  324. self.task_attempt = task_attempt
  325. self.task = task
  326. self._init_attributes();
  327. def _init_attributes(self):
  328. self.taskType = self.task_attempt.taskID.taskID.taskTypeAsString
  329. self.attemptId = self.task_attempt.taskID.asString
  330. self.attemptId_short = "_".join(self.attemptId.split("_")[-2:])
  331. self.startTimeMs = self.task_attempt.startTime
  332. self.startTimeFormatted = format_unixtime_ms(self.task_attempt.startTime)
  333. self.finishTimeMs = self.task_attempt.finishTime
  334. self.finishTimeFormatted = format_unixtime_ms(self.task_attempt.finishTime)
  335. self.state = self.task_attempt.stateAsString.lower()
  336. self.taskTrackerId = self.task_attempt.taskTracker
  337. self.phase = self.task_attempt.phaseAsString
  338. self.progress = self.task_attempt.progress
  339. self.outputSize = self.task_attempt.outputSize
  340. self.shuffleFinishTimeMs = self.task_attempt.shuffleFinishTime
  341. self.shuffleFinishTimeFormatted = format_unixtime_ms(self.task_attempt.shuffleFinishTime)
  342. self.sortFinishTimeMs = self.task_attempt.sortFinishTime
  343. self.sortFinishTimeFormatted = format_unixtime_ms(self.task_attempt.sortFinishTime)
  344. self.mapFinishTimeMs = self.task_attempt.mapFinishTime # DO NOT USE, NOT VALID IN 0.20
  345. self.mapFinishTimeFormatted = format_unixtime_ms(self.task_attempt.mapFinishTime)
  346. self.counters = self.task_attempt.counters
  347. self.is_mr2 = False
  348. def get_tracker(self):
  349. try:
  350. tracker = Tracker.from_name(self.task.jt, self.taskTrackerId)
  351. return tracker
  352. except ttypes.TaskTrackerNotFoundException, e:
  353. LOGGER.warn("Tracker %s not found: %s" % (self.taskTrackerId, e))
  354. if LOGGER.isEnabledFor(logging.DEBUG):
  355. all_trackers = self.task.jt.all_task_trackers()
  356. for t in all_trackers.trackers:
  357. LOGGER.debug("Available tracker: %s" % (t.trackerName,))
  358. raise ttypes.TaskTrackerNotFoundException(
  359. _("Cannot look up TaskTracker %(id)s.") % {'id': self.taskTrackerId})
  360. def get_task_log(self):
  361. """
  362. get_task_log(task_id) -> (stdout_text, stderr_text, syslog_text)
  363. Retrieve the task log from the TaskTracker, at this url:
  364. http://<tracker_host>:<port>/tasklog?taskid=<attempt_id>
  365. Optional query string:
  366. &filter=<source> : where <source> is 'syslog', 'stdout', or 'stderr'.
  367. &start=<offset> : specify the start offset of the log section, when using a filter.
  368. &end=<offset> : specify the end offset of the log section, when using a filter.
  369. """
  370. tracker = self.get_tracker()
  371. url = urlunparse(('http',
  372. '%s:%s' % (tracker.host, tracker.httpPort),
  373. 'tasklog',
  374. None,
  375. 'attemptid=%s' % (self.attemptId,),
  376. None))
  377. LOGGER.info('Retrieving %s' % (url,))
  378. try:
  379. data = urllib2.urlopen(url)
  380. except urllib2.URLError:
  381. raise urllib2.URLError(_("Cannot retrieve logs from TaskTracker %(id)s.") % {'id': self.taskTrackerId})
  382. et = lxml.html.parse(data)
  383. log_sections = et.findall('body/pre')
  384. logs = [section.text or '' for section in log_sections]
  385. if len(logs) < 3:
  386. LOGGER.warn('Error parsing task attempt log for %s at "%s". Found %d (not 3) log sections' %
  387. (self.attemptId, url, len(log_sections)))
  388. err = _("Hue encountered an error while retrieving logs from '%s'.") % (url,)
  389. logs += [err] * (3 - len(logs))
  390. return logs
  391. class Tracker(object):
  392. def __getitem__(self, item):
  393. """
  394. For backwards-compatibility, resolve job["foo"] as job.foo.
  395. """
  396. return getattr(self, item)
  397. @staticmethod
  398. def from_name(jt, trackername):
  399. return Tracker(jt.task_tracker(trackername))
  400. def __init__(self, thrifttracker):
  401. self.tracker = thrifttracker
  402. self._init_attributes();
  403. def _init_attributes(self):
  404. self.trackerId = self.tracker.trackerName
  405. self.httpPort = self.tracker.httpPort
  406. self.host = self.tracker.host
  407. self.lastSeenMs = self.tracker.lastSeen
  408. self.lastSeenFormatted = format_unixtime_ms(self.tracker.lastSeen)
  409. self.totalVirtualMemory = self.tracker.totalVirtualMemory
  410. self.totalPhysicalMemory = self.tracker.totalPhysicalMemory
  411. self.availableSpace = self.tracker.availableSpace
  412. self.failureCount = self.tracker.failureCount
  413. self.mapCount = self.tracker.mapCount
  414. self.reduceCount = self.tracker.reduceCount
  415. self.maxMapTasks = self.tracker.maxMapTasks
  416. self.maxReduceTasks = self.tracker.maxReduceTasks
  417. self.taskReports = self.tracker.taskReports
  418. class Cluster(object):
  419. def __getitem__(self, item):
  420. """
  421. For backwards-compatibility, resolve job["foo"] as job.foo
  422. """
  423. return getattr(self, item)
  424. def __init__(self, jt):
  425. self.status = jt.cluster_status()
  426. self._init_attributes();
  427. def _init_attributes(self):
  428. self.mapTasksInProgress = self.status.mapTasks
  429. self.reduceTasksInProgress = self.status.reduceTasks
  430. self.maxMapTasks = self.status.maxMapTasks
  431. self.maxReduceTasks = self.status.maxReduceTasks
  432. self.usedHeapMemory = self.status.usedMemory
  433. self.maxHeapMemory = self.status.maxMemory
  434. self.clusterStartTimeMs = self.status.startTime
  435. self.clusterStartTimeFormatted = format_unixtime_ms(self.status.startTime)
  436. self.identifier = self.status.identifier
  437. self.taskTrackerExpiryInterval = self.status.taskTrackerExpiryInterval
  438. self.totalJobSubmissions = self.status.totalSubmissions
  439. self.state = self.status.stateAsString
  440. self.numActiveTrackers = self.status.numActiveTrackers
  441. self.activeTrackerNames = self.status.activeTrackerNames
  442. self.numBlackListedTrackers = self.status.numBlacklistedTrackers
  443. self.blacklistedTrackerNames = self.status.blacklistedTrackerNames
  444. self.hostname = self.status.hostname
  445. self.httpPort = self.status.httpPort
  446. # self.currentTimeMs = curtime
  447. # self.currentTimeFormatted = format_unixtime_ms(curtime)
  448. def get_jobconf(jt, jobid):
  449. """
  450. Returns a dict representation of the jobconf for the job corresponding
  451. to jobid. filter_keys is an optional list of configuration keys to filter on.
  452. """
  453. jid = jt.thriftjobid_from_string(jobid)
  454. # This will throw if the the jobconf can't be found
  455. xml_data = jt.get_job_xml(jid)
  456. return confparse.ConfParse(xml_data)
  457. def format_unixtime_ms(unixtime):
  458. """
  459. Format a unix timestamp in ms to a human readable string
  460. """
  461. if unixtime:
  462. return str(datetime.datetime.fromtimestamp(unixtime/1000).strftime("%x %X %Z"))
  463. else:
  464. return ""
  465. DOTS = re.compile("\.([a-z])")
  466. def dots_to_camel_case(dots):
  467. """
  468. Takes a string delimited with periods and returns a camel-case string.
  469. Example: dots_to_camel_case("foo.bar.baz") //returns fooBarBaz
  470. """
  471. def return_upper(match):
  472. return match.groups()[0].upper()
  473. return str(DOTS.sub(return_upper, dots))
  474. def get_path(hdfs_url):
  475. """
  476. Returns the path component of an HDFS url.
  477. """
  478. # urlparse is lame, and only "uses_netloc" for a certain
  479. # set of protocols. So we replace hdfs with gopher:
  480. if hdfs_url.startswith("hdfs://"):
  481. gopher_url = "gopher://" + hdfs_url[7:]
  482. path = urlparse(gopher_url)[2] # path
  483. return path
  484. else:
  485. return hdfs_url