create_table.py 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478
  1. #!/usr/bin/env python
  2. # Licensed to Cloudera, Inc. under one
  3. # or more contributor license agreements. See the NOTICE file
  4. # distributed with this work for additional information
  5. # regarding copyright ownership. Cloudera, Inc. licenses this file
  6. # to you under the Apache License, Version 2.0 (the
  7. # "License"); you may not use this file except in compliance
  8. # with the License. You may obtain a copy of the License at
  9. #
  10. # http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. # Unless required by applicable law or agreed to in writing, software
  13. # distributed under the License is distributed on an "AS IS" BASIS,
  14. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. # See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. import csv
  18. import gzip
  19. import json
  20. import logging
  21. from django.core.urlresolvers import reverse
  22. from django.utils.translation import ugettext as _
  23. from desktop.context_processors import get_app_name
  24. from desktop.lib import django_mako, i18n
  25. from desktop.lib.django_util import render
  26. from desktop.lib.exceptions_renderable import PopupException
  27. from desktop.lib.django_forms import MultiForm
  28. from hadoop.fs import hadoopfs
  29. from hadoop.fs.fsutils import remove_header
  30. from beeswax.common import TERMINATORS
  31. from beeswax.design import hql_query
  32. from beeswax.forms import CreateTableForm, ColumnTypeFormSet,\
  33. PartitionTypeFormSet, CreateByImportFileForm, CreateByImportDelimForm,\
  34. TERMINATOR_CHOICES
  35. from beeswax.server import dbms
  36. from beeswax.views import execute_directly
  37. import re
  38. LOG = logging.getLogger(__name__)
  39. def create_table(request, database='default'):
  40. """Create a table by specifying its attributes manually"""
  41. db = dbms.get(request.user)
  42. dbs = db.get_databases()
  43. databases = [{'name':db, 'url':reverse('beeswax:create_table', kwargs={'database': db})} for db in dbs]
  44. form = MultiForm(
  45. table=CreateTableForm,
  46. columns=ColumnTypeFormSet,
  47. partitions=PartitionTypeFormSet
  48. )
  49. if request.method == "POST":
  50. form.bind(request.POST)
  51. form.table.db = db # curry is invalid
  52. form.table.database = database
  53. if request.POST.get('create'):
  54. if form.is_valid():
  55. columns = [ f.cleaned_data for f in form.columns.forms ]
  56. partition_columns = [ f.cleaned_data for f in form.partitions.forms ]
  57. proposed_query = django_mako.render_to_string("create_table_statement.mako", {
  58. 'databases': databases,
  59. 'database': database,
  60. 'table': form.table.cleaned_data,
  61. 'columns': columns,
  62. 'partition_columns': partition_columns
  63. }
  64. )
  65. # Mako outputs bytestring in utf8
  66. proposed_query = proposed_query.decode('utf-8')
  67. table_name = form.table.cleaned_data['name']
  68. return _submit_create_and_load(request, proposed_query, table_name, None, False, database=database)
  69. else:
  70. form.bind()
  71. return render("create_table_manually.mako", request, {
  72. 'action': "#",
  73. 'databases': databases,
  74. 'table_form': form.table,
  75. 'columns_form': form.columns,
  76. 'partitions_form': form.partitions,
  77. 'has_tables': len(dbms.get(request.user).get_tables()) > 0,
  78. 'database': database,
  79. })
  80. IMPORT_PEEK_SIZE = 5 * 1024**2
  81. IMPORT_PEEK_NLINES = 10
  82. DELIMITERS = [ hive_val for hive_val, desc, ascii in TERMINATORS ]
  83. DELIMITER_READABLE = {'\\001' : _('ctrl-As'),
  84. '\\002' : _('ctrl-Bs'),
  85. '\\003' : _('ctrl-Cs'),
  86. '\\t' : _('tabs'),
  87. ',' : _('commas'),
  88. ' ' : _('spaces')}
  89. FILE_READERS = [ ]
  90. def import_wizard(request, database='default'):
  91. """
  92. Help users define table and based on a file they want to import to Hive.
  93. Limitations:
  94. - Rows are delimited (no serde).
  95. - No detection for map and array types.
  96. - No detection for the presence of column header in the first row.
  97. - No partition table.
  98. - Does not work with binary data.
  99. """
  100. encoding = i18n.get_site_encoding()
  101. app_name = get_app_name(request)
  102. db = dbms.get(request.user)
  103. dbs = db.get_databases()
  104. databases = [{'name':db, 'url':reverse('beeswax:import_wizard', kwargs={'database': db})} for db in dbs]
  105. if request.method == 'POST':
  106. #
  107. # General processing logic:
  108. # - We have 3 steps. Each requires the previous.
  109. # * Step 1 : Table name and file location
  110. # * Step 2a : Display sample with auto chosen delim
  111. # * Step 2b : Display sample with user chosen delim (if user chooses one)
  112. # * Step 3 : Display sample, and define columns
  113. # - Each step is represented by a different form. The form of an earlier step
  114. # should be present when submitting to a later step.
  115. # - To preserve the data from the earlier steps, we send the forms back as
  116. # hidden fields. This way, when users revisit a previous step, the data would
  117. # be there as well.
  118. #
  119. delim_is_auto = False
  120. fields_list, n_cols = [[]], 0
  121. s3_col_formset = None
  122. s1_file_form = CreateByImportFileForm(request.POST, db=db)
  123. if s1_file_form.is_valid():
  124. do_s2_auto_delim = request.POST.get('submit_file') # Step 1 -> 2
  125. do_s2_user_delim = request.POST.get('submit_preview') # Step 2 -> 2
  126. do_s3_column_def = request.POST.get('submit_delim') # Step 2 -> 3
  127. do_hive_create = request.POST.get('submit_create') # Step 3 -> execute
  128. cancel_s2_user_delim = request.POST.get('cancel_delim') # Step 2 -> 1
  129. cancel_s3_column_def = request.POST.get('cancel_create') # Step 3 -> 2
  130. # Exactly one of these should be True
  131. if len(filter(None, (do_s2_auto_delim, do_s2_user_delim, do_s3_column_def, do_hive_create, cancel_s2_user_delim, cancel_s3_column_def))) != 1:
  132. raise PopupException(_('Invalid form submission'))
  133. if not do_s2_auto_delim:
  134. # We should have a valid delim form
  135. s2_delim_form = CreateByImportDelimForm(request.POST)
  136. if not s2_delim_form.is_valid():
  137. # Go back to picking delimiter
  138. do_s2_user_delim, do_s3_column_def, do_hive_create = True, False, False
  139. if do_hive_create:
  140. # We should have a valid columns formset
  141. s3_col_formset = ColumnTypeFormSet(prefix='cols', data=request.POST)
  142. if not s3_col_formset.is_valid():
  143. # Go back to define columns
  144. do_s3_column_def, do_hive_create = True, False
  145. #
  146. # Go to step 2: We've just picked the file. Preview it.
  147. #
  148. if do_s2_auto_delim:
  149. delim_is_auto = True
  150. fields_list, n_cols, s2_delim_form = _delim_preview(request.fs, s1_file_form, encoding, [reader.TYPE for reader in FILE_READERS], DELIMITERS)
  151. if (do_s2_user_delim or do_s3_column_def or cancel_s3_column_def) and s2_delim_form.is_valid():
  152. # Delimit based on input
  153. fields_list, n_cols, s2_delim_form = _delim_preview(request.fs, s1_file_form, encoding, (s2_delim_form.cleaned_data['file_type'],),
  154. (s2_delim_form.cleaned_data['delimiter'],))
  155. if do_s2_auto_delim or do_s2_user_delim or cancel_s3_column_def:
  156. return render('choose_delimiter.mako', request, {
  157. 'action': reverse(app_name + ':import_wizard', kwargs={'database': database}),
  158. 'delim_readable': DELIMITER_READABLE.get(s2_delim_form['delimiter'].data[0], s2_delim_form['delimiter'].data[1]),
  159. 'initial': delim_is_auto,
  160. 'file_form': s1_file_form,
  161. 'delim_form': s2_delim_form,
  162. 'fields_list': fields_list,
  163. 'delimiter_choices': TERMINATOR_CHOICES,
  164. 'n_cols': n_cols,
  165. 'database': database,
  166. 'databases': databases
  167. })
  168. #
  169. # Go to step 3: Define column.
  170. #
  171. if do_s3_column_def:
  172. if s3_col_formset is None:
  173. columns = []
  174. for i in range(n_cols):
  175. columns.append({
  176. 'column_name': 'col_%s' % (i,),
  177. 'column_type': 'string',
  178. })
  179. s3_col_formset = ColumnTypeFormSet(prefix='cols', initial=columns)
  180. try:
  181. fields_list_for_json = list(fields_list)
  182. if fields_list_for_json:
  183. fields_list_for_json[0] = map(lambda a: re.sub('[^\w]', '', a), fields_list_for_json[0]) # Cleaning headers
  184. return render('define_columns.mako', request, {
  185. 'action': reverse(app_name + ':import_wizard', kwargs={'database': database}),
  186. 'file_form': s1_file_form,
  187. 'delim_form': s2_delim_form,
  188. 'column_formset': s3_col_formset,
  189. 'fields_list': fields_list,
  190. 'fields_list_json': json.dumps(fields_list_for_json),
  191. 'n_cols': n_cols,
  192. 'database': database,
  193. 'databases': databases
  194. })
  195. except Exception, e:
  196. raise PopupException(_("The selected delimiter is creating an un-even number of columns. Please make sure you don't have empty columns."), detail=e)
  197. #
  198. # Final: Execute
  199. #
  200. if do_hive_create:
  201. delim = s2_delim_form.cleaned_data['delimiter']
  202. table_name = s1_file_form.cleaned_data['name']
  203. proposed_query = django_mako.render_to_string("create_table_statement.mako", {
  204. 'table': {
  205. 'name': table_name,
  206. 'comment': s1_file_form.cleaned_data['comment'],
  207. 'row_format': 'Delimited',
  208. 'field_terminator': delim
  209. },
  210. 'columns': [ f.cleaned_data for f in s3_col_formset.forms ],
  211. 'partition_columns': [],
  212. 'database': database,
  213. 'databases': databases
  214. }
  215. )
  216. do_load_data = s1_file_form.cleaned_data.get('do_import')
  217. path = s1_file_form.cleaned_data['path']
  218. return _submit_create_and_load(request, proposed_query, table_name, path, do_load_data, database=database)
  219. else:
  220. s1_file_form = CreateByImportFileForm()
  221. return render('choose_file.mako', request, {
  222. 'action': reverse(app_name + ':import_wizard', kwargs={'database': database}),
  223. 'file_form': s1_file_form,
  224. 'database': database,
  225. 'databases': databases
  226. })
  227. def _submit_create_and_load(request, create_hql, table_name, path, do_load, database):
  228. """
  229. Submit the table creation, and setup the load to happen (if ``do_load``).
  230. """
  231. on_success_params = {}
  232. app_name = get_app_name(request)
  233. if do_load:
  234. on_success_params['table'] = table_name
  235. on_success_params['path'] = path
  236. on_success_params['removeHeader'] = request.POST.get('removeHeader')
  237. on_success_url = reverse(app_name + ':load_after_create', kwargs={'database': database})
  238. else:
  239. on_success_url = reverse('metastore:describe_table', kwargs={'database': database, 'table': table_name})
  240. query = hql_query(create_hql, database=database)
  241. return execute_directly(request, query,
  242. on_success_url=on_success_url,
  243. on_success_params=on_success_params)
  244. def _delim_preview(fs, file_form, encoding, file_types, delimiters):
  245. """
  246. _delim_preview(fs, file_form, encoding, file_types, delimiters)
  247. -> (fields_list, n_cols, delim_form)
  248. Look at the beginning of the file and parse it according to the list of
  249. available file_types and delimiters.
  250. """
  251. assert file_form.is_valid()
  252. path = file_form.cleaned_data['path']
  253. try:
  254. file_obj = fs.open(path)
  255. delim, file_type, fields_list = _parse_fields(path, file_obj, encoding, file_types, delimiters)
  256. file_obj.close()
  257. except IOError, ex:
  258. msg = "Failed to open file '%s': %s" % (path, ex)
  259. LOG.exception(msg)
  260. raise PopupException(msg)
  261. n_cols = max([ len(row) for row in fields_list ])
  262. # ``delimiter`` is a MultiValueField. delimiter_0 and delimiter_1 are the sub-fields.
  263. delimiter_0 = delim
  264. delimiter_1 = ''
  265. # If custom delimiter
  266. if not filter(lambda val: val[0] == delim, TERMINATOR_CHOICES):
  267. delimiter_0 = '__other__'
  268. delimiter_1 = delim
  269. delim_form = CreateByImportDelimForm(dict(delimiter_0=delimiter_0,
  270. delimiter_1=delimiter_1,
  271. file_type=file_type,
  272. n_cols=n_cols))
  273. if not delim_form.is_valid():
  274. assert False, _('Internal error when constructing the delimiter form: %(error)s.') % {'error': delim_form.errors}
  275. return fields_list, n_cols, delim_form
  276. def _parse_fields(path, file_obj, encoding, filetypes, delimiters):
  277. """
  278. _parse_fields(path, file_obj, encoding, filetypes, delimiters)
  279. -> (delimiter, filetype, fields_list)
  280. Go through the list of ``filetypes`` (gzip, text) and stop at the first one
  281. that works for the data. Then apply the list of ``delimiters`` and pick the
  282. most appropriate one.
  283. ``path`` is used for debugging only.
  284. Return the best delimiter, filetype and the data broken down into rows of fields.
  285. """
  286. file_readers = [ reader for reader in FILE_READERS if reader.TYPE in filetypes ]
  287. for reader in file_readers:
  288. LOG.debug("Trying %s for file: %s" % (reader.TYPE, path))
  289. file_obj.seek(0, hadoopfs.SEEK_SET)
  290. lines = reader.readlines(file_obj, encoding)
  291. if lines is not None:
  292. delim, fields_list = _readfields(lines, delimiters)
  293. return delim, reader.TYPE, fields_list
  294. else:
  295. # Even TextFileReader doesn't work
  296. msg = _("Failed to decode file '%(path)s' into printable characters under %(encoding)s.") % {'path': path, 'encoding': encoding}
  297. LOG.error(msg)
  298. raise PopupException(msg)
  299. def _readfields(lines, delimiters):
  300. """
  301. readfields(lines, delimiters) -> (delim, a list of lists of fields)
  302. ``delimiters`` is a list of escaped characters, e.g. r'\\t', r'\\001', ','
  303. Choose the best delimiter from the given list of delimiters. Return that delimiter
  304. and the fields parsed by using that delimiter.
  305. """
  306. def score_delim(fields_list):
  307. """
  308. How good is this fields_list? Score based on variance of the number of fields
  309. The score is always non-negative. The higher the better.
  310. """
  311. n_lines = len(fields_list)
  312. len_list = [ len(fields) for fields in fields_list ]
  313. if not len_list:
  314. raise PopupException(_("Could not find any columns to import"))
  315. # All lines should break into multiple fields
  316. if min(len_list) == 1:
  317. return 0
  318. avg_n_fields = sum(len_list) / n_lines
  319. sq_of_exp = avg_n_fields * avg_n_fields
  320. len_list_sq = [ l * l for l in len_list ]
  321. exp_of_sq = sum(len_list_sq) / n_lines
  322. var = exp_of_sq - sq_of_exp
  323. # Favour more fields
  324. return (1000.0 / (var + 1)) + avg_n_fields
  325. max_score = -1
  326. res = (None, None)
  327. for delim in delimiters:
  328. # Unescape the delimiter back to its character value
  329. delimiter = delim.decode('string_escape')
  330. try:
  331. fields_list = _get_rows(lines, delimiter)
  332. except:
  333. fields_list = [line.split(delimiter) for line in lines if line]
  334. score = score_delim(fields_list)
  335. LOG.debug("'%s' gives score of %s" % (delim, score))
  336. if score > max_score:
  337. max_score = score
  338. res = (delim, fields_list)
  339. return res
  340. def _get_rows(lines, delimiter):
  341. column_reader = csv.reader(lines, delimiter=delimiter)
  342. return [row for row in column_reader if row]
  343. def _peek_file(fs, file_form):
  344. """_peek_file(fs, file_form) -> (path, initial data)"""
  345. try:
  346. path = file_form.cleaned_data['path']
  347. file_obj = fs.open(path)
  348. file_head = file_obj.read(IMPORT_PEEK_SIZE)
  349. file_obj.close()
  350. return (path, file_head)
  351. except IOError, ex:
  352. msg = _("Failed to open file '%(path)s': %(error)s.") % {'path': path, 'error': ex}
  353. LOG.exception(msg)
  354. raise PopupException(msg)
  355. class GzipFileReader(object):
  356. """Class for extracting lines from a gzipped file"""
  357. TYPE = 'gzip'
  358. @staticmethod
  359. def readlines(fileobj, encoding):
  360. """readlines(fileobj, encoding) -> list of lines"""
  361. gz = gzip.GzipFile(fileobj=fileobj, mode='rb')
  362. try:
  363. data = gz.read(IMPORT_PEEK_SIZE)
  364. except IOError:
  365. return None
  366. try:
  367. return unicode(data, encoding, errors='replace').split('\n')[:IMPORT_PEEK_NLINES]
  368. except UnicodeError:
  369. return None
  370. FILE_READERS.append(GzipFileReader)
  371. class TextFileReader(object):
  372. """Class for extracting lines from a regular text file"""
  373. TYPE = 'text'
  374. @staticmethod
  375. def readlines(fileobj, encoding):
  376. """readlines(fileobj, encoding) -> list of lines"""
  377. try:
  378. data = fileobj.read(IMPORT_PEEK_SIZE)
  379. return unicode(data, encoding, errors='replace').split('\n')[:IMPORT_PEEK_NLINES]
  380. except UnicodeError:
  381. return None
  382. FILE_READERS.append(TextFileReader)
  383. def load_after_create(request, database):
  384. """
  385. Automatically load data into a newly created table.
  386. We get here from the create's on_success_url, and expect to find
  387. ``table`` and ``path`` from the parameters.
  388. """
  389. tablename = request.REQUEST.get('table')
  390. path = request.REQUEST.get('path')
  391. is_remove_header = request.REQUEST.get('removeHeader').lower() == 'on' and not path.endswith('gz')
  392. if not tablename or not path:
  393. msg = _('Internal error: Missing needed parameter to load data into table.')
  394. LOG.error(msg)
  395. raise PopupException(msg)
  396. if is_remove_header:
  397. remove_header(request.fs, path)
  398. LOG.debug("Auto loading data from %s into table %s" % (path, tablename))
  399. hql = "LOAD DATA INPATH '%s' INTO TABLE `%s.%s`" % (path, database, tablename)
  400. query = hql_query(hql)
  401. on_success_url = reverse('metastore:describe_table', kwargs={'database': database, 'table': tablename})
  402. return execute_directly(request, query, on_success_url=on_success_url)