فهرست منبع

[hbase] Does not support int rowkeys

Romain Rigaux 12 سال پیش
والد
کامیت
f74687d
3فایلهای تغییر یافته به همراه43 افزوده شده و 27 حذف شده
  1. 21 13
      apps/hbase/src/hbase/api.py
  2. 1 1
      apps/hbase/src/hbase/server/hbase_lib.py
  3. 21 13
      apps/hbase/src/hbase/views.py

+ 21 - 13
apps/hbase/src/hbase/api.py

@@ -15,10 +15,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-try:
-  import json
-except ImportError:
-  import simplejson as json
+import json
 import logging
 import re
 import csv
@@ -36,11 +33,13 @@ from hbase import conf
 
 LOG = logging.getLogger(__name__)
 
-#format methods similar to Thrift API, for similarity with catch-all
+
+# Format methods similar to Thrift API, for similarity with catch-all
 class HbaseApi(object):
+
   def query(self, action, *args):
     try:
-      if hasattr(self,action):
+      if hasattr(self, action):
         return getattr(self, action)(*args)
       cluster = args[0]
       return self.queryCluster(action, cluster, *args[1:])
@@ -73,7 +72,7 @@ class HbaseApi(object):
       for cluster in clusters:
         if cluster["name"] == name:
           return cluster
-    except Exception, e:
+    except:
       pass
     raise PopupException(_("Cluster by the name of %s does not exist in configuration.") % name)
 
@@ -87,6 +86,14 @@ class HbaseApi(object):
                                   use_sasl=False,
                                   timeout_seconds=None)
 
+  def get(self, cluster, tableName, row, column, attributes):
+    client = self.connectCluster(cluster)
+    return client.get(tableName, smart_str(row), smart_str(column), attributes)
+
+  def getVerTs(self, cluster, tableName, row, column, timestamp, numVersions, attributesargs):
+    client = self.connectCluster(cluster)
+    return client.getVerTs(tableName, smart_str(row), smart_str(column), timestamp, numVersions, attributesargs)
+
   def createTable(self, cluster, tableName, *columns):
     client = self.connectCluster(cluster)
     client.createTable(tableName, [get_thrift_type('ColumnDescriptor')(name=column) for column in columns])
@@ -96,10 +103,10 @@ class HbaseApi(object):
     client = self.connectCluster(cluster)
     return [{'name': name,'enabled': client.isTableEnabled(name)} for name in client.getTableNames()]
 
-  def getRows(self, cluster, tableName, columns, startRowKey, numRows, prefix = False):
+  def getRows(self, cluster, tableName, columns, startRowKey, numRows, prefix=False):
     client = self.connectCluster(cluster)
     if prefix == False:
-      scanner = client.scannerOpen(tableName, startRowKey, columns, None)
+      scanner = client.scannerOpen(tableName, smart_str(startRowKey), columns, None)
     else:
       scanner = client.scannerOpenWithPrefix(tableName, smart_str(startRowKey), columns, None)
     data = client.scannerGetList(scanner, numRows)
@@ -107,12 +114,14 @@ class HbaseApi(object):
     return data
 
   def getAutocompleteRows(self, cluster, tableName, numRows, query):
+    query = smart_str(query)
     try:
       client = self.connectCluster(cluster)
       scan = get_thrift_type('TScan')(startRow=query, stopRow=None, timestamp=None, columns=[], caching=None, filterString="PrefixFilter('" + query + "') AND ColumnPaginationFilter(1,0)", batchSize=None)
       scanner = client.scannerOpenWithScan(tableName, scan, None)
       return [result.row for result in client.scannerGetList(scanner, numRows)]
-    except:
+    except Exception, e:
+      LOG.error('Autocomplete error: %s' % smart_str(e))
       return []
 
   def getRow(self, cluster, tableName, columns, startRowKey):
@@ -123,7 +132,7 @@ class HbaseApi(object):
 
   def getRowsFull(self, cluster, tableName, startRowKey, numRows):
     client = self.connectCluster(cluster)
-    return self.getRows(cluster, tableName, [column for column in client.getColumnDescriptors(tableName)], startRowKey, numRows)
+    return self.getRows(cluster, tableName, [smart_str(column) for column in client.getColumnDescriptors(tableName)], smart_str(startRowKey), numRows)
 
   def getRowFull(self, cluster, tableName, startRowKey, numRows):
     row = self.getRowsFull(cluster, tableName, smart_str(startRowKey), 1)
