Преглед изворни кода

HUE-7420 [jb] Impala Query integration

jdesjean пре 8 година
родитељ
комит
ce7f72ba4d

+ 12 - 0
apps/impala/src/impala/server.py

@@ -208,3 +208,15 @@ class ImpalaDaemonApi(object):
       return json.loads(resp)
     except ValueError, e:
       raise ImpalaDaemonApiException('ImpalaDaemonApi query_profile did not return valid JSON.')
+
+  def get_query_memory(self, query_id):
+    params = {
+      'query_id': query_id,
+      'json': 'true'
+    }
+
+    resp = self._root.get('query_memory', params=params)
+    try:
+      return json.loads(resp)
+    except ValueError, e:
+      raise ImpalaDaemonApiException('ImpalaDaemonApi query_memory did not return valid JSON.')

+ 1 - 1
apps/jobbrowser/src/jobbrowser/api2.py

@@ -124,4 +124,4 @@ def profile(request):
   response[app_property] = api.profile(app_id, app_type, app_property, app_filters)
   response['status'] = 0
 
-  return JsonResponse(response)
+  return JsonResponse(response)

+ 71 - 13
apps/jobbrowser/src/jobbrowser/apis/query_api.py

@@ -17,6 +17,7 @@
 
 import itertools
 import logging
+import re
 
 from django.utils.translation import ugettext as _
 
@@ -66,24 +67,65 @@ class QueryApi(Api):
       'total': jobs['num_in_flight_queries'] + jobs['num_executing_queries'] + jobs['num_waiting_queries']
     }
 
+  def time_in_ms(self, time, period):
+    if period == 'ns':
+      return float(time) / 1000
+    elif period == 'ms':
+      return float(time)
+    elif period == 's':
+      return float(time) * 1000
+    elif period == 'm':
+      return float(time) * 60000 #1000*60
+    elif period == 'h':
+      return float(time) * 3600000 #1000*60*60
+    else:
+      return float(time)
+
   def app(self, appid):
     api = get_impalad_api(user=self.user, url=self.server_url)
 
-    query = api.get_query(query_id=appid)
+    query = api.get_query_profile(query_id=appid)
+    user = re.search(r"^\s*User:\s*(.*)$", query['profile'], re.MULTILINE).group(1)
+    status = re.search(r"^\s*Query State:\s*(.*)$", query['profile'], re.MULTILINE).group(1)
+    stmt = re.search(r"^\s*Sql Statement:\s*(.*)$", query['profile'], re.MULTILINE).group(1)
+    partitions = re.findall(r"partitions=\s*(\d)+\s*\/\s*(\d)+", query['profile'])
+    end_time = re.search(r"^\s*End Time:\s*(.*)$", query['profile'], re.MULTILINE).group(1)
+    duration_1 = re.search(r"\s*Rows available:\s([\d.]*)(\w*)", query['profile'], re.MULTILINE)
+    duration_2 = re.search(r"\s*Request finished:\s([\d.]*)(\w*)", query['profile'], re.MULTILINE)
+    duration_3 = re.search(r"\s*Query Timeline:\s([\d.]*)(\w*)", query['profile'], re.MULTILINE)
+    submitted = re.search(r"^\s*Start Time:\s*(.*)$", query['profile'], re.MULTILINE).group(1)
+
+    progress = 0
+    if end_time:
+      progress = 100
+    elif partitions:
+      for partition in partitions:
+        progress += float(partition[0]) / float(partition[1])
+      progress /= len(partitions)
+      progress *= 100
+
+    duration = duration_1 or duration_2 or duration_3
+    if duration:
+      duration_ms = self.time_in_ms(duration.group(1), duration.group(2))
+    else:
+      duration_ms = 0
 
     common = {
         'id': appid,
-        'name': query['stmt'][:100] + ('...' if len(query['stmt']) > 100 else ''),
-        'status': query['status'],
-        'apiStatus': self._api_status(query['status']),
-        'progress': 50,
-        'duration': 10 * 3600,
-        'submitted': 0,
-        'type': 'NA',
+        'name': stmt,
+        'status': status,
+        'apiStatus': self._api_status(status),
+        'user': user,
+        'progress': progress,
+        'duration': duration_ms,
+        'submitted': submitted,
+        'type': 'queries'
     }
 
     common['properties'] = {
-      'properties': query
+      'memory': '',
+      'profile': '',
+      'plan': ''
     }
 
     return common
