beeswax_install_examples.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353
  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 logging
  18. import os
  19. import pwd
  20. import json
  21. from django.core.management.base import BaseCommand
  22. from django.contrib.auth.models import User
  23. from django.utils.translation import ugettext as _
  24. from desktop.lib.exceptions_renderable import PopupException
  25. from desktop.conf import USE_NEW_EDITOR
  26. from desktop.models import Directory, Document, Document2, Document2Permission
  27. from hadoop import cluster
  28. from notebook.models import import_saved_beeswax_query
  29. from useradmin.models import get_default_user_group, install_sample_user
  30. import beeswax.conf
  31. from beeswax.models import SavedQuery, HQL, IMPALA
  32. from beeswax.design import hql_query
  33. from beeswax.server import dbms
  34. from beeswax.server.dbms import get_query_server_config, QueryServerException
  35. LOG = logging.getLogger(__name__)
  36. class InstallException(Exception):
  37. pass
  38. class Command(BaseCommand):
  39. args = '<beeswax|impala>'
  40. help = 'Install examples but do not overwrite them.'
  41. def handle(self, *args, **options):
  42. if args:
  43. app_name = args[0]
  44. user = User.objects.get(username=pwd.getpwuid(os.getuid()).pw_name)
  45. else:
  46. app_name = options['app_name']
  47. user = options['user']
  48. tables = options['tables'] if 'tables' in options else 'tables.json'
  49. exception = None
  50. # Documents will belong to this user but we run the install as the current user
  51. try:
  52. sample_user = install_sample_user()
  53. self._install_queries(sample_user, app_name)
  54. self._install_tables(user, app_name, tables)
  55. except Exception, ex:
  56. exception = ex
  57. Document.objects.sync()
  58. if exception is not None:
  59. pretty_msg = None
  60. if "AlreadyExistsException" in exception.message:
  61. pretty_msg = _("SQL table examples already installed.")
  62. if "Permission denied" in exception.message:
  63. pretty_msg = _("Permission denied. Please check with your system administrator.")
  64. if pretty_msg is not None:
  65. raise PopupException(pretty_msg)
  66. else:
  67. raise exception
  68. def _install_tables(self, django_user, app_name, tables):
  69. data_dir = beeswax.conf.LOCAL_EXAMPLES_DATA_DIR.get()
  70. table_file = file(os.path.join(data_dir, tables))
  71. table_list = json.load(table_file)
  72. table_file.close()
  73. for table_dict in table_list:
  74. table = SampleTable(table_dict, app_name)
  75. try:
  76. table.install(django_user)
  77. except Exception, ex:
  78. raise InstallException(_('Could not install table: %s') % ex)
  79. def _install_queries(self, django_user, app_name):
  80. design_file = file(os.path.join(beeswax.conf.LOCAL_EXAMPLES_DATA_DIR.get(), 'designs.json'))
  81. design_list = json.load(design_file)
  82. design_file.close()
  83. # Filter design list to app-specific designs
  84. app_type = HQL if app_name == 'beeswax' else IMPALA
  85. design_list = filter(lambda d: int(d['type']) == app_type, design_list)
  86. for design_dict in design_list:
  87. design = SampleQuery(design_dict)
  88. try:
  89. design.install(django_user)
  90. except Exception, ex:
  91. raise InstallException(_('Could not install query: %s') % ex)
  92. class SampleTable(object):
  93. """
  94. Represents a table loaded from the tables.json file
  95. """
  96. def __init__(self, data_dict, app_name):
  97. self.name = data_dict['table_name']
  98. if 'partition_files' in data_dict:
  99. self.partition_files = data_dict['partition_files']
  100. else:
  101. self.partition_files = None
  102. self.filename = data_dict['data_file']
  103. self.hql = data_dict['create_hql']
  104. self.query_server = get_query_server_config(app_name)
  105. self.app_name = app_name
  106. # Sanity check
  107. self._data_dir = beeswax.conf.LOCAL_EXAMPLES_DATA_DIR.get()
  108. if self.partition_files:
  109. for partition_spec, filename in self.partition_files.items():
  110. filepath = os.path.join(self._data_dir, filename)
  111. self.partition_files[partition_spec] = filepath
  112. self._check_file_contents(filepath)
  113. else:
  114. self._contents_file = os.path.join(self._data_dir, self.filename)
  115. self._check_file_contents(self._contents_file)
  116. def install(self, django_user):
  117. if self.create(django_user):
  118. if self.partition_files:
  119. for partition_spec, filepath in self.partition_files.items():
  120. self.load_partition(django_user, partition_spec, filepath)
  121. else:
  122. self.load(django_user)
  123. def create(self, django_user):
  124. """
  125. Create table in the Hive Metastore.
  126. """
  127. LOG.info('Creating table "%s"' % (self.name,))
  128. db = dbms.get(django_user, self.query_server)
  129. try:
  130. # Already exists?
  131. if self.app_name == 'impala':
  132. db.invalidate(database='default', flush_all=False)
  133. db.get_table('default', self.name)
  134. msg = _('Table "%(table)s" already exists.') % {'table': self.name}
  135. LOG.error(msg)
  136. return False
  137. except Exception:
  138. query = hql_query(self.hql)
  139. try:
  140. results = db.execute_and_wait(query)
  141. if not results:
  142. msg = _('Error creating table %(table)s: Operation timeout.') % {'table': self.name}
  143. LOG.error(msg)
  144. raise InstallException(msg)
  145. return True
  146. except Exception, ex:
  147. msg = _('Error creating table %(table)s: %(error)s.') % {'table': self.name, 'error': ex}
  148. LOG.error(msg)
  149. raise InstallException(msg)
  150. def load_partition(self, django_user, partition_spec, filepath):
  151. """
  152. Upload data found at filepath to HDFS home of user, the load intto a specific partition
  153. """
  154. LOAD_PARTITION_HQL = \
  155. """
  156. ALTER TABLE %(tablename)s ADD PARTITION(%(partition_spec)s) LOCATION '%(filepath)s'
  157. """
  158. partition_dir = self._get_partition_dir(partition_spec)
  159. hdfs_root_destination = self._get_hdfs_root_destination(django_user, subdir=partition_dir)
  160. filename = filepath.split('/')[-1]
  161. hdfs_file_destination = self._upload_to_hdfs(django_user, filepath, hdfs_root_destination, filename)
  162. hql = LOAD_PARTITION_HQL % {'tablename': self.name, 'partition_spec': partition_spec, 'filepath': hdfs_root_destination}
  163. LOG.info('Running load query: %s' % hql)
  164. self._load_data_to_table(django_user, hql, hdfs_file_destination)
  165. def load(self, django_user):
  166. """
  167. Upload data to HDFS home of user then load (aka move) it into the Hive table (in the Hive metastore in HDFS).
  168. """
  169. LOAD_HQL = \
  170. """
  171. LOAD DATA INPATH
  172. '%(filename)s' OVERWRITE INTO TABLE %(tablename)s
  173. """
  174. hdfs_root_destination = self._get_hdfs_root_destination(django_user)
  175. hdfs_file_destination = self._upload_to_hdfs(django_user, self._contents_file, hdfs_root_destination)
  176. hql = LOAD_HQL % {'tablename': self.name, 'filename': hdfs_file_destination}
  177. LOG.info('Running load query: %s' % hql)
  178. self._load_data_to_table(django_user, hql, hdfs_file_destination)
  179. def _check_file_contents(self, filepath):
  180. if not os.path.isfile(filepath):
  181. msg = _('Cannot find table data in "%(file)s".') % {'file': filepath}
  182. LOG.error(msg)
  183. raise ValueError(msg)
  184. def _get_partition_dir(self, partition_spec):
  185. parts = partition_spec.split(',')
  186. last_part = parts[-1]
  187. part_value = last_part.split('=')[-1]
  188. part_dir = part_value.strip("'").replace('-', '_')
  189. return part_dir
  190. def _get_hdfs_root_destination(self, django_user, subdir=None):
  191. fs = cluster.get_hdfs()
  192. if self.app_name == 'impala':
  193. # Because Impala does not have impersonation on by default, we use a public destination for the upload.
  194. from impala.conf import IMPERSONATION_ENABLED
  195. if not IMPERSONATION_ENABLED.get():
  196. tmp_public = '/tmp/public_hue_examples'
  197. if subdir:
  198. tmp_public += '/%s' % subdir
  199. fs.do_as_user(django_user, fs.mkdir, tmp_public, '0777')
  200. hdfs_root_destination = tmp_public
  201. else:
  202. hdfs_root_destination = fs.do_as_user(django_user, fs.get_home_dir)
  203. if subdir:
  204. hdfs_root_destination += '/%s' % subdir
  205. fs.do_as_user(django_user, fs.mkdir, hdfs_root_destination, '0777')
  206. return hdfs_root_destination
  207. def _upload_to_hdfs(self, django_user, local_filepath, hdfs_root_destination, filename=None):
  208. fs = cluster.get_hdfs()
  209. if filename is None:
  210. filename = self.name
  211. hdfs_destination = '%s/%s' % (hdfs_root_destination, filename)
  212. LOG.info('Uploading local data %s to HDFS path "%s"' % (self.name, hdfs_destination))
  213. fs.do_as_user(django_user, fs.copyFromLocal, local_filepath, hdfs_destination)
  214. return hdfs_destination
  215. def _load_data_to_table(self, django_user, hql, hdfs_destination):
  216. LOG.info('Loading data into table "%s"' % (self.name,))
  217. query = hql_query(hql)
  218. try:
  219. results = dbms.get(django_user, self.query_server).execute_and_wait(query)
  220. if not results:
  221. msg = _('Error loading table %(table)s: Operation timeout.') % {'table': self.name}
  222. LOG.error(msg)
  223. raise InstallException(msg)
  224. except QueryServerException, ex:
  225. msg = _('Error loading table %(table)s: %(error)s.') % {'table': self.name, 'error': ex}
  226. LOG.error(msg)
  227. raise InstallException(msg)
  228. class SampleQuery(object):
  229. """Represents a query loaded from the designs.json file"""
  230. def __init__(self, data_dict):
  231. self.name = data_dict['name']
  232. self.desc = data_dict['desc']
  233. self.type = int(data_dict['type'])
  234. self.data = data_dict['data']
  235. def install(self, django_user):
  236. """
  237. Install queries. Raise InstallException on failure.
  238. """
  239. LOG.info('Installing sample query: %s' % (self.name,))
  240. try:
  241. # Don't overwrite
  242. query = SavedQuery.objects.get(owner=django_user, name=self.name, type=self.type)
  243. except SavedQuery.DoesNotExist:
  244. query = SavedQuery(owner=django_user, name=self.name, type=self.type, desc=self.desc)
  245. # The data field needs to be a string. The sample file writes it
  246. # as json (without encoding into a string) for readability.
  247. query.data = json.dumps(self.data)
  248. query.save()
  249. LOG.info('Successfully installed sample design: %s' % (self.name,))
  250. if USE_NEW_EDITOR.get():
  251. try:
  252. # Don't overwrite
  253. doc2 = Document2.objects.get(owner=django_user, name=self.name, type=self._document_type(self.type))
  254. except Document2.DoesNotExist:
  255. # Create document from saved query
  256. notebook = import_saved_beeswax_query(query)
  257. data = notebook.get_data()
  258. data['isSaved'] = True
  259. uuid = data.get('uuid')
  260. data = json.dumps(data)
  261. # Get or create sample user directories
  262. home_dir = Directory.objects.get_home_directory(django_user)
  263. examples_dir, created = Directory.objects.get_or_create(
  264. parent_directory=home_dir,
  265. owner=django_user,
  266. name=Document2.EXAMPLES_DIR
  267. )
  268. doc2 = Document2.objects.create(
  269. uuid=uuid,
  270. owner=django_user,
  271. parent_directory=examples_dir,
  272. name=self.name,
  273. type=self._document_type(self.type),
  274. description=self.desc,
  275. data=data
  276. )
  277. # Share with default group
  278. examples_dir.share(django_user, Document2Permission.READ_PERM, groups=[get_default_user_group()])
  279. doc2.save()
  280. LOG.info('Successfully installed sample query: %s' % (self.name,))
  281. def _document_type(self, type):
  282. if type == HQL:
  283. return 'query-hive'
  284. elif type == IMPALA:
  285. return 'query-impala'
  286. else:
  287. return None