Browse Source

HUE-8837 [catalog] Add get_databas,get_table, get_field for atlas to enable find_entity

Weixia 6 năm trước cách đây
mục cha
commit
6abe51acbb

+ 88 - 55
desktop/libs/metadata/src/metadata/catalog/atlas_client.py

@@ -124,6 +124,91 @@ class AtlasApi(Api):
 
     return nav_entity
 
+  def parse_atlas_response(self, atlas_response):
+    '''
+    REQUEST: hue:8889/metadata/api/navigator/find_entity?type=database&name=default
+    SAMPLE response for Navigator find_entity response
+    {"status": 0, "entity": {
+    "customProperties": null,
+    "deleteTime": null,
+     "fileSystemPath": "hdfs://nightly6x-1.vpc.cloudera.com:8020/user/hive/warehouse",
+     "description": null,
+     "params": null,
+      "type": "DATABASE",
+      "internalType": "hv_database",
+      "sourceType": "HIVE",
+      "tags": [],
+      "deleted": false, "technicalProperties": null,
+      "userEntity": false,
+      "originalDescription": "Default Hive database",
+      "metaClassName": "hv_database",
+      "properties": {"__cloudera_internal__hueLink": "https://nightly6x-1.vpc.cloudera.com:8889/hue/metastore/tables/default"},
+      "identity": "23",
+      "firstClassParentId": null,
+      "name": null,
+      "extractorRunId": "7##1",
+      "sourceId": "7",
+       "packageName": "nav",
+       "parentPath": null, "originalName": "default"}}
+    '''
+    response = {
+      "status": 0,
+      "entity": []
+    }
+    if not atlas_response['entities']:
+      LOG.error('No entities in atlas response to parse: %s' % json.dumps(atlas_response))
+    for atlas_entity in atlas_response['entities']:
+      response['entity'].append(self.adapt_atlas_entity_to_navigator(atlas_entity))
+    return response['entity'][0]
+
+  def get_database(self, name):
+    # Search with Atlas API for hive database with specific name
+    try:
+      dsl_query = '+'.join(['hive_db', 'where', 'name=%s']) % name
+      atlas_response = self._root.get('/v2/search/dsl?query=%s' % dsl_query, headers=self.__headers,
+                                      params=self.__params)
+      return self.parse_atlas_response(atlas_response)
+    except RestException, e:
+      LOG.error('Failed to search for entities with search query: %s' % dsl_query)
+      if e.code == 401:
+        raise CatalogAuthException(_('Failed to authenticate.'))
+      else:
+        raise CatalogApiException(e.message)
+
+  def get_table(self, database_name, table_name, is_view=False):
+    # Search with Atlas API for hive tables with specific name
+    # TODO: Need figure out way how to identify the cluster info for exact qualifiedName or use startsWith 'db.table.column'
+    try:
+      qualifiedName = '%s.%s@cl1' % (database_name, table_name)
+      dsl_query = '+'.join(['hive_table', 'where', 'qualifiedName=\"%s\"']) % qualifiedName
+      atlas_response = self._root.get('/v2/search/dsl?query=%s' % dsl_query, headers=self.__headers,
+                                      params=self.__params)
+      return self.parse_atlas_response(atlas_response)
+
+    except RestException, e:
+      LOG.error('Failed to search for entities with search query: %s' % dsl_query)
+      if e.code == 401:
+        raise CatalogAuthException(_('Failed to authenticate.'))
+      else:
+        raise CatalogApiException(e.message)
+
+  def get_field(self, database_name, table_name, field_name):
+    # Search with Atlas API for hive tables with specific qualified name
+    # TODO: Figure out how to identify the cluster info for exact qualifiedName
+    # TODO: query string for search with qualifiedName startsWith sys.test5.id
+    try:
+      qualifiedName = '%s.%s.%s@cl1' % (database_name, table_name, field_name)
+      dsl_query = '+'.join(['hive_column', 'where', 'qualifiedName=\"%s\"']) % qualifiedName
+      atlas_response = self._root.get('/v2/search/dsl?query=%s' % dsl_query, headers=self.__headers,
+                                      params=self.__params)
+      return self.parse_atlas_response(atlas_response)
+    except RestException, e:
+      LOG.error('Failed to search for entities with search query: %s' % dsl_query)
+      if e.code == 401:
+        raise CatalogAuthException(_('Failed to authenticate.'))
+      else:
+        raise CatalogApiException(e.message)
+
   def search_entities_interactive(self, query_s=None, limit=100, offset=0, facetFields=None, facetPrefix=None, facetRanges=None, filterQueries=None, firstClassEntitiesOnly=None, sources=None):
     try:
       response = {
@@ -164,13 +249,9 @@ class AtlasApi(Api):
       atlas_response = self._root.get('/v2/search/dsl?query=%s' % atlas_dsl_query)
 
       # Adapt Atlas entities to Navigator structure in the results
-      if 'entities' in atlas_response:
-        for atlas_entity in atlas_response['entities']:
-          response['results'].append(self.adapt_atlas_entity_to_navigator(atlas_entity))
+      return self.parse_atlas_response(atlas_response)
 
-      return response
     except RestException, e:
-      print(e)
       LOG.error('Failed to search for entities with search query: %s' % atlas_dsl_query)
       if e.code == 401:
         raise CatalogAuthException(_('Failed to authenticate.'))
@@ -189,58 +270,10 @@ class AtlasApi(Api):
       LOG.error(msg)
       raise CatalogApiException(e.message)
 
-
-  def find_entity(self, source_type, type, name, **filters):
-    """
-    GET /api/v3/entities?query=((sourceType:<source_type>)AND(type:<type>)AND(originalName:<name>))
-    http://cloudera.github.io/navigator/apidocs/v3/path__v3_entities.html
-    """
-    try:
-      params = self.__params
-
-      query_filters = {
-        'sourceType': source_type,
-        'originalName': name,
-        'deleted': 'false'
-      }
-
-      for key, value in filters.items():
-        query_filters[key] = value
-
-      filter_query = 'AND'.join('(%s:%s)' % (key, value) for key, value in query_filters.items())
-      filter_query = '%(type)s AND %(filter_query)s' % {
-        'type': '(type:%s)' % 'TABLE OR type:VIEW' if type == 'TABLE' else type, # Impala does not always say that a table is actually a view
-        'filter_query': filter_query
-      }
-
-      source_ids = self.get_cluster_source_ids()
-      if source_ids:
-        filter_query = source_ids + '(' + filter_query + ')'
-
-      params += (
-        ('query', filter_query),
-        ('offset', 0),
-        ('limit', 2),  # We are looking for single entity, so limit to 2 to check for multiple results
-      )
-
-      response = self._root.get('entities', headers=self.__headers, params=params)
-
-      if not response:
-        raise CatalogEntityDoesNotExistException('Could not find entity with query filters: %s' % str(query_filters))
-      elif len(response) > 1:
-        raise CatalogApiException('Found more than 1 entity with query filters: %s' % str(query_filters))
-
-      return response[0]
-    except RestException, e:
-      msg = 'Failed to find entity: %s' % str(e)
-      LOG.error(msg)
-      raise CatalogApiException(e.message)
-
-
   def get_entity(self, entity_id):
     """
-    GET /api/v3/entities/:id
-    http://cloudera.github.io/navigator/apidocs/v3/path__v3_entities_-id-.html
+    # TODO: get entity by Atlas __guid or qualifiedName
+    GET /v2/search/dsl?query=?
     """
     try:
       return self._root.get('entities/%s' % entity_id, headers=self.__headers, params=self.__params)

+ 71 - 0
desktop/libs/metadata/src/metadata/catalog_tests.py

@@ -34,11 +34,82 @@ from metadata import conf
 from metadata.conf import has_catalog, NAVIGATOR, get_navigator_auth_password, get_navigator_auth_username
 from metadata.catalog_api import _augment_highlighting
 from metadata.catalog.navigator_client import NavigatorApi
+from metadata.catalog.atlas_client import AtlasApi
+
 
 
 LOG = logging.getLogger(__name__)
 
 
+class TestAtlas(object):
+  integration = True
+
+  @classmethod
+  def setup_class(cls):
+    cls.client = make_logged_in_client(username='test', is_superuser=False)
+    cls.user = User.objects.get(username='test')
+    cls.user = rewrite_user(cls.user)
+    add_to_group('test')
+    grant_access("test", "test", "metadata")
+
+    if not is_live_cluster() or not has_catalog(cls.user):
+      raise SkipTest
+
+    cls.api = AtlasApi(cls.user)
+
+
+  @classmethod
+  def teardown_class(cls):
+    cls.user.is_superuser = False
+    cls.user.save()
+
+  def test_api_find_entity_with_type_hive_db(self, type='database', db_name='sys'):
+    # find_entity(source_type='HIVE', type='DATABASE', name='default')
+    '''
+    # query = "hive_db+where+name=sys+select+name,__guid"
+    {"queryType":"DSL","queryText":"hive_db where name=sys select name,__guid","attributes":{"name":["name","__guid"],
+    "values":[["sys","16cab673-e4b1-4ee6-83cf-c0017ed855ca"]]}}
+    '''
+    query = "hive_db+where+name=sys"
+    resp = self.client.get(reverse('metadata:catalog_find_entity'), {'type': type, 'name': db_name})
+    json_resp = json.loads(resp.content)
+    LOG.info("Hue response for find_entity with query: %s" % query)
+    LOG.info(json_resp)
+    assert_equal(0, json_resp['status'], json_resp)
+    assert_equal(json_resp['entity']['name'], db_name)
+
+  def test_api_find_entity_with_type_hive_table(self, type='table', table_name='test5', db_name="sys"):
+    '''
+    qualifiedName = '.'.join([database_name, name]) + "@cl1"
+    query = hive_column where qualifiedName='qualifiedName'
+    '''
+    qualifiedName = '.'.join([db_name, table_name]) + "@cl1"
+    query = "hive_column where qualifiedName = '%s'" % qualifiedName
+    resp = self.client.get(reverse('metadata:catalog_find_entity'),
+                           {'type': type, 'name': table_name, 'database': db_name})
+    json_resp = json.loads(resp.content)
+    LOG.info("Hue response for find_entity with query: %s" % query)
+    LOG.info(json_resp)
+    assert_equal(0, json_resp['status'], json_resp)
+    assert_equal(json_resp['entity']['name'], table_name)
+
+  def test_api_find_entity_with_type_hive_column(self, db_name='testdb2', table_name='test5', field_name='id',
+                                                 type='field'):
+    '''
+    qualifiedName = '.'.join([database_name, table_name, field_name]) + "@cl1"
+    query = hive_column where qualifiedName=''
+    '''
+    qualifiedName = '.'.join([db_name, table_name, field_name]) + "@cl1"
+    query = "hive_column where qualifiedName = '%s'" % qualifiedName
+    resp = self.client.get(reverse('metadata:catalog_find_entity'),
+                           {'type': type, 'name': field_name, 'database': db_name, 'table': table_name})
+    json_resp = json.loads(resp.content)
+    LOG.info("Hue response for find_entity with query: %s" % query)
+    LOG.info(json_resp)
+    assert_equal(0, json_resp['status'], json_resp)
+    assert_equal(json_resp['entity']['name'], field_name)
+
+
 class TestNavigator(object):
   integration = True
 

+ 11 - 11
desktop/libs/metadata/src/metadata/urls.py

@@ -25,17 +25,17 @@ from metadata import manager_api as metadata_manager_api
 
 # Catalog
 urlpatterns = [
-  url(r'^api/catalog/search_entities/?$', metadata_catalog_api.search_entities, name='search_entities'),
-  url(r'^api/catalog/search_entities_interactive/?$', metadata_catalog_api.search_entities_interactive, name='search_entities_interactive'),
-  url(r'^api/catalog/find_entity/?$', metadata_catalog_api.find_entity, name='find_entity'),
-  url(r'^api/catalog/get_entity/?$', metadata_catalog_api.get_entity, name='get_entity'),
-  url(r'^api/catalog/add_tags/?$', metadata_catalog_api.add_tags, name='add_tags'),
-  url(r'^api/catalog/delete_tags/?$', metadata_catalog_api.delete_tags, name='delete_tags'),
-  url(r'^api/catalog/list_tags/?$', metadata_catalog_api.list_tags, name='list_tags'),
-  url(r'^api/catalog/suggest/?$', metadata_catalog_api.suggest, name='suggest'),
-  url(r'^api/catalog/update_properties/?$', metadata_catalog_api.update_properties, name='update_properties'),
-  url(r'^api/catalog/delete_metadata_properties/?$', metadata_catalog_api.delete_metadata_properties, name='delete_metadata_properties'),
-  url(r'^api/catalog/lineage/?$', metadata_catalog_api.get_lineage, name='get_lineage'),
+  url(r'^api/catalog/search_entities/?$', metadata_catalog_api.search_entities, name='catalog_search_entities'),
+  url(r'^api/catalog/search_entities_interactive/?$', metadata_catalog_api.search_entities_interactive, name='catalog_search_entities_interactive'),
+  url(r'^api/catalog/find_entity/?$', metadata_catalog_api.find_entity, name='catalog_find_entity'),
+  url(r'^api/catalog/get_entity/?$', metadata_catalog_api.get_entity, name='catalog_get_entity'),
+  url(r'^api/catalog/add_tags/?$', metadata_catalog_api.add_tags, name='catalog_add_tags'),
+  url(r'^api/catalog/delete_tags/?$', metadata_catalog_api.delete_tags, name='catalog_delete_tags'),
+  url(r'^api/catalog/list_tags/?$', metadata_catalog_api.list_tags, name='catalog_list_tags'),
+  url(r'^api/catalog/suggest/?$', metadata_catalog_api.suggest, name='catalog_suggest'),
+  url(r'^api/catalog/update_properties/?$', metadata_catalog_api.update_properties, name='catalog_update_properties'),
+  url(r'^api/catalog/delete_metadata_properties/?$', metadata_catalog_api.delete_metadata_properties, name='catalog_delete_metadata_properties'),
+  url(r'^api/catalog/lineage/?$', metadata_catalog_api.get_lineage, name='catalog_get_lineage'),
 ]
 # Navigator API (deprecated, renamed to Catalog)
 urlpatterns += [