@@ -144,7 +153,7 @@ class HbaseApi(object):
     return client.mutateRow(tableName, smart_str(row), mutations, None)
 
   def deleteColumn(self, cluster, tableName, row, column):
-    return self.deleteColumns(cluster, tableName, smart_str(row), [column])
+    return self.deleteColumns(cluster, tableName, smart_str(row), [smart_str(column)])
 
   def putRow(self, cluster, tableName, row, data):
     client = self.connectCluster(cluster)
@@ -200,4 +209,3 @@ class HbaseApi(object):
       batches += [BatchMutation(row=row_key, mutations=mutations)]
     client.mutateRows(tableName, batches, None)
     return True
-

+ 1 - 1
apps/hbase/src/hbase/server/hbase_lib.py

@@ -48,4 +48,4 @@ def get_thrift_attributes(name):
   for spec in thrift_type.thrift_spec:
     if spec is not None:
       attrs[spec[2]] = spec[1]
-  return attrs
+  return attrs

+ 21 - 13
apps/hbase/src/hbase/views.py

@@ -15,13 +15,10 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-try:
-  import json
-except ImportError:
-  import simplejson as json
+
+import json
 import logging
 import re
-import base64
 import StringIO
 
 from avro import schema, datafile, io
@@ -43,30 +40,38 @@ LOG = logging.getLogger(__name__)
 def app(request):
   return render('app.mako', request, {})
 
-#action/cluster/arg1/arg2/arg3...
-def api_router(request, url): #on split, deserialize anything
+# action/cluster/arg1/arg2/arg3...
+def api_router(request, url): # On split, deserialize anything
+
   def safe_json_load(raw):
     try:
       return json.loads(re.sub(r'(?:\")([0-9]+)(?:\")', r'\1', str(raw)))
     except:
       return raw
+
   def deserialize(data):
     if type(data) == dict:
       special_type = get_thrift_type(data.pop('hue-thrift-type', ''))
       if special_type:
         return special_type(data)
+
     if hasattr(data, "__iter__"):
       for i, item in enumerate(data):
-        data[i] = deserialize(item) #sets local binding, needs to set in data
+        data[i] = deserialize(item) # Sets local binding, needs to set in data
     return data
-  url_params = [safe_json_load((arg, request.POST.get(arg[0:16], arg))[arg[0:15] == 'hbase-post-key-']) for arg in re.split(r'(?<!\\)/', url.strip('/'))] #deserialize later
+
+  url_params = [safe_json_load((arg, request.POST.get(arg[0:16], arg))[arg[0:15] == 'hbase-post-key-'])
+                for arg in re.split(r'(?<!\\)/', url.strip('/'))] # Deserialize later
+
   if request.POST.get('dest', False):
     url_params += [request.FILES.get(request.REQUEST.get('dest'))]
+
   return api_dump(HbaseApi().query(*url_params))
 
 def api_dump(response):
-  ignored_fields = ('thrift_spec', "__.+__")
+  ignored_fields = ('thrift_spec', '__.+__')
   trunc_limit = conf.TRUNCATE_LIMIT.get()
+
   def clean(data):
     try:
       json.dumps(data)
@@ -74,7 +79,7 @@ def api_dump(response):
     except:
       cleaned = {}
       lim = [0]
-      if isinstance(data, str): #not JSON dumpable, meaning some sort of bytestring or byte data
+      if isinstance(data, str): # Not JSON dumpable, meaning some sort of bytestring or byte data
         #detect if avro file
         if(data[:3] == '\x4F\x62\x6A'):
           #write data to file in memory
@@ -86,6 +91,7 @@ def api_dump(response):
           df_reader = datafile.DataFileReader(output, rec_reader)
           return json.dumps(clean([record for record in df_reader]))
         return base64.b64encode(data)
+
       if hasattr(data, "__iter__"):
         if type(data) is dict:
           for i in data:
@@ -100,7 +106,9 @@ def api_dump(response):
       else:
         for key in dir(data):
           value = getattr(data, key)
-          if value is not None and not hasattr(value, '__call__') and sum([int(bool(re.search(ignore, key))) for ignore in ignored_fields]) == 0:
+          if value is not None and not hasattr(value, '__call__') and sum([int(bool(re.search(ignore, key)))
+                                                                           for ignore in ignored_fields]) == 0:
             cleaned[key] = clean(value)
       return cleaned
-  return HttpResponse(json.dumps({ 'data': clean(response), 'truncated': True, 'limit': trunc_limit }), content_type="application/json")
+
+  return HttpResponse(json.dumps({ 'data': clean(response), 'truncated': True, 'limit': trunc_limit }), content_type="application/json")