@@ -96,14 +138,30 @@ class QueryApi(Api):
   def logs(self, appid, app_type, log_name=None):
     return {'logs': ''}
 
+  def profile(self, appid, app_type, app_property, app_filters):
+    if app_property == 'memory':
+      return self._memory(appid, app_type, app_property, app_filters)
+    elif app_property == 'profile':
+      return self._query_profile(appid)
+    else:
+      return self._query(appid)
 
-  def profile(self, appid, app_type, app_property):
-    return {}
+  def _memory(self, appid, app_type, app_property, app_filters):
+    api = get_impalad_api(user=self.user, url=self.server_url)
+    return api.get_query_memory(query_id=appid);
+
+  def _query(self, appid):
+    api = get_impalad_api(user=self.user, url=self.server_url)
+    return api.get_query(query_id=appid)
+
+  def _query_profile(self, appid):
+    api = get_impalad_api(user=self.user, url=self.server_url)
+    return api.get_query_profile(query_id=appid)
 
   def _api_status(self, status):
-    if status in ['CREATING', 'CREATED', 'TERMINATING']:
+    if status in ['RUNNING', 'CREATED']:
       return 'RUNNING'
-    elif status in ['COMPLETED']:
+    elif status in ['FINISHED']:
       return 'SUCCEEDED'
     else:
       return 'FAILED' # INTERRUPTED , KILLED, TERMINATED and FAILED

+ 164 - 0
apps/jobbrowser/src/jobbrowser/static/jobbrowser/js/impala_dagre.js

