analyze.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453
  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 base64
  18. import json
  19. import os
  20. import re
  21. import sys
  22. sys.path.append(os.path.join(os.path.dirname(__file__), "../..", 'gen-py'))
  23. from RuntimeProfile.ttypes import *
  24. from thrift.protocol import TCompactProtocol
  25. from thrift.transport import TTransport
  26. from libanalyze import dot
  27. from libanalyze import gjson as jj
  28. from libanalyze import models
  29. class Node(object):
  30. """Simple Node"""
  31. def __init__(self, val):
  32. super(Node, self).__init__()
  33. self.val = val
  34. self.children = []
  35. self.fragment = None
  36. self.fragment_instance = None
  37. self.pos = 0
  38. def add_child(self, c):
  39. self.children.append(c)
  40. def find_by_name(self, pattern):
  41. """Returns the first node whose name matches 'name'."""
  42. if re.search(pattern, self.val.name) is not None:
  43. return self
  44. for x in self.children:
  45. tmp = x.find_by_name(pattern)
  46. if tmp:
  47. return tmp
  48. def find_all_by_name(self, pattern):
  49. result = []
  50. if re.search(pattern, self.val.name):
  51. result.append(self)
  52. for x in self.children:
  53. result += x.find_all_by_name(pattern)
  54. return result
  55. def find_all_non_fragment_nodes(self):
  56. result = []
  57. if not self.is_fragment() and not self.is_fragment_instance():
  58. result.append(self)
  59. for x in self.children:
  60. result += x.find_all_non_fragment_nodes()
  61. return result
  62. def is_fragment(self):
  63. return re.search(r'(.*?Fragment) (F\d+)', self.val.name) is not None
  64. def is_fragment_instance(self):
  65. return re.search(r'Instance\s(.*?)\s\(host=(.*?)\)', self.val.name) is not None
  66. def is_regular(self):
  67. id = self.id()
  68. matches = id and re.search(r'[a-zA-Z]+', id)
  69. return id and matches is None
  70. def name(self):
  71. matches = re.search(r'(.*?)(\s+\(((dst_)?id)=(\d+)\))?$', self.val.name)
  72. if matches and matches.group(5):
  73. return matches.group(1)
  74. elif self.is_fragment():
  75. return re.search(r'(.*?Fragment) (F\d+)', self.val.name).group(1)
  76. else:
  77. return self.val.name
  78. def id(self):
  79. matches = re.search(r'(.*?)(\s+\(((dst_)?id)=(\d+)\))?$', self.val.name)
  80. if matches and matches.group(5) and not matches.group(4):
  81. return matches.group(5)
  82. elif self.is_fragment():
  83. return re.search(r'(.*?Fragment) (F\d+)', self.val.name).group(2)
  84. elif self.is_fragment_instance():
  85. return re.search(r'Instance\s(.*?)\s\(host=(.*?)\)', self.val.name).group(1)
  86. elif self.fragment:
  87. return self.fragment.id() + ' ' + str(self.pos)
  88. def is_plan_node(self):
  89. matches = re.search('(.*?)(\s+\(((dst_)?id)=(\d+)\))?$', self.val.name)
  90. return matches and not matches.group(4) and matches.group(5)
  91. def find_by_id(self, pattern):
  92. results = []
  93. if self.id() == pattern:
  94. results.append(self)
  95. for x in self.children:
  96. tmp = x.find_by_id(pattern)
  97. results += tmp
  98. return results
  99. def find_all_fragments(self):
  100. results = []
  101. if self.is_fragment_instance():
  102. results.append(self)
  103. for x in self.children:
  104. tmp = x.find_all_fragments()
  105. results += tmp
  106. return results
  107. def foreach_lambda(self, method, fragment=None, fragment_instance=None, pos=0):
  108. self.fragment = fragment
  109. self.fragment_instance = fragment_instance
  110. self.pos = pos
  111. if self.is_fragment():
  112. fragment = self
  113. elif self.is_fragment_instance():
  114. fragment_instance = self
  115. for idx, x in enumerate(self.children):
  116. x.foreach_lambda(method, fragment=fragment, fragment_instance=fragment_instance, pos=idx)
  117. method(self) # Post execution, because some results need child to have processed
  118. def find_metric_by_name(self, pattern):
  119. node = self
  120. ctr_map = node.counter_map()
  121. counters = []
  122. for k in ctr_map:
  123. if pattern == k:
  124. counters.append({'name': ctr_map[k].name, 'value': ctr_map[k].value,
  125. 'unit': ctr_map[k].unit, 'node': node})
  126. for k in node.child_counters_map():
  127. v = node.child_counters_map()[k]
  128. if pattern == k:
  129. parent = None
  130. if k in ctr_map:
  131. parent = {'name': ctr_map[k].name, 'value': ctr_map[k].value,
  132. 'unit': ctr_map[k].unit}
  133. for cc in v:
  134. counters.append({'name': ctr_map[cc].name, 'value': ctr_map[cc].value,
  135. 'unit': ctr_map[cc].unit, 'parent': parent, 'node': node})
  136. return counters
  137. def find_info_by_name(self, pattern):
  138. node = self
  139. ctr_map = node.val.info_strings
  140. counters = []
  141. if ctr_map.get(pattern):
  142. counters.append({'name': pattern, 'value': ctr_map.get(pattern), 'node': node})
  143. return counters
  144. # Only for fragments
  145. def is_averaged(self):
  146. return re.search(r"Averaged", self.val.name) is not None
  147. # Only for fragments
  148. def is_coordinator(self):
  149. return re.match(r'Coordinator', self.val.name) is not None
  150. # Only for fragments
  151. def host(self):
  152. if self.fragment_instance:
  153. c = self.fragment_instance
  154. elif self.fragment:
  155. c = self.fragment.children[0]
  156. else:
  157. return None
  158. m = re.search(r'Instance\s(.*?)\s\(host=(.*?)\)', c.val.name)
  159. if m:
  160. #frag.instance_id = m.group(1)
  161. #frag.host = m.group(2)
  162. #frag_node = c
  163. return m.group(2)
  164. def augmented_host(self):
  165. if self.fragment_instance:
  166. c = self.fragment_instance
  167. elif self.fragment:
  168. if self.fragment.is_averaged():
  169. return 'averaged'
  170. c = self.fragment.children[0]
  171. elif self.is_fragment():
  172. if self.is_averaged():
  173. return 'averaged'
  174. else:
  175. c = self.children[0]
  176. else:
  177. return None
  178. m = re.search(r'Instance\s(.*?)\s\(host=(.*?)\)', c.val.name)
  179. if m:
  180. #frag.instance_id = m.group(1)
  181. #frag.host = m.group(2)
  182. #frag_node = c
  183. return m.group(2)
  184. def info_strings(self):
  185. return self.val.info_strings
  186. def info_string_order(self):
  187. return self.val.info_strings_display_order
  188. def child_counters_map(self):
  189. return self.val.child_counters_map
  190. def counter_map(self):
  191. ctr = {}
  192. if self.val.counters:
  193. for c in self.val.counters:
  194. ctr[c.name] = c
  195. return ctr
  196. def metric_map(self):
  197. ctr = {}
  198. if self.val.counters:
  199. for c in self.val.counters:
  200. ctr[c.name] = { 'name': c.name, 'value': c.value, 'unit': c.unit }
  201. return ctr
  202. def repr(self, indent):
  203. buffer = indent + self.val.name + "\n"
  204. if self.val.info_strings:
  205. for k in self.val.info_strings:
  206. buffer += 2 * indent + " - " + k + "=" + \
  207. self.val.info_strings[k][:40] + "\n"
  208. if self.val.event_sequences:
  209. for s in self.val.event_sequences:
  210. buffer += 2 * indent + "- " + s.name + \
  211. " S[" + ", ".join(s.labels) + "]" + "\n"
  212. ctr = {}
  213. if self.val.counters:
  214. for c in self.val.counters:
  215. buffer += 2 * indent + c.name + ":" + str(c) + "\n"
  216. ctr[c.name] = c
  217. return buffer
  218. def decode_thrift(val):
  219. """Deserialize a binary string into the TRuntimeProfileTree structure"""
  220. transport = TTransport.TMemoryBuffer(val)
  221. protocol = TCompactProtocol.TCompactProtocol(transport)
  222. rp = TRuntimeProfileTree()
  223. rp.read(protocol)
  224. return rp
  225. def decompress(val):
  226. return val.decode("zlib")
  227. def summary(profile):
  228. summary = profile.find_by_name('Summary')
  229. execution_profile = profile.find_by_name('Execution Profile')
  230. counter_map = summary.counter_map()
  231. counter_map_execution_profile = execution_profile.counter_map()
  232. host_list = models.host_by_metric(profile, 'PeakMemoryUsage', exprs=[max])
  233. host_list = sorted(host_list, key=lambda x: x[1], reverse=True)
  234. peak_memory = models.TCounter(value=host_list[0][1], unit=3) if host_list else models.TCounter(value=0, unit=3) # The value is not always present
  235. return [{ 'key': 'PlanningTime', 'value': counter_map['PlanningTime'].value, 'unit': counter_map['PlanningTime'].unit }, {'key': 'RemoteFragmentsStarted', 'value': counter_map['RemoteFragmentsStarted'].value, 'unit': counter_map['RemoteFragmentsStarted'].unit}, {'key': 'TotalTime', 'value': counter_map_execution_profile['TotalTime'].value, 'unit': counter_map_execution_profile['TotalTime'].unit}, {'key': 'PeakMemoryUsage', 'value': peak_memory.value, 'unit': peak_memory.unit}]
  236. def metrics(profile):
  237. execution_profile = profile.find_by_name('Execution Profile')
  238. if not execution_profile:
  239. return {}
  240. counter_map = {}
  241. def get_metric(node, counter_map=counter_map):
  242. if not node.is_plan_node():
  243. return
  244. nid = node.id()
  245. if counter_map.get(nid) is None:
  246. counter_map[nid] = {}
  247. host = node.augmented_host()
  248. if host:
  249. counter_map[nid][host] = node.metric_map()
  250. else:
  251. counter_map[nid] = node.metric_map()
  252. execution_profile.foreach_lambda(get_metric)
  253. return counter_map
  254. def heatmap_by_host(profile, counter_name):
  255. rows = models.host_by_metric(profile,
  256. counter_name,
  257. exprs=[max, sum])
  258. # Modify the data to contain the relative data as well
  259. sum_sum = 0
  260. max_max = 0
  261. if (rows):
  262. sum_sum = float(sum([v[2] for v in rows]))
  263. max_max = float(max([v[1] for v in rows]))
  264. result = []
  265. for r in rows:
  266. result.append([r[0], float(r[1]), float(r[2]),
  267. float(r[1]) / float(max_max) if float(max_max) != 0 else 0,
  268. float(r[2]) / float(sum_sum) if float(sum_sum) != 0 else 0,
  269. rows.unit])
  270. return { 'data': result, 'max': max_max, 'unit': rows.unit }
  271. def parse(file_name):
  272. """Given a file_name, open the file and decode the first line of the file
  273. into the TRuntimeProfileTree structure."""
  274. with open(file_name) as fid:
  275. for line in fid:
  276. val = base64.decodestring(line.strip())
  277. try:
  278. val = decompress(val.strip())
  279. except:
  280. pass
  281. return decode_thrift(val)
  282. def parse_data(data):
  283. val = base64.decodestring(data)
  284. try:
  285. val = decompress(val.strip())
  286. except:
  287. pass
  288. return decode_thrift(val)
  289. def pre_order_traversal(nodes, index, level=0):
  290. # print index, nodes[index].num_children, nodes[index].name, level
  291. node = Node(nodes[index])
  292. pos = index
  293. for x in range(node.val.num_children):
  294. child_node, pos = pre_order_traversal(nodes, pos + 1, level + 1)
  295. node.add_child(child_node)
  296. return node, pos
  297. def analyze(profile):
  298. """The runtime profile tree is pre-order flattened"""
  299. node, _ = pre_order_traversal(profile.nodes, 0)
  300. return node
  301. def get_plan(profile):
  302. return profile.find_by_name("Sumary").val.info_strings["Plan"]
  303. def to_dot(profile):
  304. fragments = [
  305. x for x in profile.find_by_name("Execution Profile").children if re.search(
  306. "Averaged|Coordinator",
  307. x.val.name)]
  308. return dot.graph_to_dot(fragments)
  309. def to_json(profile):
  310. fragments = [
  311. x for x in profile.find_by_name("Execution Profile").children if re.search(
  312. "Averaged|Coordinator",
  313. x.val.name)]
  314. return json.dumps(jj.graph_to_json(fragments))
  315. def print_tree(node, level, indent):
  316. if level == 0:
  317. return
  318. print node.repr(indent)
  319. for c in node.children:
  320. print_tree(c, level - 1, indent + " ")
  321. if __name__ == '__main__':
  322. pass
  323. # from datetime import datetime
  324. # from dateutil.parser import parse as dtparse
  325. # db_engine = create_engine('sqlite:///profiles.db', echo=False)
  326. thrift = parse(sys.argv[1])
  327. root = analyze(thrift)
  328. # for n in root.find_by_name("Summary").info_strings():
  329. # print n
  330. #
  331. summary = root.find_by_name("Execution Profile")
  332. print_tree(summary, 4, " ")
  333. # db.Base.metadata.create_all(db_engine)
  334. # Session2 = sessionmaker(bind=db_engine)
  335. # query_id = "ea48c505d8604dc4:d4c46af6d57afa4"
  336. # from sqlalchemy.schema import *
  337. # impala_engine = create_engine('impala://mgrund-desktop.ca.cloudera.com/default', echo=False)
  338. # dstat = Table('dstat', MetaData(bind=impala_engine), autoload=True)
  339. # Session = sessionmaker(bind=impala_engine)
  340. # session = Session()
  341. # data = session.query(dstat).all()
  342. # sss = Session2()
  343. # # tt = db.Dstat.__table__
  344. # # cols = [c.key for c in db.Dstat.__table__.columns]
  345. # # for i, d in enumerate(data):
  346. # # sss.add(db.Dstat(**dict(zip(cols[1:], d))))
  347. # sss.commit()
  348. # from sqlalchemy.sql import func
  349. # #sss.query(db.RuntimeProfile).filter(db.RuntimeProfile.query_id==query_id)
  350. # instances = db.query_node_by_metric(sss, query_id, "HASH_JOIN_NODE", "TotalTime", 5)
  351. # for i in instances:
  352. # node = sss.query(db.Node).filter(db.Node.id==i[3]).one()
  353. # infos = {x.name: x.val for x in node.info_strings}
  354. # left = (dtparse(infos["StartTime"]) - datetime(1970,1,1)).total_seconds() + 7*3600
  355. # right = (dtparse(infos["StopTime"]) - datetime(1970,1,1)).total_seconds() + 7*3600
  356. # print "HASH_JOIN_NODE", i.fid, i.host, sss.query(func.avg(db.Dstat.usr),
  357. # func.max(db.Dstat.usr)).filter(db.Dstat.ts<=left,
  358. # db.Dstat.ts<=right).all()
  359. # #to_db(db_engine, root)
  360. #summary = root.find_by_name("Execution Profile")
  361. # #print summary.info_strings()["ExecSummary"]
  362. #print_tree(summary, 3, " ")
  363. # #list = root.find_all_by_name("CodeGen")
  364. # #print next(x for x in list[0].val.counters if x.name == "TotalTime").value / float(10**9)
  365. # #n = root.find_by_name("Averaged Fragment F01")
  366. # #print_tree(n, 2, " ")
  367. # #summary = root.find_by_name("ImpalaServer")
  368. # #print summary.val.info_strings.keys()
  369. # #print summary.val.info_strings["Plan"]
  370. # #print summary.val.info_strings["ExecSummary"]