浏览代码

HUE-8948 [hive] Support transaction tables as examples

Romain 6 年之前
父节点
当前提交
2c1c3905e4

+ 27 - 0
apps/beeswax/data/tables_transactional.json

@@ -0,0 +1,27 @@
+[
+  {
+    "data_file": "sample_07.csv",
+    "create_hql": "CREATE TABLE `sample_07` (\n  `code` string ,\n  `description` string ,\n  `total_emp` int ,\n  `salary` int )\nSTORED AS parquet\nTBLPROPERTIES ('transactional'='true', 'transactional_properties'='insert_only')\n",
+    "table_name": "sample_07"
+  },
+  {
+    "data_file": "sample_08.csv",
+    "create_hql": "CREATE TABLE `sample_08` (\n  `code` string ,\n  `description` string ,\n  `total_emp` int ,\n  `salary` int )\nSTORED AS parquet\nTBLPROPERTIES ('transactional'='true', 'transactional_properties'='insert_only')\n",
+    "table_name": "sample_08"
+  },
+  {
+    "data_file": "customers.csv",
+    "create_hql": "CREATE TABLE `customers` (\n `id` INT,\n `name` STRING,\n `email_preferences` STRUCT<`email_format`:STRING, `frequency`:STRING, `categories`:STRUCT<`promos`:BOOLEAN, `surveys`:BOOLEAN>>,\n `addresses` MAP<STRING, STRUCT<`street_1`:STRING, `street_2`:STRING, `city`:STRING, `state`:STRING, `zip_code`:STRING>>,\n `orders` ARRAY<STRUCT<`order_id`:STRING, `order_date`:STRING, `items`:ARRAY<STRUCT<`product_id`:INT, `sku`:STRING, `name`:STRING, `price`:DOUBLE, `qty`:INT>>>>)\nSTORED AS parquet\nTBLPROPERTIES ('transactional'='true', 'transactional_properties'='insert_only')",
+    "table_name": "customers"
+  },
+  {
+    "partition_files": {
+      "`date`='2015-11-18'": "web_logs_1.csv",
+      "`date`='2015-11-19'": "web_logs_2.csv",
+      "`date`='2015-11-20'": "web_logs_3.csv",
+      "`date`='2015-11-21'": "web_logs_4.csv"
+    },
+    "create_hql": "CREATE TABLE `web_logs`  (  `_version_` bigint,   `app` string,   `bytes` smallint,   `city` string,   `client_ip` string,   `code` tinyint,   `country_code` string,   `country_code3` string,   `country_name` string,   `device_family` string,   `extension` string,   `latitude` float,   `longitude` float,   `method` string,   `os_family` string,   `os_major` string,   `protocol` string,   `record` string,   `referer` string,   `region_code` bigint,   `request` string,   `subapp` string,   `time` string,   `url` string,   `user_agent` string,   `user_agent_family` string,   `user_agent_major` string,   `id` string)\nPARTITIONED BY (  `date` string  )\nSTORED AS parquet\nTBLPROPERTIES ('transactional'='true', 'transactional_properties'='insert_only')",
+    "table_name": "web_logs"
+  }
+]

+ 2 - 2
apps/beeswax/src/beeswax/hive_site.py

@@ -181,8 +181,8 @@ def get_use_sasl():
 
 
 
 
 def has_concurrency_support():
 def has_concurrency_support():
-  '''For SQL transactions like INSERT, DELETE, UPDATE'''
-  return get_conf().get(_CNF_HIVE_SUPPORT_CONCURRENCY, 'FALSE').upper() == 'TRUE'
+  '''For SQL transactions like INSERT, DELETE, UPDATE since Hive 3.'''
+  return get_conf().get(_CNF_HIVE_SUPPORT_CONCURRENCY, 'TRUE').upper() == 'TRUE'
 
 
 
 
 def _parse_hive_site():
 def _parse_hive_site():

+ 17 - 0
apps/beeswax/src/beeswax/management/commands/__init__.py

@@ -0,0 +1,17 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.