@@ -0,0 +1,164 @@
+/* Builds and then renders a plan graph using Dagre / D3. The JSON for the current query
+is retrieved by an HTTP call, and then the graph of nodes and edges is built by walking
+over each plan fragment in turn. Plan fragments are connected wherever a node has a
+data_stream_target attribute.
+<script src="desktop/ext/js/d3.v3.js"></script>
+<script src="desktop/ext/js/dagre-d3-min.js"></script>
+Copied from https://github.com/apache/incubator-impala/blob/master/www/query_plan.tmpl
+*/
+function impalaDagre(id, dataSource) {
+  var d3 = window.d3v3;
+  var dagreD3 = window.dagreD3;
+  var g = new dagreD3.graphlib.Graph().setGraph({rankDir: "BT"});
+  var svg = d3.select("#"+id);
+  var inner = svg.select("g");
+  var _impalaDagree = {
+    start: function(refreshInterval) {
+      this._refreshInterval = refreshInterval;
+      // Force one refresh before starting the timer.
+      refresh();
+    },
+    stop: function() {
+      this._refreshInterval = null;
+    }
+  };
+
+  // Set up zoom support
+  var zoom = d3.behavior.zoom().on("zoom", function() {
+    inner.attr("transform", "translate(" + d3.event.translate + ")" +
+               "scale(" + d3.event.scale + ")");
+  });
+  svg.call(zoom);
+
+  // Set of colours to use, with the same colour used for every node in the same plan
+  // fragment.
+  var colours = ["#A9A9A9", "#FF8C00", "#8A2BE2", "#A52A2A", "#00008B", "#006400",
+                 "#228B22", "#4B0082", "#DAA520", "#008B8B", "#000000", "#DC143C"];
+
+  // Shades of red in order of intensity, used for colouring nodes by time taken
+  var cols_by_time = ["#000000", "#1A0500", "#330A00", "#4C0F00", "#661400", "#801A00",
+                      "#991F00", "#B22400", "#CC2900", "#E62E00", "#FF3300", "#FF4719"];
+
+  // Recursively build a list of edges and states that comprise the plan graph
+  function build(node, parent, edges, states, colour_idx, max_node_time) {
+    states.push({ "name": node["label"],
+                  "detail": node["label_detail"],
+                  "num_instances": node["num_instances"],
+                  "num_active": node["num_active"],
+                  "max_time": node["max_time"],
+                  "avg_time": node["avg_time"],
+                  "is_broadcast": node["is_broadcast"],
+                  "max_time_val": node["max_time_val"],
+                  "style": "fill: " + colours[colour_idx]});
+    if (parent) {
+      var label_val = "" + node["output_card"].toLocaleString();
+      edges.push({ start: node["label"], end: parent,
+                   style: { label: label_val }});
+    }
+    // Add an inter-fragment edge. We use a red dashed line to show that rows are crossing
+    // the fragment boundary.
+    if (node["data_stream_target"]) {
+      edges.push({ "start": node["label"],
+                   "end": node["data_stream_target"],
+                   "style": { label: "" + node["output_card"].toLocaleString(),
+                              style: "stroke: #f66; stroke-dasharray: 5, 5;"}});
+    }
+    max_node_time = Math.max(node["max_time_val"], max_node_time)
+    for (var i = 0; i < node["children"].length; ++i) {
+      max_node_time = build(
+        node["children"][i], node["label"], edges, states, colour_idx, max_node_time);
+    }
+    return max_node_time;
+  }
+
+  var is_first = true;
+
+  function renderGraph(plan) {
+    if (!plan || !plan.plan_nodes || !plan.plan_nodes.length) return;
+    var states = [];
+    var edges = [];
+    var colour_idx = 0;
+
+    var max_node_time = 0;
+    plan["plan_nodes"].forEach(function(parent) {
+      max_node_time = Math.max(
+        build(parent, null, edges, states, colour_idx, max_node_time));
+      // Pick a new colour for each plan fragment
+      colour_idx = (colour_idx + 1) % colours.length;
+    });
+
+    // Keep a map of names to states for use when processing edges.
+    var states_by_name = { };
+    states.forEach(function(state) {
+      // Build the label for the node from the name and the detail
+      var html = "<span>" + state.name + "</span><br/>";
+      html += "<span>" + state.detail + "</span><br/>";
+      html += "<span>" + state.num_instances + " instance";
+      if (state.num_instances > 1) {
+        html += "s";
+      }
+      html += "</span><br/>";
+      html += "<span>Max: " + state.max_time + ", avg: " + state.avg_time + "</span>";
+
+      var style = state.style;
+
+      // If colouring nodes by total time taken, choose a shade in the cols_by_time list
+      // with idx proportional to the max time of the node divided by the max time over all
+      // nodes.
+      /*if (document.getElementById("colour_scheme").checked) {
+        var idx = (cols_by_time.length - 1) * (state.max_time_val / (1.0 * max_node_time));
+        style = "fill: " + cols_by_time[Math.floor(idx)];
+      }*/
+      g.setNode(state.name, { "label": html,
+                              "labelType": "html",
+                              "style": style });
+      states_by_name[state.name] = state;
+    });
+
+    edges.forEach(function(edge) {
+      // Impala marks 'broadcast' as a property of the receiver, not the sender. We use
+      // '(BCAST)' to denote that a node is duplicating its output to all receivers.
+      if (states_by_name[edge.end].is_broadcast) {
+        edge.style.label += " \n(BCAST * " + states_by_name[edge.end].num_instances + ")";
+      }
+      g.setEdge(edge.start, edge.end, edge.style);
+    });
+
+    g.nodes().forEach(function(v) {
+      var node = g.node(v);
+      node.rx = node.ry = 5;
+    });
+
+    // Create the renderer
+    var render = new dagreD3.render();
+
+    // Run the renderer. This is what draws the final graph.
+    render(inner, g);
+
+    // Center the graph, but only the first time through (so as to not lose user zooms).
+    if (is_first) {
+      var initialScale = 0.75;
+      zoom.translate([((svg.attr("width") || $("#"+id).width()) - g.graph().width * initialScale) / 2, 20])
+        .scale(initialScale)
+        .event(svg);
+      svg.attr('height', Math.max(g.graph().height * initialScale + 40, 600));
+      is_first = false;
+    }
+
+  }
+
+  // Called periodically, fetches the plan JSON from Impala and passes it to renderGraph()
+  // for display.
+  function refresh() {
+    if (!$("#"+id).parent().hasClass("active")) return;
+    dataSource().then(function(data) {
+      if (!$("#"+id).parent().hasClass("active")) return;
+      renderGraph(data);
+      if (_impalaDagree._refreshInterval) {
+        setTimeout(refresh, _impalaDagree._refreshInterval);
+      }
+    });
+  }
+
+  return _impalaDagree;
+}

