|
@@ -263,6 +263,8 @@ def guess_field_types(request):
|
|
|
kafkaFieldTypes = [
|
|
kafkaFieldTypes = [
|
|
|
'string'
|
|
'string'
|
|
|
] * len(kafkaFieldNames)
|
|
] * len(kafkaFieldNames)
|
|
|
|
|
+ kafkaFieldNames.append('timeDate')
|
|
|
|
|
+ kafkaFieldTypes.append('date')
|
|
|
else:
|
|
else:
|
|
|
kafkaFieldNames = file_format.get('kafkaFieldNames', '').split(',')
|
|
kafkaFieldNames = file_format.get('kafkaFieldNames', '').split(',')
|
|
|
kafkaFieldTypes = file_format.get('kafkaFieldTypes', '').split(',')
|
|
kafkaFieldTypes = file_format.get('kafkaFieldTypes', '').split(',')
|
|
@@ -477,21 +479,23 @@ def _large_indexing(request, file_format, collection_name, query=None, start_tim
|
|
|
|
|
|
|
|
client = SolrClient(user=request.user)
|
|
client = SolrClient(user=request.user)
|
|
|
|
|
|
|
|
- if not client.exists(collection_name):
|
|
|
|
|
|
|
+ if not client.exists(collection_name): # if destination['isTargetExisting']:
|
|
|
client.create_index(
|
|
client.create_index(
|
|
|
name=collection_name,
|
|
name=collection_name,
|
|
|
fields=request.POST.get('fields', schema_fields),
|
|
fields=request.POST.get('fields', schema_fields),
|
|
|
unique_key_field=unique_field
|
|
unique_key_field=unique_field
|
|
|
# No df currently
|
|
# No df currently
|
|
|
)
|
|
)
|
|
|
|
|
+ else:
|
|
|
|
|
+ # TODO: check if format matches
|
|
|
|
|
+ pass
|
|
|
|
|
|
|
|
if file_format['inputFormat'] == 'table':
|
|
if file_format['inputFormat'] == 'table':
|
|
|
db = dbms.get(request.user)
|
|
db = dbms.get(request.user)
|
|
|
table_metadata = db.get_table(database=file_format['databaseName'], table_name=file_format['tableName'])
|
|
table_metadata = db.get_table(database=file_format['databaseName'], table_name=file_format['tableName'])
|
|
|
input_path = table_metadata.path_location
|
|
input_path = table_metadata.path_location
|
|
|
elif file_format['inputFormat'] == 'stream':
|
|
elif file_format['inputFormat'] == 'stream':
|
|
|
- pass
|
|
|
|
|
- #job_handle = _envelope_job(request, source, destination, start_time=start_time, lib_path=destination['indexerJobLibPath'])
|
|
|
|
|
|
|
+ return _envelope_job(request, file_format, collection_name, start_time=start_time, lib_path=lib_path)
|
|
|
elif file_format['inputFormat'] == 'file':
|
|
elif file_format['inputFormat'] == 'file':
|
|
|
input_path = '${nameNode}%s' % urllib.unquote(file_format["path"])
|
|
input_path = '${nameNode}%s' % urllib.unquote(file_format["path"])
|
|
|
else:
|
|
else:
|
|
@@ -571,13 +575,14 @@ def _envelope_job(request, file_format, destination, start_time=None, lib_path=N
|
|
|
elif destination['outputFormat'] == 'index':
|
|
elif destination['outputFormat'] == 'index':
|
|
|
properties['collectionName'] = collection_name
|
|
properties['collectionName'] = collection_name
|
|
|
properties['connection'] = SOLR_URL.get()
|
|
properties['connection'] = SOLR_URL.get()
|
|
|
- if destination['isTargetExisting']:
|
|
|
|
|
- # Todo: check if format matches
|
|
|
|
|
- pass
|
|
|
|
|
- else:
|
|
|
|
|
- client = SolrClient(request.user)
|
|
|
|
|
- kwargs = {}
|
|
|
|
|
- _create_solr_collection(request.user, request.fs, client, destination, collection_name, kwargs)
|
|
|
|
|
|
|
+# No needed anymore
|
|
|
|
|
+# if destination['isTargetExisting']:
|
|
|
|
|
+# # Todo: check if format matches
|
|
|
|
|
+# pass
|
|
|
|
|
+# else:
|
|
|
|
|
+# client = SolrClient(request.user)
|
|
|
|
|
+# kwargs = {}
|
|
|
|
|
+# _create_solr_collection(request.user, request.fs, client, destination, collection_name, kwargs)
|
|
|
|
|
|
|
|
if destination['outputFormat'] == 'stream':
|
|
if destination['outputFormat'] == 'stream':
|
|
|
manager = ManagerApi()
|
|
manager = ManagerApi()
|