Browse Source

HUE-8704 [jb] Improve profile performance.

jdesjean 6 năm trước cách đây
mục cha
commit
074c509

+ 41 - 14
desktop/libs/libanalyze/src/libanalyze/analyze.py

@@ -43,13 +43,19 @@ class Node(object):
     self.fragment_instance = None
     self.plan_node = None
     self.pos = 0
+    self.node_name = None
+    self.node_id = None
+    self.node_is_fragment = None
+    self.node_is_fragment_instance = None
+    self.node_is_regular = None
+    self.node_is_plan_node = None
 
   def add_child(self, c):
     self.children.append(c)
 
   def find_by_name(self, pattern):
     """Returns the first node whose name matches 'name'."""
-    if re.search(pattern, self.val.name) is not None:
+    if self.val.name.find(pattern) >= 0:
       return self
 
     for x in self.children:
@@ -59,7 +65,7 @@ class Node(object):
 
   def find_all_by_name(self, pattern):
     result = []
-    if re.search(pattern, self.val.name):
+    if self.val.name.find(pattern) >= 0:
       result.append(self)
     for x in self.children:
       result += x.find_all_by_name(pattern)
@@ -74,45 +80,66 @@ class Node(object):
     return result
 
   def is_fragment(self):
-    return re.search(r'(.*?Fragment) (F\d+)', self.val.name) is not None
+    if self.node_is_fragment is not None:
+      return self.node_is_fragment
+    self.node_is_fragment = re.search(r'(.*?Fragment) (F\d+)', self.val.name) is not None
+    return self.node_is_fragment
 
   def is_fragment_instance(self):
-    return re.search(r'Instance\s(.*?)\s\(host=(.*?)\)', self.val.name) is not None
+    if self.node_is_fragment_instance is not None:
+      return self.node_is_fragment_instance
+    self.node_is_fragment_instance = re.search(r'Instance\s(.*?)\s\(host=(.*?)\)', self.val.name) is not None
+    return self.node_is_fragment_instance
 
   def is_regular(self):
+    if self.node_is_regular is not None:
+      return self.node_is_regular
     id = self.id()
     matches = id and re.search(r'^\d*$', id)
-    return id and matches
+    self.node_is_regular = id and matches
+    return self.node_is_regular
 
   def name(self):
+    if self.node_name:
+      return self.node_name
     matches = re.search(r'(.*?)(\s+\(((dst_)?id)=(\d+)\))?$', self.val.name)
     if matches and matches.group(5):
-      return matches.group(1)
+      self.node_name = matches.group(1)
     elif self.is_fragment():
-      return re.search(r'(.*?Fragment) (F\d+)', self.val.name).group(1)
+      self.node_name = re.search(r'(.*?Fragment) (F\d+)', self.val.name).group(1)
     else:
       matches = re.search(r'(.*?)(\s+\(.*?\))?$', self.val.name)
       if matches.group(2):
-        return matches.group(1)
+        self.node_name = matches.group(1)
       else:
-        return self.val.name
+        self.node_name = self.val.name
+    return self.node_name
+
 
   def id(self):
+    if self.node_id:
+      return self.node_id
     matches = re.search(r'(.*?)(\s+\(((dst_)?id)=(\d+)\))?$', self.val.name)
     if matches and matches.group(5) and not matches.group(4):
-      return matches.group(5)
+      self.node_id = matches.group(5)
     elif self.is_fragment():
-      return re.search(r'(.*?Fragment) (F\d+)', self.val.name).group(2)
+      self.node_id = re.search(r'(.*?Fragment) (F\d+)', self.val.name).group(2)
     elif self.is_fragment_instance():
-      return re.search(r'Instance\s(.*?)\s\(host=(.*?)\)', self.val.name).group(1)
+      self.node_id = re.search(r'Instance\s(.*?)\s\(host=(.*?)\)', self.val.name).group(1)
     elif self.fragment:
-      return self.fragment.id() + ' ' + str(self.pos)
+      self.node_id = self.fragment.id() + ' ' + str(self.pos)
+    return self.node_id
 
   def is_plan_node(self):
+    if self.node_is_plan_node is not None:
+      self.node_is_plan_node
     matches = re.search('(.*?)(\s+\(((dst_)?id)=(\d+)\))?$', self.val.name)
-    return matches and not matches.group(4) and matches.group(5)
+    self.node_is_plan_node = matches and not matches.group(4) and matches.group(5)
+    return self.node_is_plan_node
 
   def find_by_id(self, pattern):
+    if self.nodes and self.nodes.get(pattern):
+      return self.nodes[pattern]
     results = []
     if self.id() == pattern:
       results.append(self)

Những thai đổi đã bị hủy bỏ vì nó quá lớn
+ 4 - 4
desktop/libs/libanalyze/src/libanalyze/analyze_test.py


+ 28 - 21
desktop/libs/libanalyze/src/libanalyze/rules.py