+ 35 - 15
apps/beeswax/src/beeswax/management/commands/beeswax_install_examples.py

@@ -16,6 +16,7 @@
 # limitations under the License.
 # limitations under the License.
 
 
 from builtins import object
 from builtins import object
+import csv
 import logging
 import logging
 import os
 import os
 import pwd
 import pwd
@@ -35,11 +36,13 @@ from useradmin.models import get_default_user_group, install_sample_user
 import beeswax.conf
 import beeswax.conf
 from beeswax.models import SavedQuery, HQL, IMPALA
 from beeswax.models import SavedQuery, HQL, IMPALA
 from beeswax.design import hql_query
 from beeswax.design import hql_query
+from beeswax.hive_site import has_concurrency_support
 from beeswax.server import dbms
 from beeswax.server import dbms
 from beeswax.server.dbms import get_query_server_config, QueryServerException
 from beeswax.server.dbms import get_query_server_config, QueryServerException
 
 
 
 
 LOG = logging.getLogger(__name__)
 LOG = logging.getLogger(__name__)
+MAX_INSERTED_ROWS = 1000
 
 
 
 
 class InstallException(Exception):
 class InstallException(Exception):
@@ -60,7 +63,7 @@ class Command(BaseCommand):
       db_name = options.get('db_name', 'default')
       db_name = options.get('db_name', 'default')
       user = options['user']
       user = options['user']
 
 
-    tables = options['tables'] if 'tables' in options else 'tables.json'
+    tables = options['tables'] if 'tables' in options else ('tables_transactional.json' if has_concurrency_support() else 'tables.json')
 
 
     exception = None
     exception = None
 
 
@@ -76,7 +79,7 @@ class Command(BaseCommand):
 
 
     if exception is not None:
     if exception is not None:
       pretty_msg = None
       pretty_msg = None
-      
+
       if "AlreadyExistsException" in exception.message:
       if "AlreadyExistsException" in exception.message:
         pretty_msg = _("SQL table examples already installed.")
         pretty_msg = _("SQL table examples already installed.")
       if "Permission denied" in exception.message:
       if "Permission denied" in exception.message:
@@ -84,7 +87,7 @@ class Command(BaseCommand):
 
 
       if pretty_msg is not None:
       if pretty_msg is not None:
         raise PopupException(pretty_msg)
         raise PopupException(pretty_msg)
-      else: 
+      else:
         raise exception
         raise exception
 
 
   def _install_tables(self, django_user, app_name, db_name, tables):
   def _install_tables(self, django_user, app_name, db_name, tables):
@@ -199,25 +202,42 @@ class SampleTable(object):
 
 
     hql = LOAD_PARTITION_HQL % {'tablename': self.name, 'partition_spec': partition_spec, 'filepath': hdfs_root_destination}
     hql = LOAD_PARTITION_HQL % {'tablename': self.name, 'partition_spec': partition_spec, 'filepath': hdfs_root_destination}
     LOG.info('Running load query: %s' % hql)
     LOG.info('Running load query: %s' % hql)
-    self._load_data_to_table(django_user, hql, hdfs_file_destination)
+    self._load_data_to_table(django_user, hql)
 
 
 
 
   def load(self, django_user):
   def load(self, django_user):
     """
     """
     Upload data to HDFS home of user then load (aka move) it into the Hive table (in the Hive metastore in HDFS).
     Upload data to HDFS home of user then load (aka move) it into the Hive table (in the Hive metastore in HDFS).
     """
     """
