base.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388
  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 json
  18. import logging
  19. import re
  20. import uuid
  21. from django.utils.translation import ugettext as _
  22. from desktop.lib.exceptions_renderable import PopupException
  23. from desktop.lib.i18n import smart_unicode
  24. from notebook.conf import get_ordered_interpreters
  25. from desktop.models import Cluster
  26. LOG = logging.getLogger(__name__)
  27. class SessionExpired(Exception):
  28. pass
  29. class QueryExpired(Exception):
  30. def __init__(self, message=None):
  31. super(QueryExpired, self).__init__()
  32. self.message = message
  33. class AuthenticationRequired(Exception):
  34. pass
  35. class OperationTimeout(Exception):
  36. pass
  37. class OperationNotSupported(Exception):
  38. pass
  39. class QueryError(Exception):
  40. def __init__(self, message, handle=None):
  41. self.message = message or _('No error message, please check the logs.')
  42. self.handle = handle
  43. self.extra = {}
  44. def __unicode__(self):
  45. return smart_unicode(self.message)
  46. class Notebook(object):
  47. def __init__(self, document=None, **options):
  48. self.document = None
  49. if document is not None:
  50. self.data = document.data
  51. self.document = document
  52. else:
  53. _data = {
  54. 'name': 'My Notebook',
  55. 'uuid': str(uuid.uuid4()),
  56. 'description': '',
  57. 'type': 'notebook',
  58. 'isSaved': False,
  59. 'isManaged': False, # Aka isTask
  60. 'skipHistorify': False,
  61. 'sessions': [],
  62. 'snippets': [],
  63. }
  64. _data.update(options)
  65. self.data = json.dumps(_data)
  66. def get_json(self):
  67. _data = self.get_data()
  68. return json.dumps(_data)
  69. def get_data(self):
  70. _data = json.loads(self.data)
  71. if self.document is not None:
  72. _data['id'] = self.document.id
  73. _data['is_history'] = self.document.is_history
  74. return _data
  75. def get_str(self):
  76. return '\n\n\n'.join(['USE %s;\n\n%s' % (snippet['database'], snippet['statement_raw']) for snippet in self.get_data()['snippets']])
  77. def add_hive_snippet(self, database, sql):
  78. _data = json.loads(self.data)
  79. _data['snippets'].append(self._make_snippet({
  80. 'status': 'running',
  81. 'statement_raw': sql,
  82. 'statement': sql,
  83. 'type': 'hive',
  84. 'properties': {
  85. 'files': [],
  86. 'functions': [],
  87. 'settings': [],
  88. },
  89. 'database': database,
  90. }))
  91. self._add_session(_data, 'hive')
  92. self.data = json.dumps(_data)
  93. def add_java_snippet(self, clazz, app_jar, arguments, files):
  94. _data = json.loads(self.data)
  95. _data['snippets'].append(self._make_snippet({
  96. u'type': u'java',
  97. u'status': u'running',
  98. u'properties': {
  99. u'files': files,
  100. u'class': clazz,
  101. u'app_jar': app_jar,
  102. u'arguments': arguments,
  103. u'archives': [],
  104. }
  105. }))
  106. self._add_session(_data, 'java')
  107. self.data = json.dumps(_data)
  108. def add_sqoop_snippet(self, statement, arguments, files):
  109. _data = json.loads(self.data)
  110. _data['snippets'].append(self._make_snippet({
  111. u'type': u'sqoop1',
  112. u'status': u'running',
  113. u'properties': {
  114. u'files': files,
  115. u'arguments': arguments,
  116. u'archives': [],
  117. u'statement': statement
  118. }
  119. }))
  120. self._add_session(_data, 'java')
  121. self.data = json.dumps(_data)
  122. def add_spark_snippet(self, clazz, jars, arguments, files):
  123. _data = json.loads(self.data)
  124. _data['snippets'].append(self._make_snippet({
  125. u'type': u'spark',
  126. u'status': u'running',
  127. u'properties': {
  128. u'files': files,
  129. u'class': clazz,
  130. u'app_jar': jars,
  131. u'arguments': arguments,
  132. u'archives': [],
  133. u'spark_opts': ''
  134. }
  135. }))
  136. self._add_session(_data, 'spark')
  137. self.data = json.dumps(_data)
  138. def add_shell_snippet(self, shell_command, arguments=None, archives=None, files=None, env_var=None, last_executed=None):
  139. _data = json.loads(self.data)
  140. if arguments is None:
  141. arguments = []
  142. if archives is None:
  143. archives = []
  144. if files is None:
  145. files = []
  146. if env_var is None:
  147. env_var = []
  148. _data['snippets'].append(self._make_snippet({
  149. u'type': u'shell',
  150. u'status': u'running',
  151. u'properties': {
  152. u'files': files,
  153. u'shell_command': shell_command,
  154. u'arguments': arguments,
  155. u'archives': archives,
  156. u'env_var': env_var,
  157. u'command_path': shell_command
  158. },
  159. u'lastExecuted': last_executed
  160. }))
  161. self._add_session(_data, 'shell')
  162. self.data = json.dumps(_data)
  163. def _make_snippet(self, _snippet):
  164. return {
  165. 'status': _snippet.get('status', 'ready'),
  166. 'id': str(uuid.uuid4()),
  167. 'statement_raw': _snippet.get('statement', ''),
  168. 'statement': _snippet.get('statement', ''),
  169. 'type': _snippet.get('type'),
  170. 'properties': _snippet['properties'],
  171. 'name': _snippet.get('name', '%(type)s snippet' % _snippet),
  172. 'database': _snippet.get('database'),
  173. 'result': {},
  174. 'variables': [],
  175. 'lastExecuted': _snippet.get('lastExecuted')
  176. }
  177. def _add_session(self, data, snippet_type):
  178. from notebook.connectors.hiveserver2 import HS2Api # Cyclic dependency
  179. if snippet_type not in [_s['type'] for _s in data['sessions']]:
  180. data['sessions'].append({
  181. 'type': snippet_type,
  182. 'properties': HS2Api.get_properties(snippet_type),
  183. 'id': None
  184. }
  185. )
  186. def execute(self, request, batch=False):
  187. from notebook.api import _execute_notebook # Cyclic dependency
  188. notebook_data = self.get_data()
  189. snippet = notebook_data['snippets'][0]
  190. snippet['wasBatchExecuted'] = batch
  191. return _execute_notebook(request, notebook_data, snippet)
  192. def get_api(request, snippet):
  193. from notebook.connectors.oozie_batch import OozieApi
  194. if snippet.get('wasBatchExecuted'):
  195. return OozieApi(user=request.user, request=request)
  196. interpreter = [interpreter for interpreter in get_ordered_interpreters(request.user) if interpreter['type'] == snippet['type']]
  197. if not interpreter:
  198. if snippet['type'] == 'hbase':
  199. interpreter = [{
  200. 'name': 'hbase',
  201. 'type': 'hbase',
  202. 'interface': 'hbase',
  203. 'options': {},
  204. 'is_sql': False
  205. }]
  206. elif snippet['type'] == 'kafka':
  207. interpreter = [{
  208. 'name': 'kafka',
  209. 'type': 'kafka',
  210. 'interface': 'kafka',
  211. 'options': {},
  212. 'is_sql': False
  213. }]
  214. elif snippet['type'] == 'solr':
  215. interpreter = [{
  216. 'name': 'solr',
  217. 'type': 'solr',
  218. 'interface': 'solr',
  219. 'options': {},
  220. 'is_sql': False
  221. }]
  222. else:
  223. raise PopupException(_('Snippet type %(type)s is not configured in hue.ini') % snippet)
  224. interpreter = interpreter[0]
  225. interface = interpreter['interface']
  226. # Multi cluster
  227. cluster = Cluster(request.user)
  228. if cluster and cluster.get_type() == 'dataeng':
  229. interface = 'dataeng'
  230. if interface == 'hiveserver2':
  231. from notebook.connectors.hiveserver2 import HS2Api
  232. return HS2Api(user=request.user, request=request)
  233. elif interface == 'oozie':
  234. return OozieApi(user=request.user, request=request)
  235. elif interface == 'livy':
  236. from notebook.connectors.spark_shell import SparkApi
  237. return SparkApi(request.user)
  238. elif interface == 'livy-batch':
  239. from notebook.connectors.spark_batch import SparkBatchApi
  240. return SparkBatchApi(request.user)
  241. elif interface == 'text' or interface == 'markdown':
  242. from notebook.connectors.text import TextApi
  243. return TextApi(request.user)
  244. elif interface == 'rdbms':
  245. from notebook.connectors.rdbms import RdbmsApi
  246. return RdbmsApi(request.user, interpreter=snippet['type'])
  247. elif interface == 'dataeng':
  248. from notebook.connectors.dataeng import DataEngApi
  249. return DataEngApi(user=request.user, request=request, cluster_name=cluster.get_interface())
  250. elif interface == 'jdbc' or interface == 'teradata':
  251. from notebook.connectors.jdbc import JdbcApi
  252. return JdbcApi(request.user, interpreter=interpreter)
  253. elif interface == 'solr':
  254. from notebook.connectors.solr import SolrApi
  255. return SolrApi(request.user, interpreter=interpreter)
  256. elif interface == 'hbase':
  257. from notebook.connectors.hbase import HBaseApi
  258. return HBaseApi(request.user)
  259. elif interface == 'kafka':
  260. from notebook.connectors.kafka import KafkaApi
  261. return KafkaApi(request.user)
  262. elif interface == 'pig':
  263. return OozieApi(user=request.user, request=request) # Backward compatibility until Hue 4
  264. else:
  265. raise PopupException(_('Notebook connector interface not recognized: %s') % interface)
  266. def _get_snippet_session(notebook, snippet):
  267. session = [session for session in notebook['sessions'] if session['type'] == snippet['type']]
  268. if not session:
  269. raise SessionExpired()
  270. else:
  271. return session[0]
  272. # Base API
  273. class Api(object):
  274. def __init__(self, user, interpreter=None, request=None):
  275. self.user = user
  276. self.interpreter = interpreter
  277. self.request = request
  278. def create_session(self, lang, properties=None):
  279. return {
  280. 'type': lang,
  281. 'id': None,
  282. 'properties': properties if not None else []
  283. }
  284. def close_session(self, session):
  285. pass
  286. def fetch_result(self, notebook, snippet, rows, start_over):
  287. pass
  288. def fetch_result_size(self, notebook, snippet):
  289. raise OperationNotSupported()
  290. def download(self, notebook, snippet, format, user_agent=None):
  291. pass
  292. def get_log(self, notebook, snippet, startFrom=None, size=None):
  293. return 'No logs'
  294. def autocomplete(self, snippet, database=None, table=None, column=None, nested=None):
  295. return {}
  296. def progress(self, snippet, logs=None):
  297. return 50
  298. def get_jobs(self, notebook, snippet, logs):
  299. return []
  300. def get_sample_data(self, snippet, database=None, table=None, column=None, async=False): raise NotImplementedError()
  301. def export_data_as_hdfs_file(self, snippet, target_file, overwrite): raise NotImplementedError()
  302. def export_data_as_table(self, notebook, snippet, destination, is_temporary=False, location=None): raise NotImplementedError()
  303. def export_large_data_to_hdfs(self, notebook, snippet, destination): raise NotImplementedError()
  304. def statement_risk(self, notebook, snippet): raise NotImplementedError()
  305. def statement_compatibility(self, notebook, snippet, source_platform, target_platform): raise NotImplementedError()
  306. def statement_similarity(self, notebook, snippet, source_platform, target_platform): raise NotImplementedError()
  307. def _get_snippet_name(notebook, unique=False, table_format=False):
  308. name = (('%(name)s' + ('-%(id)s' if unique else '') if notebook.get('name') else '%(type)s-%(id)s') % notebook)
  309. if table_format:
  310. name = re.sub('[-|\s:]', '_', name)
  311. return name