Эх сурвалжийг харах

HUE-8948 [hive] Fix INSERT data into customer table which is partitioned

Romain 6 жил өмнө
parent
commit
d4bac9d0dd

+ 1 - 1
apps/beeswax/data/tables_transactional.json

@@ -18,6 +18,6 @@
     },
     "create_hql": "CREATE TABLE `web_logs`  (  `_version_` bigint,   `app` string,   `bytes` int,   `city` string,   `client_ip` string,   `code` smallint,   `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` string,   `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",
-    "column_types": ["bigint", "string", "int", "string", "string", "smallint", "string", "string", "string", "string", "string", "float", "float", "string", "string", "string", "string", "string", "string", "string", "string", "string", "string", "string", "string", "string", "string", "string", "string"]
+    "columns": [{"name": "_version_", "type": "bigint"}, {"name": "app", "type": "string"}, {"name": "bytes", "type": "int"}, {"name": "city", "type": "string"}, {"name": "client_ip", "type": "string"}, {"name": "code", "type": "smallint"}, {"name": "country_code", "type": "string"}, {"name": "country_code3", "type": "string"}, {"name": "country_name", "type": "string"}, {"name": "device_family", "type": "string"}, {"name": "extension", "type": "string"}, {"name": "latitude", "type": "float"}, {"name": "longitude", "type": "float"}, {"name": "method", "type": "string"}, {"name": "os_family", "type": "string"}, {"name": "os_major", "type": "string"}, {"name": "protocol", "type": "string"}, {"name": "record", "type": "string"}, {"name": "referer", "type": "string"}, {"name": "region_code", "type": "string"}, {"name": "request", "type": "string"}, {"name": "subapp", "type": "string"}, {"name": "time", "type": "string"}, {"name": "url", "type": "string"}, {"name": "user_agent", "type": "string"}, {"name": "user_agent_family", "type": "string"}, {"name": "user_agent_major", "type": "string"}, {"name": "id", "type": "string"}, {"name": "date", "type": "string"}]
   }
 ]

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

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

+ 14 - 6
apps/beeswax/src/beeswax/management/commands/beeswax_install_examples.py

@@ -135,6 +135,7 @@ class SampleTable(object):
     self.query_server = get_query_server_config(app_name)
     self.app_name = app_name
     self.db_name = db_name
+    self.columns = data_dict.get('columns')
 
     # Sanity check
     self._data_dir = beeswax.conf.LOCAL_EXAMPLES_DATA_DIR.get()
@@ -152,10 +153,11 @@ class SampleTable(object):
     if self.create(django_user):
       if self.partition_files:
         for partition_spec, filepath in list(self.partition_files.items()):
-          self.load_partition(django_user, partition_spec, filepath)
+          self.load_partition(django_user, partition_spec, filepath, columns=self.columns)
       else:
         self.load(django_user)
 
+
   def create(self, django_user):
     """
     Create table in the Hive Metastore.
@@ -185,7 +187,7 @@ class SampleTable(object):
         LOG.error(msg)
         raise InstallException(msg)
 
-  def load_partition(self, django_user, partition_spec, filepath):
+  def load_partition(self, django_user, partition_spec, filepath, columns):
     if has_concurrency_support():
       with open(filepath) as f:
         hql = \
@@ -196,7 +198,7 @@ class SampleTable(object):
           """ % {
             'tablename': self.name,
             'partition_spec': partition_spec,
-            'values': self._get_sql_insert_values(f)
+            'values': self._get_sql_insert_values(f, columns)
           }
     else:
       # Upload data found at filepath to HDFS home of user, the load intto a specific partition
@@ -308,19 +310,25 @@ class SampleTable(object):
       raise InstallException(msg)
 
 
-  def _get_sql_insert_values(self, f):
+  def _get_sql_insert_values(self, f, columns=None):
     data = f.read()
     dialect = csv.Sniffer().sniff(data)
     reader = csv.reader(data.splitlines(), delimiter=dialect.delimiter)
 
-    rows = [', '.join(
-        col if col.replace('.', '' , 1).isdigit() or col == 'NULL' else 'NULL' if col == '' else "'%s'" % col.replace("'", "\\'") for col in row
+    rows = [
+      ', '.join(
+        col if is_number(col, i, columns) else "'%s'" % col.replace("'", "\\'") for i, col in enumerate(row)
       ) for row in reader
     ]
 
     return ', '.join('(%s)' % row for row in rows)
 
 
+def is_number(col, i, columns):
+  '''Basic check. For proper check, use columns headers like for the web_logs table.'''
+  return columns[i]['type'] != 'string' if columns else col.isdigit() or col == 'NULL'
+
+
 class SampleQuery(object):
 
   """Represents a query loaded from the designs.json file"""

+ 15 - 7
apps/beeswax/src/beeswax/management/commands/beeswax_install_examples_tests.py

@@ -53,20 +53,28 @@ class TestTransactionalTables():
       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)
+        SampleTable(table_data, 'beeswax', 'default').install(self.user)
 
         get.assert_called()
 
 
-  def test_load_tables_concurrency_support(self):
+  def test_load_web_logs_with_concurrency_support(self):
+    table_data = {
+      "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` int,   `city` string,   `client_ip` string,   `code` smallint,   `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` string,   `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",
+      "columns": [{"name": "_version_", "type": "bigint"}, {"name": "app", "type": "string"}, {"name": "bytes", "type": "int"}, {"name": "city", "type": "string"}, {"name": "client_ip", "type": "string"}, {"name": "code", "type": "smallint"}, {"name": "country_code", "type": "string"}, {"name": "country_code3", "type": "string"}, {"name": "country_name", "type": "string"}, {"name": "device_family", "type": "string"}, {"name": "extension", "type": "string"}, {"name": "latitude", "type": "float"}, {"name": "longitude", "type": "float"}, {"name": "method", "type": "string"}, {"name": "os_family", "type": "string"}, {"name": "os_major", "type": "string"}, {"name": "protocol", "type": "string"}, {"name": "record", "type": "string"}, {"name": "referer", "type": "string"}, {"name": "region_code", "type": "string"}, {"name": "request", "type": "string"}, {"name": "subapp", "type": "string"}, {"name": "time", "type": "string"}, {"name": "url", "type": "string"}, {"name": "user_agent", "type": "string"}, {"name": "user_agent_family", "type": "string"}, {"name": "user_agent_major", "type": "string"}, {"name": "id", "type": "string"}, {"name": "date", "type": "string"}]
+    }
+
     with patch('beeswax.server.dbms.get') as get:
       with patch('beeswax.management.commands.beeswax_install_examples.has_concurrency_support') as has_concurrency_support:
-        get.return_value = Mock(
-          get_table=Exception('Table could not be found')
-        )
         has_concurrency_support.return_value = True
 
-        cmd = Command()
-        cmd._install_tables(self.user, 'beeswax', 'default', 'tables_transactional.json')
+        SampleTable(table_data, 'beeswax', 'default').install(self.user)
 
         get.assert_called()