+ 82 - 26
apps/jobbrowser/src/jobbrowser/templates/job_browser.mako

@@ -57,6 +57,12 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
 <script src="${ static('desktop/ext/js/knockout-sortable.min.js') }"></script>
 <script src="${ static('desktop/js/ko.editable.js') }"></script>
 
+% if ENABLE_QUERY_BROWSER.get():
+<script src="${ static('desktop/ext/js/d3.v3.js') }"></script>
+<script src="${ static('desktop/ext/js/dagre-d3-min.js') }"></script>
+<script src="${ static('jobbrowser/js/impala_dagre.js') }"></script>
+% endif
+
 % if not is_mini:
 <div id="jobbrowserComponents" class="jobbrowser-components jobbrowser-full jb-panel">
 % else:
@@ -893,37 +899,54 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
 
       <ul class="nav nav-pills margin-top-20">
         <li>
-          <a href="#livy-session-page-statements${ SUFFIX }" data-bind="click: function(){ fetchProfile('properties'); $('a[href=\'#livy-session-page-statements${ SUFFIX }\']').tab('show'); }">
-            ${ _('Properties') }</a>
+          <a href="#queries-page-plan${ SUFFIX }" data-bind="click: function(){ $('a[href=\'#queries-page-plan${ SUFFIX }\']').tab('show'); }, event: {'shown': onQueriesPlanShown}">
+            ${ _('Plan') }</a>
+        </li>
+        <li>
+          <a href="#queries-page-stmt${ SUFFIX }" data-bind="click: function(){ $('a[href=\'#queries-page-stmt${ SUFFIX }\']').tab('show'); }">
+            ${ _('Query') }</a>
+        </li>
+        <li>
+          <a href="#queries-page-plan-text${ SUFFIX }" data-bind="click: function(){ $('a[href=\'#queries-page-plan-text${ SUFFIX }\']').tab('show'); }">
+            ${ _('Text Plan') }</a>
+        </li>
+        <li>
+          <a href="#queries-page-summary${ SUFFIX }" data-bind="click: onQueriesSummaryClick">
+            ${ _('Summary') }</a>
+        </li>
+        <li>
+          <a href="#queries-page-profile${ SUFFIX }" data-bind="click: function(){ fetchProfile('profile'); $('a[href=\'#queries-page-profile${ SUFFIX }\']').tab('show'); }">
+            ${ _('Profile') }</a>
+        </li>
+        <li>
+          <a href="#queries-page-memory${ SUFFIX }" data-bind="click: function(){ fetchProfile('memory'); $('a[href=\'#queries-page-memory${ SUFFIX }\']').tab('show'); }">
+            ${ _('Memory') }</a>
         </li>
       </ul>
 
       <div class="clearfix"></div>
 
       <div class="tab-content">
