Browse Source

[impala] Refactor Impala API and DBMS query server out of beeswax

Jenny Kim 10 years ago
parent
commit
e89da22824

+ 13 - 50
apps/beeswax/src/beeswax/server/dbms.py

@@ -24,7 +24,6 @@ from django.utils.encoding import force_unicode
 from django.utils.translation import ugettext as _
 
 from desktop.lib.django_util import format_preserving_redirect
-from desktop.lib.i18n import smart_str
 from desktop.lib.parameterization import substitute_variables
 from filebrowser.views import location_to_url
 
@@ -38,11 +37,6 @@ from beeswax.models import QueryHistory, QUERY_TYPES
 
 LOG = logging.getLogger(__name__)
 
-try:
-  from impala.dbms import ImpalaDbms 
-except ImportError, e:
-  LOG.info('Impala app enabled: %s' % e)
-
 
 DBMS_CACHE = {}
 DBMS_CACHE_LOCK = threading.Lock()
@@ -51,9 +45,6 @@ def get(user, query_server=None):
   global DBMS_CACHE
   global DBMS_CACHE_LOCK
 
-  # Avoid circular dependency
-  from beeswax.server.hive_server2_lib import HiveServerClientCompatible, HiveServerClient
-
   if query_server is None:
     query_server = get_query_server_config()
 
@@ -62,7 +53,14 @@ def get(user, query_server=None):
     DBMS_CACHE.setdefault(user.username, {})
 
     if query_server['server_name'] not in DBMS_CACHE[user.username]:
-      DBMS_CACHE[user.username][query_server['server_name']] = HiveServer2Dbms(HiveServerClientCompatible(HiveServerClient(query_server, user)), QueryHistory.SERVER_TYPE[1][0])
+      # Avoid circular dependency
+      from beeswax.server.hive_server2_lib import HiveServerClientCompatible, HiveServerClient
+
+      if query_server['server_name'] == 'impala':
+        from impala.dbms import ImpalaDbms
+        DBMS_CACHE[user.username][query_server['server_name']] = ImpalaDbms(HiveServerClientCompatible(HiveServerClient(query_server, user)), QueryHistory.SERVER_TYPE[1][0])
+      else:
+        DBMS_CACHE[user.username][query_server['server_name']] = HiveServer2Dbms(HiveServerClientCompatible(HiveServerClient(query_server, user)), QueryHistory.SERVER_TYPE[1][0])
 
     return DBMS_CACHE[user.username][query_server['server_name']]
   finally:
@@ -71,22 +69,8 @@ def get(user, query_server=None):
 
 def get_query_server_config(name='beeswax', server=None):
   if name == 'impala':
-    from impala.conf import SERVER_HOST as IMPALA_SERVER_HOST, SERVER_PORT as IMPALA_SERVER_PORT, \
-        IMPALA_PRINCIPAL, IMPERSONATION_ENABLED, QUERYCACHE_ROWS, QUERY_TIMEOUT_S, AUTH_USERNAME as IMPALA_AUTH_USERNAME, AUTH_PASSWORD as IMPALA_AUTH_PASSWORD, \
-        SESSION_TIMEOUT_S
-
-    query_server = {
-        'server_name': 'impala',
-        'server_host': IMPALA_SERVER_HOST.get(),
-        'server_port': IMPALA_SERVER_PORT.get(),
-        'principal': IMPALA_PRINCIPAL.get(),
-        'impersonation_enabled': IMPERSONATION_ENABLED.get(),
-        'querycache_rows': QUERYCACHE_ROWS.get(),
-        'QUERY_TIMEOUT_S': QUERY_TIMEOUT_S.get(),
-        'SESSION_TIMEOUT_S': SESSION_TIMEOUT_S.get(),
-        'auth_username': IMPALA_AUTH_USERNAME.get(),
-        'auth_password': IMPALA_AUTH_PASSWORD.get()
-    }
+    from impala.dbms import get_query_server_config as impala_query_server_config
+    query_server = impala_query_server_config()
   else:
     kerberos_principal = hive_site.get_hiveserver2_kerberos_principal(HIVE_SERVER_HOST.get())
 
@@ -320,15 +304,10 @@ class HiveServer2Dbms(object):
     if not table.is_view:
       limit = min(100, BROWSE_PARTITIONED_TABLE_LIMIT.get())
 
-      if column or nested: # Could do column for any type, then nested with partitions 
-        if self.server_name == 'impala':
-          select_clause, from_clause = ImpalaDbms.get_nested_select(database, table.name, column, nested)
-          hql = 'SELECT %s FROM %s LIMIT %s' % (select_clause, from_clause, limit)
+      if table.partition_keys:  # Filter on max # of partitions for partitioned tables
+        hql = self._get_sample_partition_query(database, table, limit)
       else:
-        if table.partition_keys:  # Filter on max # of partitions for partitioned tables
-          hql = self._get_sample_partition_query(database, table, limit)
-        else:
-          hql = "SELECT * FROM `%s`.`%s` LIMIT %s" % (database, table.name, limit)
+        hql = "SELECT * FROM `%s`.`%s` LIMIT %s" % (database, table.name, limit)
 
       if hql:
         query = hql_query(hql)
