Procházet zdrojové kódy

HUE-8948 [hive] Support transaction tables as examples

Romain před 6 roky
rodič
revize
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():
-  '''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():

+ 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.
 
 from builtins import object
+import csv
 import logging
 import os
 import pwd
@@ -35,11 +36,13 @@ from useradmin.models import get_default_user_group, install_sample_user
 import beeswax.conf
 from beeswax.models import SavedQuery, HQL, IMPALA
 from beeswax.design import hql_query
+from beeswax.hive_site import has_concurrency_support
 from beeswax.server import dbms
 from beeswax.server.dbms import get_query_server_config, QueryServerException
 
 
 LOG = logging.getLogger(__name__)
+MAX_INSERTED_ROWS = 1000
 
 
 class InstallException(Exception):
@@ -60,7 +63,7 @@ class Command(BaseCommand):
       db_name = options.get('db_name', 'default')
       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
 
@@ -76,7 +79,7 @@ class Command(BaseCommand):
 
     if exception is not None:
       pretty_msg = None
-      
+
       if "AlreadyExistsException" in exception.message:
         pretty_msg = _("SQL table examples already installed.")
       if "Permission denied" in exception.message:
@@ -84,7 +87,7 @@ class Command(BaseCommand):
 
       if pretty_msg is not None:
         raise PopupException(pretty_msg)
-      else: 
+      else:
         raise exception
 
   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}
     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):
     """
     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)
-    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):
@@ -270,7 +290,7 @@ class SampleTable(object):
     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,))
     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()