-        <div class="tab-pane active" id="livy-session-page-statements${ SUFFIX }">
-          <table id="actionsTable" class="datatables table table-condensed">
-            <thead>
-            <tr>
-              <th>${_('Id')}</th>
-              <th>${_('State')}</th>
-              <th>${_('Output')}</th>
-            </tr>
-            </thead>
-            <tbody data-bind="foreach: properties['statements']">
-              <tr data-bind="click: function() {  $root.job().id(id); $root.job().fetchJob(); }" class="pointer">
-                <td>
-                  <a data-bind="hueLink: '/jobbrowser/jobs/' + id(), clickBubble: false">
-                    <i class="fa fa-tasks"></i>
-                  </a>
-                </td>
-                <td data-bind="text: state"></td>
-                <td data-bind="text: output"></td>
-              </tr>
-            </tbody>
-          </table>
+        <div class="tab-pane" id="queries-page-plan${ SUFFIX }">
+          <svg style="border: 1px solid darkgray;width:100%;height:100%;" id="queries-page-plan-svg${ SUFFIX }">
+            <g/>
+          </svg>
+        </div>
+        <div class="tab-pane" id="queries-page-stmt${ SUFFIX }">
+          <pre data-bind="text: properties.plan().stmt"/>
+        </div>
+        <div class="tab-pane" id="queries-page-plan-text${ SUFFIX }">
+          <pre data-bind="text: properties.plan().plan"/>
+        </div>
+        <div class="tab-pane" id="queries-page-summary${ SUFFIX }">
+          <pre data-bind="text: properties.plan().summary"/>
+        </div>
+        <div class="tab-pane" id="queries-page-profile${ SUFFIX }">
+          <pre data-bind="text: properties.profile().profile"/>
+        </div>
+        <div class="tab-pane" id="queries-page-memory${ SUFFIX }">
+          <pre data-bind="text: properties.memory().mem_usage"/>
         </div>
-
       </div>
     </div>
   </div>
@@ -1687,6 +1710,34 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
       self.logs = ko.observable('');
 
       self.properties = ko.mapping.fromJS(job.properties || {});
+      self.onQueriesPlanShown = function () {
+        if (this._impalaDagre) {
+          this._impalaDagre.stop();
+        }
+        var self = this;
+        var dataSource = function () {
+          return new Promise(function(resolve, reject) {
+            self.fetchProfile('plan', function (data) {
+              resolve(data.plan.plan_json);
+            });
+          });
+        }
+        this._impalaDagre = impalaDagre('queries-page-plan-svg${ SUFFIX }', dataSource);
+        this._impalaDagre.start(5000);
+      };
+      self.onQueriesSummaryClick = function(){
+        $('a[href=\'#queries-page-summary${ SUFFIX }\']').tab('show');
+        var self = this;
+        var refresh = function () {
+          if (!$('#queries-page-summary${ SUFFIX }').hasClass('active')) {
+            return;
+          }
+          self.fetchProfile('plan', function (data) {
+            setTimeout(refresh, 5000);
+          });
+        };
+        setTimeout(refresh, 5000);
+      };
       self.mainType = ko.observable(vm.interface());
 
       self.coordinatorActions = ko.pureComputed(function() {
@@ -1888,7 +1939,9 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
 
             crumbs.push({'id': vm.job().id(), 'name': vm.job().name(), 'type': vm.job().type()});
             vm.resetBreadcrumbs(crumbs);
-
+            if (vm.job().type() === 'queries' && !$("#queries-page-plan${ SUFFIX }").parent().children().hasClass("active")) {
+              $("a[href=\'#queries-page-plan${ SUFFIX }\']").tab("show");
+            }
             %if not is_mini:
             if (vm.job().type() === 'workflow' && !vm.job().workflowGraphLoaded) {
               vm.job().updateWorkflowGraph();
@@ -1946,7 +1999,7 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
         });
       };
 
-      self.fetchProfile = function (name) {
+      self.fetchProfile = function (name, callback) {
         $.post("/jobbrowser/api/job/profile", {
           app_id: ko.mapping.toJSON(self.id),
           interface: ko.mapping.toJSON(vm.interface),
@@ -1956,6 +2009,9 @@ ${ commonheader("Job Browser", "jobbrowser", user, request) | n,unicode }
         }, function (data) {
           if (data.status == 0) {
             self.properties[name](data[name]);
+            if (callback) {
+              callback(data);
+            }
           } else {
             $(document).trigger("error", data.message);
           }

Разлика између датотеке није приказан због своје велике величине
+ 44 - 0
desktop/core/src/desktop/static/desktop/ext/js/dagre-d3-min.js


Неке датотеке нису приказане због велике количине промена