@@ -502,22 +481,6 @@ class HiveServer2Dbms(object):
     return self.execute_query(query, design)
 
 
-  def invalidate_tables(self, database, tables):
-    handle = None
-
-    for table in tables:
-      try:
-        hql = "INVALIDATE METADATA `%s`.`%s`" % (database, table,)
-        query = hql_query(hql, database, query_type=QUERY_TYPES[1])
-
-        handle = self.execute_and_wait(query, timeout_sec=10.0)
-      except Exception, e:
-        LOG.warn('Refresh tables cache out of sync: %s' % smart_str(e))
-      finally:
-        if handle:
-          self.close(handle)
-
-
   def drop_database(self, database):
     return self.execute_statement("DROP DATABASE `%s`" % database)
 

+ 7 - 8
apps/impala/src/impala/views.py → apps/impala/src/impala/api.py

@@ -23,27 +23,26 @@ import logging
 import json
 
 from desktop.lib.django_util import JsonResponse
-from desktop.context_processors import get_app_name
 
-from beeswax.server import dbms
-from beeswax.server.dbms import get_query_server_config
+from beeswax.server import dbms as beeswax_dbms
+
+from impala import dbms
 
 
 LOG = logging.getLogger(__name__)
 
 
 def refresh_tables(request):
-  app_name = get_app_name(request)
-  query_server = get_query_server_config(app_name)
-  db = dbms.get(request.user, query_server=query_server)
+  query_server = dbms.get_query_server_config()
+  db = beeswax_dbms.get(request.user, query_server=query_server)
 
   response = {'status': 0, 'message': ''}
 
   if request.method == "POST":
     try:
       database = json.loads(request.POST['database'])
-      added = json.loads(request.POST['added'])
-      removed = json.loads(request.POST['removed'])
+      added = json.loads(request.POST.get('added', []))
+      removed = json.loads(request.POST.get('removed', []))
 
       db.invalidate_tables(database, added + removed)
     except Exception, e:

+ 98 - 23
apps/impala/src/impala/dbms.py

@@ -19,38 +19,39 @@ import logging
 
 from desktop.lib.i18n import smart_str
 
+from beeswax.conf import BROWSE_PARTITIONED_TABLE_LIMIT
 from beeswax.design import hql_query
+from beeswax.models import QUERY_TYPES
+from beeswax.server.dbms import HiveServer2Dbms
+
+from impala import conf
 
 
 LOG = logging.getLogger(__name__)
 
 
-class ImpalaDbms():
+def get_query_server_config():
+  query_server = {
+        'server_name': 'impala',
+        'server_host': conf.SERVER_HOST.get(),
+        'server_port': conf.SERVER_PORT.get(),
+        'principal': conf.IMPALA_PRINCIPAL.get(),
+        'impersonation_enabled': conf.IMPERSONATION_ENABLED.get(),
+        'querycache_rows': conf.QUERYCACHE_ROWS.get(),
+        'QUERY_TIMEOUT_S': conf.QUERY_TIMEOUT_S.get(),
+        'SESSION_TIMEOUT_S': conf.SESSION_TIMEOUT_S.get(),
+        'auth_username': conf.AUTH_USERNAME.get(),
+        'auth_password': conf.AUTH_PASSWORD.get()
+  }
 
-  def get_histogram(self, database, table, column, nested=None):
-    """
-    Returns the results of an Impala SELECT histogram() FROM query for a given column or nested type.
+  debug_query_server = query_server.copy()
+  debug_query_server['auth_password_used'] = bool(debug_query_server.pop('auth_password'))
+  LOG.debug("Query Server: %s" % debug_query_server)
 
-    Assumes that the column/nested type is scalar.
-    """
-    results = []
+  return query_server
 
-    hql = self.get_histogram_query(database, table, column, nested)
-    query = hql_query(hql)
-    handle = self.execute_and_wait(query, timeout_sec=5.0)
-
-    if handle:
-      result = self.fetch(handle)
-      try:
-        histogram = list(result.rows())[0][0]  # actual histogram results is in first-and-only result row
-        unique_values = set(histogram.split(', '))
-        results = list(unique_values)
-      except IndexError, e:
-        LOG.warn('Failed to get histogram results, result set has unexpected format: %s' % smart_str(e))
-      finally:
-        self.close(handle)
 
-    return results
+class ImpalaDbms(HiveServer2Dbms):
 
   @classmethod
   def get_nested_select(cls, database, table, column, nested=None):
@@ -83,4 +84,78 @@ class ImpalaDbms():
   def get_histogram_query(cls, database, table, column, nested=None):
     select_clause, from_clause = cls.get_nested_select(database, table, column, nested)
     return 'SELECT histogram(%s) FROM %s' % (select_clause, from_clause)