-    LOAD_HQL = \
-      """
-      LOAD DATA INPATH
-      '%(filename)s' OVERWRITE INTO TABLE %(tablename)s
-      """
-
-    hdfs_root_destination = self._get_hdfs_root_destination(django_user)
-    hdfs_file_destination = self._upload_to_hdfs(django_user, self._contents_file, hdfs_root_destination)
+    if has_concurrency_support():
+      with open(self._contents_file) as f:
+        data = f.read()
+        dialect = csv.Sniffer().sniff(data)
+        reader = csv.reader(data.splitlines(), delimiter=dialect.delimiter)
+
+        rows = [', '.join("'%s'" % col.replace("'", "\\'") for col in row) for row in reader][:MAX_INSERTED_ROWS]
+        hql = \
+          """
+          INSERT INTO TABLE %(tablename)s
+          VALUES %(values)s
+          """ % {
+            'tablename': self.name,
+            'values': ', '.join('(%s)' % row for row in rows)
+          }
+    else:
+      hdfs_root_destination = self._get_hdfs_root_destination(django_user)
+      hdfs_file_destination = self._upload_to_hdfs(django_user, self._contents_file, hdfs_root_destination)
+      hql = \
+        """
+        LOAD DATA INPATH
+        '%(filename)s' OVERWRITE INTO TABLE %(tablename)s
+        """ % {
+          'tablename': self.name,
+          'filename': hdfs_file_destination
+        }
 
 
-    hql = LOAD_HQL % {'tablename': self.name, 'filename': hdfs_file_destination}
     LOG.info('Running load query: %s' % hql)
     LOG.info('Running load query: %s' % hql)
-    self._load_data_to_table(django_user, hql, hdfs_file_destination)
+    self._load_data_to_table(django_user, hql)
 
 
 
 
   def _check_file_contents(self, filepath):
   def _check_file_contents(self, filepath):
@@ -270,7 +290,7 @@ class SampleTable(object):
     return hdfs_destination
     return hdfs_destination
 
 
 
 
-  def _load_data_to_table(self, django_user, hql, hdfs_destination):
+  def _load_data_to_table(self, django_user, hql):
     LOG.info('Loading data into table "%s"' % (self.name,))
     LOG.info('Loading data into table "%s"' % (self.name,))
     query = hql_query(hql)
     query = hql_query(hql)
 
 

+ 71 - 0
apps/beeswax/src/beeswax/management/commands/beeswax_install_examples_tests.py

@@ -0,0 +1,71 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+
+from mock import patch, Mock, MagicMock
+from nose.tools import assert_equal, assert_not_equal, assert_true, assert_false
+
+from desktop.auth.backend import rewrite_user
+from desktop.lib.django_test_util import make_logged_in_client
+from desktop.lib.test_utils import add_to_group, grant_access
+
+from beeswax.management.commands.beeswax_install_examples import SampleTable, Command
+
+from django.contrib.auth.models import User
+
+
+LOG = logging.getLogger(__name__)
+
+
+class TestTransactionalTables():
+
+  def setUp(self):
+    self.client = make_logged_in_client(username="test", groupname="default", recreate=True, is_superuser=False)
+
+    self.user = rewrite_user(User.objects.get(username="test"))
+    grant_access("test", "default", "notebook")
+
+
+  def test_load_sample_07_with_concurrency_support(self):
+
+    table_data =   {
+      "data_file": "sample_07.csv",
+      "create_hql": "CREATE TABLE `sample_07` (\n  `code` string ,\n  `description` string ,\n  `total_emp` int ,\n  `salary` int )\nSTORED AS parquet\nTBLPROPERTIES ('transactional'='true', 'transactional_properties'='insert_only')\n",
+      "table_name": "sample_07"
+    }
+
+    with patch('beeswax.server.dbms.get') as get:
+      with patch('beeswax.management.commands.beeswax_install_examples.has_concurrency_support') as has_concurrency_support:
+        has_concurrency_support.return_value = True
+
+        SampleTable(table_data, 'beeswax', 'default').load(self.user)
+
+        get.assert_called()
+
+  def test_load_tables_concurrency_support(self):
+
+    with patch('beeswax.server.dbms.get') as get:
+      with patch('beeswax.management.commands.beeswax_install_examples.has_concurrency_support') as has_concurrency_support:
+        has_concurrency_support.return_value = True
+
+        cmd = Command()
+        # cmd.handle(app_name='beeswax', db_name='default', user=self.user)
+        cmd._install_tables(self.user, 'beeswax', 'default', 'tables_transactional.json')
+
+        get.assert_called()