@@ -622,7 +622,7 @@ class TopDownAnalysis:
             has_spilled = False
 
             # Make sure to substract the wait time for the exchange node
-            if is_plan_node and re.search(r'EXCHANGE_NODE', node.val.name) is not None:
+            if is_plan_node and node.node_name == 'EXCHANGE_NODE':
                 async_time = counter_map.get('AsyncTotalTime', models.TCounter(value=0)).value
                 inactive_time = counter_map['InactiveTotalTime'].value
                 if inactive_time == 0:
@@ -630,38 +630,35 @@ class TopDownAnalysis:
                   inactive_time = dequeue.counter_map().get('DataWaitTime', models.TCounter(value=0)).value if dequeue else 0
                 local_time = counter_map['TotalTime'].value - inactive_time - async_time
                 child_time = counter_map['TotalTime'].value - local_time
-            if re.search(r'KrpcDataStreamSender', node.val.name) is not None and node.fragment_instance:
+            elif node.node_name == 'KrpcDataStreamSender' and node.fragment_instance:
               local_time = counter_map.get('SerializeBatchTime', models.TCounter(value=0)).value
               child_time = counter_map['TotalTime'].value - local_time
-            if re.search(r'HBASE_SCAN_NODE', node.val.name):
+            elif node.node_name == 'HBASE_SCAN_NODE':
               local_time = counter_map['TotalTime'].value - counter_map.get('TotalRawHBaseReadTime(*)', models.TCounter(value=0)).value
               child_time = counter_map['TotalTime'].value - local_time
-            if re.search(r'KUDU_SCAN_NODE', node.val.name):
+            elif node.node_name == 'KUDU_SCAN_NODE':
               child_time = counter_map.get('KuduClientTime', models.TCounter(value=0)).value
               local_time = counter_map['TotalTime'].value
-            if re.search(r'HDFS_SCAN_NODE', node.val.name):
+            elif node.node_name == 'HDFS_SCAN_NODE':
               child_time = counter_map.get('TotalRawHdfsReadTime(*)', models.TCounter(value=0)).value
               local_time = counter_map['TotalTime'].value
-            if re.search(r'Buffer pool', node.val.name):
+            elif node.node_name == 'Buffer pool':
               local_time = counter_map.get('WriteIoWaitTime', models.TCounter(value=0)).value + counter_map.get('ReadIoWaitTime', models.TCounter(value=0)).value + counter_map.get('AllocTime', models.TCounter(value=0)).value
-            if counter_map.get('SpilledPartitions', 0) > 0:
-              has_spilled = True
-
-            if re.search(r'AGGREGATION', node.val.name):
+            elif node.node_name == 'AGGREGATION':
               grouping_aggregator = node.find_by_name('GroupingAggregator')
               if grouping_aggregator and grouping_aggregator.counter_map().get('SpilledPartitions', models.TCounter(value=0)).value > 0:
                 has_spilled = True
-
-            # For Hash Join, if the "LocalTime" metrics
-            if is_plan_node and re.search(r'HASH_JOIN_NODE', node.val.name) is not None:
-                hash_join_builder = node.find_by_name('Hash Join Builder')
-                if hash_join_builder and hash_join_builder.counter_map().get('SpilledPartitions', models.TCounter(value=0)).value > 0:
-                  has_spilled = True
-                if ("LocalTime" in counter_map):
-                    local_time = counter_map["LocalTime"].value
-                else:
-                    local_time = counter_map["ProbeTime"].value +\
-                        counter_map["BuildTime"].value
+            elif is_plan_node and node.node_name == 'HASH_JOIN_NODE': # For Hash Join, if the "LocalTime" metrics
+              hash_join_builder = node.find_by_name('Hash Join Builder')
+              if hash_join_builder and hash_join_builder.counter_map().get('SpilledPartitions', models.TCounter(value=0)).value > 0:
+                has_spilled = True
+              if ("LocalTime" in counter_map):
+                  local_time = counter_map["LocalTime"].value
+              else:
+                  local_time = counter_map["ProbeTime"].value +\
+                      counter_map["BuildTime"].value
+            if counter_map.get('SpilledPartitions', 0) > 0:
+              has_spilled = True
 
             # Add two virtual metrics for local_time and child_time
             if has_spilled:
@@ -673,6 +670,16 @@ class TopDownAnalysis:
             node.val.counters.append(models.TCounter(name='LocalTime', value=local_time, unit=5))
             node.val.counters.append(models.TCounter(name='ChildTime', value=child_time, unit=5))
 
+        nodes = {}
+        def create_map(node, nodes=nodes):
+          nid = node.id()
+          if nid:
+            if not nodes.get(nid):
+              nodes[nid] = []
+            nodes[nid].append(node)
+        profile.foreach_lambda(create_map)
+        profile.nodes = nodes
+
         profile.foreach_lambda(add_host)
 
     def run(self, profile):

Một số tệp đã không được hiển thị bởi vì quá nhiều tập tin thay đổi trong này khác