api3.py 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715
  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. from __future__ import print_function
  18. from future import standard_library
  19. standard_library.install_aliases()
  20. from builtins import oct
  21. from builtins import zip
  22. from past.builtins import basestring
  23. import chardet
  24. import json
  25. import logging
  26. import urllib.request, urllib.error
  27. import sys
  28. from django.urls import reverse
  29. from django.utils.translation import ugettext as _
  30. from django.views.decorators.http import require_POST
  31. from simple_salesforce.api import Salesforce
  32. from simple_salesforce.exceptions import SalesforceRefusedRequest
  33. from desktop.lib import django_mako
  34. from desktop.lib.django_util import JsonResponse
  35. from desktop.lib.exceptions_renderable import PopupException
  36. from desktop.lib.i18n import smart_unicode
  37. from desktop.models import Document2
  38. from kafka.kafka_api import get_topics
  39. from metadata.manager_client import ManagerApi
  40. from notebook.connectors.base import get_api, Notebook
  41. from notebook.decorators import api_error_handler
  42. from notebook.models import make_notebook, MockedDjangoRequest, escape_rows
  43. from indexer.controller import CollectionManagerController
  44. from indexer.file_format import HiveFormat
  45. from indexer.fields import Field
  46. from indexer.indexers.envelope import EnvelopeIndexer
  47. from indexer.models import _save_pipeline
  48. from indexer.indexers.morphline import MorphlineIndexer
  49. from indexer.indexers.rdbms import run_sqoop, _get_api
  50. from indexer.indexers.sql import SQLIndexer
  51. from indexer.solr_client import SolrClient, MAX_UPLOAD_SIZE
  52. from indexer.indexers.flume import FlumeIndexer
  53. if sys.version_info[0] > 2:
  54. from io import string_io as string_io
  55. from urllib.parse import urlparse, unquote as urllib_unquote
  56. else:
  57. from StringIO import StringIO as string_io
  58. from urllib import unquote as urllib_unquote
  59. from urlparse import urlparse
  60. LOG = logging.getLogger(__name__)
  61. try:
  62. from beeswax.server import dbms
  63. except ImportError as e:
  64. LOG.warn('Hive and HiveServer2 interfaces are not enabled')
  65. try:
  66. from filebrowser.views import detect_parquet
  67. except ImportError as e:
  68. LOG.warn('File Browser interface is not enabled')
  69. try:
  70. from search.conf import SOLR_URL
  71. except ImportError as e:
  72. LOG.warn('Solr Search interface is not enabled')
  73. def _escape_white_space_characters(s, inverse = False):
  74. MAPPINGS = {
  75. "\n": "\\n",
  76. "\t": "\\t",
  77. "\r": "\\r",
  78. " ": "\\s"
  79. }
  80. to = 1 if inverse else 0
  81. from_ = 0 if inverse else 1
  82. for pair in MAPPINGS.items():
  83. s = s.replace(pair[to], pair[from_]).encode('utf-8')
  84. return s
  85. def _convert_format(format_dict, inverse=False):
  86. for field in format_dict:
  87. if isinstance(format_dict[field], basestring):
  88. format_dict[field] = _escape_white_space_characters(format_dict[field], inverse)
  89. @api_error_handler
  90. def guess_format(request):
  91. file_format = json.loads(request.POST.get('fileFormat', '{}'))
  92. if file_format['inputFormat'] == 'file':
  93. path = urllib_unquote(file_format["path"])
  94. indexer = MorphlineIndexer(request.user, request.fs)
  95. if not request.fs.isfile(path):
  96. raise PopupException(_('Path %(path)s is not a file') % file_format)
  97. stream = request.fs.open(path)
  98. format_ = indexer.guess_format({
  99. "file": {
  100. "stream": stream,
  101. "name": path
  102. }
  103. })
  104. _convert_format(format_)
  105. elif file_format['inputFormat'] == 'table':
  106. db = dbms.get(request.user)
  107. try:
  108. table_metadata = db.get_table(database=file_format['databaseName'], table_name=file_format['tableName'])
  109. except Exception as e:
  110. raise PopupException(e.message if hasattr(e, 'message') and e.message else e)
  111. storage = {}
  112. for delim in table_metadata.storage_details:
  113. if delim['data_type']:
  114. if '=' in delim['data_type']:
  115. key, val = delim['data_type'].split('=', 1)
  116. storage[key] = val
  117. else:
  118. storage[delim['data_type']] = delim['comment']
  119. if table_metadata.details['properties']['format'] == 'text':
  120. format_ = {"quoteChar": "\"", "recordSeparator": '\\n', "type": "csv", "hasHeader": False, "fieldSeparator": storage.get('field.delim', ',')}
  121. elif table_metadata.details['properties']['format'] == 'parquet':
  122. format_ = {"type": "parquet", "hasHeader": False,}
  123. else:
  124. raise PopupException('Hive table format %s is not supported.' % table_metadata.details['properties']['format'])
  125. elif file_format['inputFormat'] == 'query':
  126. format_ = {"quoteChar": "\"", "recordSeparator": "\\n", "type": "csv", "hasHeader": False, "fieldSeparator": "\u0001"}
  127. elif file_format['inputFormat'] == 'rdbms':
  128. format_ = {"type": "csv"}
  129. elif file_format['inputFormat'] == 'stream':
  130. if file_format['streamSelection'] == 'kafka':
  131. format_ = {"type": "csv", "fieldSeparator": ",", "hasHeader": True, "quoteChar": "\"", "recordSeparator": "\\n", 'topics': get_topics()}
  132. elif file_format['streamSelection'] == 'flume':
  133. format_ = {"type": "csv", "fieldSeparator": ",", "hasHeader": True, "quoteChar": "\"", "recordSeparator": "\\n"}
  134. elif file_format['inputFormat'] == 'connector':
  135. if file_format['connectorSelection'] == 'sfdc':
  136. sf = Salesforce(
  137. username=file_format['streamUsername'],
  138. password=file_format['streamPassword'],
  139. security_token=file_format['streamToken']
  140. )
  141. format_ = {"type": "csv", "fieldSeparator": ",", "hasHeader": True, "quoteChar": "\"", "recordSeparator": "\\n", 'objects': [sobject['name'] for sobject in sf.restful('sobjects/')['sobjects'] if sobject['queryable']]}
  142. else:
  143. raise PopupException(_('Input format %(inputFormat)s connector not recognized: $(connectorSelection)s') % file_format)
  144. else:
  145. raise PopupException(_('Input format not recognized: %(inputFormat)s') % file_format)
  146. format_['status'] = 0
  147. return JsonResponse(format_)
  148. def guess_field_types(request):
  149. file_format = json.loads(request.POST.get('fileFormat', '{}'))
  150. if file_format['inputFormat'] == 'file':
  151. indexer = MorphlineIndexer(request.user, request.fs)
  152. path = urllib_unquote(file_format["path"])
  153. stream = request.fs.open(path)
  154. encoding = chardet.detect(stream.read(10000)).get('encoding')
  155. stream.seek(0)
  156. _convert_format(file_format["format"], inverse=True)
  157. format_ = indexer.guess_field_types({
  158. "file": {
  159. "stream": stream,
  160. "name": path
  161. },
  162. "format": file_format['format']
  163. })
  164. # Note: Would also need to set charset to table (only supported in Hive)
  165. if 'sample' in format_ and format_['sample']:
  166. format_['sample'] = escape_rows(format_['sample'], nulls_only=True, encoding=encoding)
  167. for col in format_['columns']:
  168. col['name'] = smart_unicode(col['name'], errors='replace', encoding=encoding)
  169. elif file_format['inputFormat'] == 'table':
  170. sample = get_api(request, {'type': 'hive'}).get_sample_data({'type': 'hive'}, database=file_format['databaseName'], table=file_format['tableName'])
  171. db = dbms.get(request.user)
  172. table_metadata = db.get_table(database=file_format['databaseName'], table_name=file_format['tableName'])
  173. format_ = {
  174. "sample": sample['rows'][:4],
  175. "columns": [
  176. Field(col.name, HiveFormat.FIELD_TYPE_TRANSLATE.get(col.type, 'string')).to_dict()
  177. for col in table_metadata.cols
  178. ]
  179. }
  180. elif file_format['inputFormat'] == 'query':
  181. query_id = file_format['query']['id'] if file_format['query'].get('id') else file_format['query']
  182. notebook = Notebook(document=Document2.objects.document(user=request.user, doc_id=query_id)).get_data()
  183. snippet = notebook['snippets'][0]
  184. db = get_api(request, snippet)
  185. if file_format.get('sampleCols'):
  186. columns = file_format.get('sampleCols')
  187. sample = file_format.get('sample')
  188. else:
  189. snippet['query'] = snippet['statement']
  190. try:
  191. sample = db.fetch_result(notebook, snippet, 4, start_over=True)['rows'][:4]
  192. except Exception as e:
  193. LOG.warn('Skipping sample data as query handle might be expired: %s' % e)
  194. sample = [[], [], [], [], []]
  195. columns = db.autocomplete(snippet=snippet, database='', table='')
  196. columns = [
  197. Field(col['name'], HiveFormat.FIELD_TYPE_TRANSLATE.get(col['type'], 'string')).to_dict()
  198. for col in columns['extended_columns']
  199. ]
  200. format_ = {
  201. "sample": sample,
  202. "columns": columns,
  203. }
  204. elif file_format['inputFormat'] == 'rdbms':
  205. api = _get_api(request)
  206. sample = api.get_sample_data(None, database=file_format['rdbmsDatabaseName'], table=file_format['tableName'])
  207. format_ = {
  208. "sample": list(sample['rows'])[:4],
  209. "columns": [
  210. Field(col['name'], col['type']).to_dict()
  211. for col in sample['full_headers']
  212. ]
  213. }
  214. elif file_format['inputFormat'] == 'stream':
  215. if file_format['streamSelection'] == 'kafka':
  216. if file_format.get('kafkaSelectedTopics') == 'NavigatorAuditEvents':
  217. kafkaFieldNames = [
  218. 'id',
  219. 'additionalInfo', 'allowed', 'collectionName', 'databaseName', 'db',
  220. 'DELEGATION_TOKEN_ID', 'dst', 'entityId', 'family', 'impersonator',
  221. 'ip', 'name', 'objectType', 'objType', 'objUsageType',
  222. 'operationParams', 'operationText', 'op', 'opText', 'path',
  223. 'perms', 'privilege', 'qualifier', 'QUERY_ID', 'resourcePath',
  224. 'service', 'SESSION_ID', 'solrVersion', 'src', 'status',
  225. 'subOperation', 'tableName', 'table', 'time', 'type',
  226. 'url', 'user'
  227. ]
  228. kafkaFieldTypes = [
  229. 'string'
  230. ] * len(kafkaFieldNames)
  231. kafkaFieldNames.append('timeDate')
  232. kafkaFieldTypes.append('date')
  233. else:
  234. # Note: mocked here, should come from SFDC or Kafka API or sampling job
  235. kafkaFieldNames = file_format.get('kafkaFieldNames', '').split(',')
  236. kafkaFieldTypes = file_format.get('kafkaFieldTypes', '').split(',')
  237. data = """%(kafkaFieldNames)s
  238. %(data)s""" % {
  239. 'kafkaFieldNames': ','.join(kafkaFieldNames),
  240. 'data': '\n'.join([','.join(['...'] * len(kafkaFieldTypes))] * 5)
  241. }
  242. stream = string_io()
  243. stream.write(data)
  244. _convert_format(file_format["format"], inverse=True)
  245. indexer = MorphlineIndexer(request.user, request.fs)
  246. format_ = indexer.guess_field_types({
  247. "file": {
  248. "stream": stream,
  249. "name": file_format['path']
  250. },
  251. "format": file_format['format']
  252. })
  253. type_mapping = dict(list(zip(kafkaFieldNames, kafkaFieldTypes)))
  254. for col in format_['columns']:
  255. col['keyType'] = type_mapping[col['name']]
  256. col['type'] = type_mapping[col['name']]
  257. elif file_format['streamSelection'] == 'flume':
  258. if 'hue-httpd/access_log' in file_format['channelSourcePath']:
  259. columns = [
  260. {'name': 'id', 'type': 'string', 'unique': True},
  261. {'name': 'client_ip', 'type': 'string'},
  262. {'name': 'time', 'type': 'date'},
  263. {'name': 'request', 'type': 'string'},
  264. {'name': 'code', 'type': 'plong'},
  265. {'name': 'bytes', 'type': 'plong'},
  266. {'name': 'method', 'type': 'string'},
  267. {'name': 'url', 'type': 'string'},
  268. {'name': 'protocol', 'type': 'string'},
  269. {'name': 'app', 'type': 'string'},
  270. {'name': 'subapp', 'type': 'string'}
  271. ]
  272. else:
  273. columns = [{'name': 'message', 'type': 'string'}]
  274. format_ = {
  275. "sample": [['...'] * len(columns)] * 4,
  276. "columns": [
  277. Field(col['name'], HiveFormat.FIELD_TYPE_TRANSLATE.get(col['type'], 'string'), unique=col.get('unique')).to_dict()
  278. for col in columns
  279. ]
  280. }
  281. elif file_format['inputFormat'] == 'connector':
  282. if file_format['connectorSelection'] == 'sfdc':
  283. sf = Salesforce(
  284. username=file_format['streamUsername'],
  285. password=file_format['streamPassword'],
  286. security_token=file_format['streamToken']
  287. )
  288. table_metadata = [{
  289. 'name': column['name'],
  290. 'type': column['type']
  291. } for column in sf.restful('sobjects/%(streamObject)s/describe/' % file_format)['fields']
  292. ]
  293. query = 'SELECT %s FROM %s LIMIT 4' % (', '.join([col['name'] for col in table_metadata]), file_format['streamObject'])
  294. print(query)
  295. try:
  296. records = sf.query_all(query)
  297. except SalesforceRefusedRequest as e:
  298. raise PopupException(message=str(e))
  299. format_ = {
  300. "sample": [list(row.values())[1:] for row in records['records']],
  301. "columns": [
  302. Field(col['name'], HiveFormat.FIELD_TYPE_TRANSLATE.get(col['type'], 'string')).to_dict()
  303. for col in table_metadata
  304. ]
  305. }
  306. else:
  307. raise PopupException(_('Connector format not recognized: %(connectorSelection)s') % file_format)
  308. else:
  309. raise PopupException(_('Input format not recognized: %(inputFormat)s') % file_format)
  310. return JsonResponse(format_)
  311. @api_error_handler
  312. def importer_submit(request):
  313. source = json.loads(request.POST.get('source', '{}'))
  314. outputFormat = json.loads(request.POST.get('destination', '{}'))['outputFormat']
  315. destination = json.loads(request.POST.get('destination', '{}'))
  316. destination['ouputFormat'] = outputFormat # Workaround a very weird bug
  317. start_time = json.loads(request.POST.get('start_time', '-1'))
  318. if source['inputFormat'] == 'file':
  319. if source['path']:
  320. path = urllib_unquote(source['path'])
  321. source['path'] = request.fs.netnormpath(path)
  322. if destination['ouputFormat'] in ('database', 'table'):
  323. destination['nonDefaultLocation'] = request.fs.netnormpath(destination['nonDefaultLocation']) if destination['nonDefaultLocation'] else destination['nonDefaultLocation']
  324. if destination['ouputFormat'] == 'index':
  325. source['columns'] = destination['columns']
  326. index_name = destination["name"]
  327. if destination['indexerRunJob'] or source['inputFormat'] == 'stream':
  328. _convert_format(source["format"], inverse=True)
  329. job_handle = _large_indexing(request, source, index_name, start_time=start_time, lib_path=destination['indexerJobLibPath'], destination=destination)
  330. else:
  331. client = SolrClient(request.user)
  332. job_handle = _small_indexing(request.user, request.fs, client, source, destination, index_name)
  333. elif source['inputFormat'] in ('stream', 'connector') or destination['ouputFormat'] == 'stream':
  334. job_handle = _envelope_job(request, source, destination, start_time=start_time, lib_path=destination['indexerJobLibPath'])
  335. elif source['inputFormat'] == 'altus':
  336. # BDR copy or DistCP + DDL + Sentry DDL copy
  337. pass
  338. elif source['inputFormat'] == 'rdbms':
  339. if destination['outputFormat'] in ('database', 'file', 'table', 'hbase'):
  340. job_handle = run_sqoop(request, source, destination, start_time)
  341. elif destination['ouputFormat'] == 'database':
  342. job_handle = _create_database(request, source, destination, start_time)
  343. else:
  344. job_handle = _create_table(request, source, destination, start_time)
  345. request.audit = {
  346. 'operation': 'EXPORT',
  347. 'operationText': 'User %(username)s exported %(inputFormat)s to %(ouputFormat)s: %(name)s' % {
  348. 'username': request.user.username,
  349. 'inputFormat': source['inputFormat'],
  350. 'ouputFormat': destination['ouputFormat'],
  351. 'name': destination['name'],
  352. },
  353. 'allowed': True
  354. }
  355. return JsonResponse(job_handle)
  356. def _small_indexing(user, fs, client, source, destination, index_name):
  357. kwargs = {}
  358. errors = []
  359. if source['inputFormat'] not in ('manual', 'table', 'query_handle'):
  360. path = urllib_unquote(source["path"])
  361. stats = fs.stats(path)
  362. if stats.size > MAX_UPLOAD_SIZE:
  363. raise PopupException(_('File size is too large to handle!'))
  364. indexer = MorphlineIndexer(user, fs)
  365. fields = indexer.get_field_list(destination['columns'])
  366. _create_solr_collection(user, fs, client, destination, index_name, kwargs)
  367. if source['inputFormat'] == 'file':
  368. path = urllib_unquote(source["path"])
  369. data = fs.read(path, 0, MAX_UPLOAD_SIZE)
  370. if client.is_solr_six_or_more():
  371. kwargs['processor'] = 'tolerant'
  372. kwargs['map'] = 'NULL:'
  373. try:
  374. if source['inputFormat'] == 'query':
  375. query_id = source['query']['id'] if source['query'].get('id') else source['query']
  376. notebook = Notebook(document=Document2.objects.document(user=user, doc_id=query_id)).get_data()
  377. request = MockedDjangoRequest(user=user)
  378. snippet = notebook['snippets'][0]
  379. searcher = CollectionManagerController(user)
  380. columns = [field['name'] for field in fields if field['name'] != 'hue_id']
  381. fetch_handle = lambda rows, start_over: get_api(request, snippet).fetch_result(notebook, snippet, rows=rows, start_over=start_over) # Assumes handle still live
  382. rows = searcher.update_data_from_hive(index_name, columns, fetch_handle=fetch_handle, indexing_options=kwargs)
  383. # TODO if rows == MAX_ROWS truncation warning
  384. elif source['inputFormat'] == 'manual':
  385. pass # No need to do anything
  386. else:
  387. response = client.index(name=index_name, data=data, **kwargs)
  388. errors = [error.get('message', '') for error in response['responseHeader'].get('errors', [])]
  389. except Exception as e:
  390. try:
  391. client.delete_index(index_name, keep_config=False)
  392. except Exception as e2:
  393. LOG.warn('Error while cleaning-up config of failed collection creation %s: %s' % (index_name, e2))
  394. raise e
  395. return {'status': 0, 'on_success_url': reverse('indexer:indexes', kwargs={'index': index_name}), 'pub_sub_url': 'assist.collections.refresh', 'errors': errors}
  396. def _create_database(request, source, destination, start_time):
  397. database = destination['name']
  398. comment = destination['description']
  399. use_default_location = destination['useDefaultLocation']
  400. external_path = destination['nonDefaultLocation']
  401. sql = django_mako.render_to_string("gen/create_database_statement.mako", {
  402. 'database': {
  403. 'name': database,
  404. 'comment': comment,
  405. 'use_default_location': use_default_location,
  406. 'external_location': external_path,
  407. 'properties': [],
  408. }
  409. }
  410. )
  411. editor_type = destination['apiHelperType']
  412. on_success_url = reverse('metastore:show_tables', kwargs={'database': database}) + "?source_type=" + source.get('sourceType', 'hive')
  413. notebook = make_notebook(
  414. name=_('Creating database %(name)s') % destination,
  415. editor_type=editor_type,
  416. statement=sql,
  417. status='ready',
  418. on_success_url=on_success_url,
  419. last_executed=start_time,
  420. is_task=True
  421. )
  422. return notebook.execute(request, batch=False)
  423. def _create_table(request, source, destination, start_time=-1):
  424. notebook = SQLIndexer(user=request.user, fs=request.fs).create_table_from_a_file(source, destination, start_time)
  425. if request.POST.get('show_command'):
  426. return {'status': 0, 'commands': notebook.get_str()}
  427. else:
  428. return notebook.execute(request, batch=False)
  429. def _large_indexing(request, file_format, collection_name, query=None, start_time=None, lib_path=None, destination=None):
  430. indexer = MorphlineIndexer(request.user, request.fs)
  431. unique_field = indexer.get_unique_field(file_format)
  432. is_unique_generated = indexer.is_unique_generated(file_format)
  433. schema_fields = indexer.get_kept_field_list(file_format['columns'])
  434. if is_unique_generated:
  435. schema_fields += [{"name": unique_field, "type": "string"}]
  436. client = SolrClient(user=request.user)
  437. if not client.exists(collection_name) and not request.POST.get('show_command'): # if destination['isTargetExisting']:
  438. client.create_index(
  439. name=collection_name,
  440. fields=request.POST.get('fields', schema_fields),
  441. unique_key_field=unique_field
  442. # No df currently
  443. )
  444. else:
  445. # TODO: check if format matches
  446. pass
  447. if file_format['inputFormat'] == 'table':
  448. db = dbms.get(request.user)
  449. table_metadata = db.get_table(database=file_format['databaseName'], table_name=file_format['tableName'])
  450. input_path = table_metadata.path_location
  451. elif file_format['inputFormat'] == 'stream' and file_format['streamSelection'] == 'flume':
  452. indexer = FlumeIndexer(user=request.user)
  453. if request.POST.get('show_command'):
  454. configs = indexer.generate_config(file_format, destination)
  455. return {'status': 0, 'commands': configs[-1]}
  456. else:
  457. return indexer.start(collection_name, file_format, destination)
  458. elif file_format['inputFormat'] == 'stream':
  459. return _envelope_job(request, file_format, destination, start_time=start_time, lib_path=lib_path)
  460. elif file_format['inputFormat'] == 'file':
  461. input_path = '${nameNode}%s' % urllib_unquote(file_format["path"])
  462. else:
  463. input_path = None
  464. morphline = indexer.generate_morphline_config(collection_name, file_format, unique_field, lib_path=lib_path)
  465. return indexer.run_morphline(request, collection_name, morphline, input_path, query, start_time=start_time, lib_path=lib_path)
  466. def _envelope_job(request, file_format, destination, start_time=None, lib_path=None):
  467. collection_name = destination['name']
  468. indexer = EnvelopeIndexer(request.user, request.fs)
  469. lib_path = None # Todo optional input field
  470. input_path = None
  471. if file_format['inputFormat'] == 'table':
  472. db = dbms.get(request.user)
  473. table_metadata = db.get_table(database=file_format['databaseName'], table_name=file_format['tableName'])
  474. input_path = table_metadata.path_location
  475. elif file_format['inputFormat'] == 'file':
  476. input_path = file_format["path"]
  477. properties = {
  478. 'input_path': input_path,
  479. 'format': 'csv'
  480. }
  481. elif file_format['inputFormat'] == 'stream' and file_format['streamSelection'] == 'flume':
  482. pass
  483. elif file_format['inputFormat'] == 'stream':
  484. if file_format['streamSelection'] == 'kafka':
  485. manager = ManagerApi()
  486. properties = {
  487. "brokers": manager.get_kafka_brokers(),
  488. "topics": file_format['kafkaSelectedTopics'],
  489. "kafkaFieldType": file_format['kafkaFieldType'],
  490. "kafkaFieldDelimiter": file_format['kafkaFieldDelimiter'],
  491. }
  492. if file_format.get('kafkaSelectedTopics') == 'NavigatorAuditEvents':
  493. schema_fields = MorphlineIndexer.get_kept_field_list(file_format['sampleCols'])
  494. properties.update({
  495. "kafkaFieldNames": ', '.join([_field['name'] for _field in schema_fields]),
  496. "kafkaFieldTypes": ', '.join([_field['type'] for _field in schema_fields])
  497. })
  498. else:
  499. properties.update({
  500. "kafkaFieldNames": file_format['kafkaFieldNames'],
  501. "kafkaFieldTypes": file_format['kafkaFieldTypes']
  502. })
  503. if True:
  504. properties['window'] = ''
  505. else: # For "KafkaSQL"
  506. properties['window'] = '''
  507. window {
  508. enabled = true
  509. milliseconds = 60000
  510. }'''
  511. elif file_format['inputFormat'] == 'connector':
  512. if file_format['streamSelection'] == 'flume':
  513. properties = {
  514. 'streamSelection': file_format['streamSelection'],
  515. 'channelSourceHosts': file_format['channelSourceHosts'],
  516. 'channelSourceSelectedHosts': file_format['channelSourceSelectedHosts'],
  517. 'channelSourcePath': file_format['channelSourcePath'],
  518. }
  519. else:
  520. # sfdc
  521. properties = {
  522. 'streamSelection': file_format['streamSelection'],
  523. 'streamUsername': file_format['streamUsername'],
  524. 'streamPassword': file_format['streamPassword'],
  525. 'streamToken': file_format['streamToken'],
  526. 'streamEndpointUrl': file_format['streamEndpointUrl'],
  527. 'streamObject': file_format['streamObject'],
  528. }
  529. if destination['outputFormat'] == 'table':
  530. if destination['isTargetExisting']: # Todo: check if format matches
  531. pass
  532. else:
  533. destination['importData'] = False # Avoid LOAD DATA
  534. if destination['tableFormat'] == 'kudu':
  535. properties['kafkaFieldNames'] = properties['kafkaFieldNames'].lower() # Kudu names should be all lowercase
  536. # Create table
  537. if not request.POST.get('show_command'):
  538. SQLIndexer(user=request.user, fs=request.fs).create_table_from_a_file(file_format, destination).execute(request)
  539. if destination['tableFormat'] == 'kudu':
  540. manager = ManagerApi()
  541. properties["output_table"] = "impala::%s" % collection_name
  542. properties["kudu_master"] = manager.get_kudu_master()
  543. else:
  544. properties['output_table'] = collection_name
  545. elif destination['outputFormat'] == 'stream':
  546. manager = ManagerApi()
  547. properties['brokers'] = manager.get_kafka_brokers()
  548. properties['topics'] = file_format['kafkaSelectedTopics']
  549. properties['kafkaFieldDelimiter'] = file_format['kafkaFieldDelimiter']
  550. elif destination['outputFormat'] == 'file':
  551. properties['path'] = file_format["path"]
  552. if file_format['inputFormat'] == 'stream':
  553. properties['format'] = 'csv'
  554. else:
  555. properties['format'] = file_format['tableFormat'] # or csv
  556. elif destination['outputFormat'] == 'index':
  557. properties['collectionName'] = collection_name
  558. properties['connection'] = SOLR_URL.get()
  559. properties["app_name"] = 'Data Ingest'
  560. properties["inputFormat"] = file_format['inputFormat']
  561. properties["ouputFormat"] = destination['ouputFormat']
  562. properties["streamSelection"] = file_format["streamSelection"]
  563. configs = indexer.generate_config(properties)
  564. if request.POST.get('show_command'):
  565. return {'status': 0, 'commands': configs['envelope.conf']}
  566. else:
  567. return indexer.run(request, collection_name, configs, input_path, start_time=start_time, lib_path=lib_path)
  568. def _create_solr_collection(user, fs, client, destination, index_name, kwargs):
  569. unique_key_field = destination['indexerPrimaryKey'] and destination['indexerPrimaryKey'][0] or None
  570. df = destination['indexerDefaultField'] and destination['indexerDefaultField'][0] or None
  571. indexer = MorphlineIndexer(user, fs)
  572. fields = indexer.get_field_list(destination['columns'])
  573. skip_fields = [field['name'] for field in fields if not field['keep']]
  574. kwargs['fieldnames'] = ','.join([field['name'] for field in fields])
  575. for field in fields:
  576. for operation in field['operations']:
  577. if operation['type'] == 'split':
  578. field['multiValued'] = True # Solr requires multiValued to be set when splitting
  579. kwargs['f.%(name)s.split' % field] = 'true'
  580. kwargs['f.%(name)s.separator' % field] = operation['settings']['splitChar'] or ','
  581. if skip_fields:
  582. kwargs['skip'] = ','.join(skip_fields)
  583. fields = [field for field in fields if field['name'] not in skip_fields]
  584. if not unique_key_field:
  585. unique_key_field = 'hue_id'
  586. fields += [{"name": unique_key_field, "type": "string"}]
  587. kwargs['rowid'] = unique_key_field
  588. if not destination['hasHeader']:
  589. kwargs['header'] = 'false'
  590. else:
  591. kwargs['skipLines'] = 1
  592. if not client.exists(index_name):
  593. client.create_index(
  594. name=index_name,
  595. config_name=destination.get('indexerConfigSet'),
  596. fields=fields,
  597. unique_key_field=unique_key_field,
  598. df=df,
  599. shards=destination['indexerNumShards'],
  600. replication=destination['indexerReplicationFactor']
  601. )
  602. @api_error_handler
  603. @require_POST
  604. # @check_document_modify_permission()
  605. def save_pipeline(request):
  606. response = {'status': -1}
  607. notebook = json.loads(request.POST.get('notebook', '{}'))
  608. notebook_doc, save_as = _save_pipeline(notebook, request.user)
  609. response['status'] = 0
  610. response['save_as'] = save_as
  611. response.update(notebook_doc.to_dict())
  612. response['message'] = request.POST.get('editorMode') == 'true' and _('Query saved successfully') or _('Notebook saved successfully')
  613. return JsonResponse(response)