Bladeren bron

[importer_direct_upload_impala] supporting impala dialect for direct upload

ayush.goyal 4 jaren geleden
bovenliggende
commit
81f5e03e88
1 gewijzigde bestanden met toevoegingen van 41 en 0 verwijderingen
  1. 41 0
      desktop/libs/indexer/src/indexer/indexers/sql.py

+ 41 - 0
desktop/libs/indexer/src/indexer/indexers/sql.py

@@ -329,6 +329,15 @@ CONSTRAINT my_pk PRIMARY KEY (%(primary_keys)s));
           'primary_keys': ', '.join(destination.get('primaryKeys'))
       }
 
+    elif dialect == 'impala':
+      sql = '''CREATE TABLE IF NOT EXISTS %(database)s.%(table_name)s_tmp (
+%(columns)s);
+      ''' % {
+                'database': database,
+                'table_name': table_name,
+                'columns': ',\n'.join(['  `%(name)s` string' % col for col in columns]),
+            }                        #Impala does not implicitly cast between string and numeric or Boolean types.
+
     path = urllib_unquote(source['path'])
 
     if path:                                                     # data insertion
@@ -359,6 +368,38 @@ CONSTRAINT my_pk PRIMARY KEY (%(primary_keys)s));
                     'table_name': table_name,
                     'csv_row': _sql
                   }
+        elif dialect == 'impala':
+          sql += '''\nINSERT INTO %(database)s.%(table_name)s_tmp VALUES %(csv_rows)s;
+          '''% {
+                  'database': database,
+                  'table_name': table_name,
+                  'csv_rows': csv_rows
+                }
+          sql += '''\nCREATE TABLE IF NOT EXISTS %(database)s.%(table_name)s
+AS SELECT'''% {
+                'database': database,
+                'table_name': table_name,
+              }
+          for count, col in enumerate(columns):
+            if col['type'] == 'boolean':                              # casting from string to boolean is not allowed in impala
+              col['type'] = 'string'
+            sql += '''\n  CAST ( `%(col_name)s` AS %(col_type)s ) `%(col_name)s`'''%{
+              'col_name': col['name'],
+              'col_type': col['type']
+            }
+            if count != len(columns)-1:
+              sql += ','
+          
+          sql += '''\nFROM  %(database)s.%(table_name)s_tmp;
+          '''% {
+                  'database': database,
+                  'table_name': table_name,
+                }
+          sql += '''\nDROP TABLE IF EXISTS %(database)s.%(table_name)s_tmp;
+          '''% {
+                  'database': database,
+                  'table_name': table_name,
+                }
 
     on_success_url = reverse('metastore:describe_table', kwargs={'database': database, 'table': final_table_name}) + \
         '?source_type=' + source_type