Browse Source

HUE-5491 [metadata] Gather Hive table stats

Romain Rigaux 9 years ago
parent
commit
1ecc4b3

+ 35 - 1
desktop/libs/metadata/src/metadata/optimizer_api.py

@@ -35,6 +35,12 @@ from metadata.optimizer_client import OptimizerApi
 LOG = logging.getLogger(__name__)
 
 
+try:
+  from beeswax.api import get_table_stats
+except ImportError, e:
+  LOG.warn("Hive lib not enabled")
+
+
 def error_handler(view_fn):
   def decorator(*args, **kwargs):
     try:
@@ -303,7 +309,35 @@ def upload_history(request):
 
   api = OptimizerApi()
 
-  response['upload_history'] = api.upload(queries=queries, source_platform=source_platform)
+  response['upload_history'] = api.upload(data=queries, data_type='queries', source_platform=source_platform)
+  response['status'] = 0
+
+  return JsonResponse(response)
+
+
+@require_POST
+@error_handler
+def upload_table_stats(request):
+  response = {'status': -1}
+
+  db_tables = json.loads(request.POST.get('dbTables'), '[]')
+  source_platform = request.POST.get('sourcePlatform', 'hive')
+
+  data = []
+  for db_table in db_tables:
+    path = _get_table_name(db_table)
+
+    try:
+      table = get_table_stats(request, database=path['database'], table=path['table'])
+      stats = dict((stat['data_type'], stat['comment']) for stat in json.loads(table.content)['stats'])
+
+      data.append((path['table'], stats.get('numRows', -1)))
+    except Exception, e:
+      LOG.warning('Skipping upload of %s: %s' % (db_table, e))
+
+  api = OptimizerApi()
+
+  response['upload_history'] = api.upload(data=data, data_type='table_stats', source_platform=source_platform)
   response['status'] = 0
 
   return JsonResponse(response)

+ 89 - 39
desktop/libs/metadata/src/metadata/optimizer_client.py

@@ -51,6 +51,70 @@ class OptimizerApiException(PopupException):
 
 class OptimizerApi(object):
 
+  UPLOAD = {
+    'queries': {
+      'headers': ['SQL_ID', 'ELAPSED_TIME', 'SQL_FULLTEXT'],
+      'file_headers': """{
+    "fileLocation": "%(query_file)s",
+    "tenant": "%(tenant)s",
+    "fileName": "%(query_file_name)s",
+    "sourcePlatform": "%(source_platform)s",
+    "colDelim": ",",
+    "rowDelim": "\\n",
+    "headerFields": [
+        {
+            "count": 0,
+            "coltype": "SQL_ID",
+            "use": true,
+            "tag": "",
+            "name": "SQL_ID"
+        },
+        {
+            "count": 0,
+            "coltype": "NONE",
+            "use": true,
+            "tag": "",
+            "name": "ELAPSED_TIME"
+        },
+        {
+            "count": 0,
+            "coltype": "SQL_QUERY",
+            "use": true,
+            "tag": "",
+            "name": "SQL_FULLTEXT"
+        }
+    ]
+}"""
+    },
+    'table_stats': {
+        'headers': ['SQL_ID', 'ELAPSED_TIME', 'SQL_FULLTEXT'],
+        'file_headers': """{
+    "fileLocation": "%(query_file)s",
+    "tenant": "%(tenant)s",
+    "fileName": "%(query_file_name)s",
+    "sourcePlatform": "%(source_platform)s",
+    "colDelim": ",",
+    "rowDelim": "\\n",
+    "headerFields": [
+        {
+            "count": 0,
+            "coltype": "NONE",
+            "use": true,
+            "tag": "",
+            "name": "TABLE_NAME"
+        },
+        {
+            "count": 0,
+            "coltype": "NONE",
+            "use": true,
+            "tag": "",
+            "name": "NUM_ROWS"
+        }
+    ]
+}"""
+    }
+  }
+
   def __init__(self, api_url=None, product_name=None, product_secret=None, ssl_cert_ca_verify=OPTIMIZER.SSL_CERT_CA_VERIFY.get(), product_auth_secret=None):
     self._api_url = (api_url or get_optimizer_url()).strip('/')
     self._email = OPTIMIZER.EMAIL.get()
@@ -146,8 +210,15 @@ class OptimizerApi(object):
       raise PopupException(e, title=_('Error while accessing Optimizer'))
 
 
-  def upload(self, queries, source_platform='generic', workload_id=None):
-    f_queries_path = NamedTemporaryFile(suffix='.csv')
+  def upload(self, data, data_type='queries', source_platform='generic', workload_id=None):
+    if data_type == 'table_stats':
+      data_headers = OptimizerApi.HEADERS_UPLOAD_TABLE_STATS
+      data_suffix = '.log'
+    else:
+      data_headers = OptimizerApi.HEADERS_UPLOAD_QUERIES
+      data_suffix = '.csv'
+
+    f_queries_path = NamedTemporaryFile(suffix=data_suffix)
     f_format_path = NamedTemporaryFile(suffix='.json')
     f_queries_path.close()
     f_format_path.close() # Reopened as real file below to work well with the command
@@ -157,43 +228,18 @@ class OptimizerApi(object):
       f_format = open(f_format_path.name, 'w+')
 
       try:
-        content_generator = OptimizerDataAdapter(queries)
+        content_generator = OptimizerDataAdapter(data, data_type=data_type)
         queries_csv = export_csvxls.create_generator(content_generator, 'csv')
 
         for row in queries_csv:
           f_queries.write(row)
 
-        f_format.write("""{
-    "fileLocation": "%(query_file)s",
-    "tenant": "%(tenant)s",
-    "fileName": "%(query_file_name)s",
-    "sourcePlatform": "hive",
-    "colDelim": ",",
-    "rowDelim": "\\n",
-    "headerFields": [
-        {
-            "count": 0,
-            "coltype": "SQL_ID",
-            "use": true,
-            "tag": "",
-            "name": "SQL_ID"
-        },
-        {
-            "count": 0,
-            "coltype": "NONE",
-            "use": true,
-            "tag": "",
-            "name": "ELAPSED_TIME"
-        },
-        {
-            "count": 0,
-            "coltype": "SQL_QUERY",
-            "use": true,
-            "tag": "",
-            "name": "SQL_FULLTEXT"
-        }
-    ]
-}""" % {'tenant': self._product_name, 'query_file': f_queries.name, 'query_file_name': os.path.basename(f_queries.name)})
+        f_format.write(data_headers % {
+            'source_platform': source_platform,
+            'tenant': self._product_name,
+            'query_file': f_queries.name,
+            'query_file_name': os.path.basename(f_queries.name)
+        })
 
       finally:
         f_queries.close()
@@ -312,12 +358,16 @@ class OptimizerApi(object):
     return self._exec('get-top-data-bases', args)
 
 
-def OptimizerDataAdapter(queries):
-  headers = ['SQL_ID', 'ELAPSED_TIME', 'SQL_FULLTEXT']
-  if queries and len(queries[0]) == 3:
-    rows = queries
+def OptimizerDataAdapter(data, data_type='queries'):
+  headers = OptimizerApi.UPLOAD[data_type]['headers']
+
+  if data_type == 'table_stats':
+    rows = data
   else:
-    rows = ([str(uuid.uuid4()), 0.0, q] for q in queries)
+    if data and len(data[0]) == 3:
+      rows = data
+    else:
+      rows = ([str(uuid.uuid4()), 0.0, q] for q in data)
 
   yield headers, rows
 

+ 17 - 1
desktop/libs/metadata/src/metadata/optimizer_client_tests.py

@@ -109,7 +109,7 @@ class TestOptimizerApi(object):
         "select mgr.name from mgr where mgr.reports > 10 group by mgr.state;"
     ]
 
-    resp = self.api.upload(queries=queries)
+    resp = self.api.upload(data=queries, data_type='queries')
 
     assert_equal('status' in resp, resp)
     assert_equal('state' in resp['status'], resp)
@@ -123,6 +123,22 @@ class TestOptimizerApi(object):
     assert_equal('workloadId' in resp['status'], resp)
 
 
+  def test_upload_table_stats(self):
+    stats = [
+        "TABLE_NAME,NUM_ROWS",
+        "TEST_TABLE,10",
+        "TEST1,110",
+    ]
+
+    resp = self.api.upload(data=stats, data_type='table_stats')
+
+    assert_equal('status' in resp, resp)
+    assert_equal('state' in resp['status'], resp)
+    assert_equal('workloadId' in resp['status'], resp)
+
+    assert_true(resp['status']['state'] in ('WAITING', 'FINISHED', 'FAILED'), resp['status']['state'])
+
+
   def test_top_tables(self):
     database_name = 'default'
     resp = self.api.top_tables(database_name=database_name)

+ 3 - 2
desktop/libs/metadata/src/metadata/urls.py

@@ -36,8 +36,9 @@ urlpatterns = patterns('metadata.navigator_api',
 
 # Optimizer API
 urlpatterns += patterns('metadata.optimizer_api',
-  url(r'^api/optimizer/upload_history/?$', 'upload_history', name='upload_history'),
-  url(r'^api/optimizer/upload_status/?$', 'upload_status', name='upload_status'),
+  url(r'^api/optimizer/upload/history/?$', 'upload_history', name='upload_history'),
+  url(r'^api/optimizer/upload/table_stats/?$', 'upload_table_stats', name='upload_table_stats'),
+  url(r'^api/optimizer/upload/status/?$', 'upload_status', name='upload_status'),
 
   #v2
   url(r'^api/optimizer/get_tenant/?$', 'get_tenant', name='get_tenant'),

+ 2 - 2
desktop/libs/notebook/src/notebook/static/notebook/js/notebook.ko.js

@@ -1422,7 +1422,7 @@ var EditorViewModel = (function() {
     self.loadQueryHistory = function (n) {
       logGA('load_query_history');
 
-      $.post("/metadata/api/optimizer/upload_history", {
+      $.post("/metadata/api/optimizer/upload/history", {
         n: typeof n != "undefined" ? n : null,
         sourcePlatform: self.type()
       }, function(data) {
@@ -1436,7 +1436,7 @@ var EditorViewModel = (function() {
     };
 
     self.watchUploadStatus = function (workloadId) {
-      $.post("/metadata/api/optimizer/upload_status", {
+      $.post("/metadata/api/optimizer/upload/status", {
         workloadId: workloadId
       }, function(data) {
         if (data.status == 0) {