-        
+
+
+  def invalidate_tables(self, database, tables=None):
+    handle = None
+
+    try:
+      if tables:
+        for table in tables:
+          hql = "INVALIDATE METADATA `%s`.`%s`" % (database, table,)
+          print hql
+          query = hql_query(hql, database, query_type=QUERY_TYPES[1])
+          handle = self.execute_and_wait(query, timeout_sec=10.0)
+      else:  # call INVALIDATE on entire DB to pick up newly created tables
+        hql = "INVALIDATE METADATA `%s`" % database
+        print hql
+        query = hql_query(hql, database, query_type=QUERY_TYPES[1])
+        handle = self.execute_and_wait(query, timeout_sec=10.0)
+    except Exception, e:
+      LOG.warn('Refresh tables cache out of sync: %s' % smart_str(e))
+    finally:
+      if handle:
+        self.close(handle)
+
+
+  def get_sample(self, database, table, column=None, nested=None):
+    result = None
+    hql = None
+
+    if not table.is_view:
+      limit = min(100, BROWSE_PARTITIONED_TABLE_LIMIT.get())
+
+      if column or nested: # Could do column for any type, then nested with partitions
+        select_clause, from_clause = ImpalaDbms.get_nested_select(database, table.name, column, nested)
+        hql = 'SELECT %s FROM %s LIMIT %s' % (select_clause, from_clause, limit)
+      else:
+        if table.partition_keys:  # Filter on max # of partitions for partitioned tables
+          hql = self._get_sample_partition_query(database, table, limit)
+        else:
+          hql = "SELECT * FROM `%s`.`%s` LIMIT %s" % (database, table.name, limit)
+
+      if hql:
+        query = hql_query(hql)
+        handle = self.execute_and_wait(query, timeout_sec=5.0)
+
+        if handle:
+          result = self.fetch(handle, rows=100)
+          self.close(handle)
+
+    return result
+
+
+  def get_histogram(self, database, table, column, nested=None):
+    """
+    Returns the results of an Impala SELECT histogram() FROM query for a given column or nested type.
+
+    Assumes that the column/nested type is scalar.
+    """
+    results = []
+
+    hql = self.get_histogram_query(database, table, column, nested)
+    query = hql_query(hql)
+    handle = self.execute_and_wait(query, timeout_sec=5.0)
+
+    if handle:
+      result = self.fetch(handle)
+      try:
+        histogram = list(result.rows())[0][0]  # actual histogram results is in first-and-only result row
+        unique_values = set(histogram.split(', '))
+        results = list(unique_values)
+      except IndexError, e:
+        LOG.warn('Failed to get histogram results, result set has unexpected format: %s' % smart_str(e))
+      finally:
+        self.close(handle)
+
+    return results

+ 13 - 15
apps/impala/src/impala/tests.py

@@ -39,7 +39,6 @@ from beeswax.tests import _make_query
 from hadoop.pseudo_hdfs4 import get_db_prefix, is_live_cluster
 
 from impala import conf
-from impala.conf import SERVER_HOST
 from impala.dbms import ImpalaDbms
 
 
@@ -232,6 +231,7 @@ class TestImpalaIntegration:
     assert_true(data['properties'].get('http_addr'))
 
 
+
 # Could be refactored with SavedQuery.create_empty()
 def create_saved_query(app_name, owner):
     query_type = SavedQuery.TYPES_MAPPING[app_name]
@@ -295,17 +295,15 @@ def test_ssl_validate():
         reset()
 
 
-class TestImpalaDbms():
-
-  def test_get_impala_nested_select(self):
-    assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'id', None), ('id', '`default`.`customers`'))
-    assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'email_preferences', 'categories/promos/'),
-                 ('email_preferences.categories.promos', '`default`.`customers`'))
-    assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'addresses', 'key'),
-                 ('key', '`default`.`customers`.`addresses`'))
-    assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'addresses', 'value/street_1/'),
-                 ('street_1', '`default`.`customers`.`addresses`'))
-    assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'orders', 'item/order_date'),
-                 ('order_date', '`default`.`customers`.`orders`'))
-    assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'orders', 'item/items/item/product_id'),
-                 ('product_id', '`default`.`customers`.`orders`.`items`'))
+def test_get_impala_nested_select():
+  assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'id', None), ('id', '`default`.`customers`'))
+  assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'email_preferences', 'categories/promos/'),
+               ('email_preferences.categories.promos', '`default`.`customers`'))
+  assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'addresses', 'key'),
+               ('key', '`default`.`customers`.`addresses`'))
+  assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'addresses', 'value/street_1/'),
+               ('street_1', '`default`.`customers`.`addresses`'))
+  assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'orders', 'item/order_date'),
+               ('order_date', '`default`.`customers`.`orders`'))
+  assert_equal(ImpalaDbms.get_nested_select('default', 'customers', 'orders', 'item/items/item/product_id'),
+               ('product_id', '`default`.`customers`.`orders`.`items`'))

+ 1 - 1
apps/impala/src/impala/urls.py

@@ -20,7 +20,7 @@ from django.conf.urls import patterns, url
 from beeswax.urls import urlpatterns as beeswax_urls
 
 
-urlpatterns = patterns('impala.views',
+urlpatterns = patterns('impala.api',
   url(r'^api/refresh_tables$', 'refresh_tables', name='refresh_tables'),
 )