yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-11-10 08:21:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Support MCE save Schemaless (#908)

Browse Source
This commit is contained in:
Yi (Alan) Wang 2017-12-10 11:47:50 -08:00 committed by GitHub
parent c9cd9b3676
commit 67913fd752
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 26 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
wherehows-dao/src/main/java/wherehows/dao/table/FieldDetailDao.java
View File

@ -73,6 +73,18 @@ public class FieldDetailDao extends BaseDao {
// update field comments? // update field comments?
} }
/**
* Insert or update Schemaless from MetadataChangeEvent
* @param identifier DatasetIdentifier
* @param datasetId int
* @param auditStamp ChangeAuditStamp
* @throws Exception
*/
public void insertUpdateSchemaless(@Nonnull DatasetIdentifier identifier, int datasetId,
@Nonnull ChangeAuditStamp auditStamp) throws Exception {
throw new UnsupportedOperationException("Support for Schemaless not yet implemented.");
}
/** /**
* Fill in DictFieldDetail information from FieldSchema * Fill in DictFieldDetail information from FieldSchema
* @param fs FieldSchema * @param fs FieldSchema

26
wherehows-kafka/src/main/java/wherehows/processors/MetadataChangeProcessor.java
View File

@ -65,7 +65,7 @@ public class MetadataChangeProcessor extends KafkaMessageProcessor {
log.debug("Processing Metadata Change Event record."); log.debug("Processing Metadata Change Event record.");
MetadataChangeEvent event = (MetadataChangeEvent) indexedRecord; final MetadataChangeEvent event = (MetadataChangeEvent) indexedRecord;
try { try {
processEvent(event); processEvent(event);
} catch (Exception exception) { } catch (Exception exception) {
@ -91,23 +91,25 @@ public class MetadataChangeProcessor extends KafkaMessageProcessor {
throw new Exception("Dataset name too long: " + identifier); throw new Exception("Dataset name too long: " + identifier);
} }
// if DELETE, mark dataset as removed and return
if (changeType == ChangeType.DELETE) { if (changeType == ChangeType.DELETE) {
_dictDatasetDao.setDatasetRemoved(identifier, true, changeAuditStamp); _dictDatasetDao.setDatasetRemoved(identifier, true, changeAuditStamp);
return; return;
} }
final DatasetSchema dsSchema = final DatasetSchema dsSchema = event.schema instanceof DatasetSchema ? (DatasetSchema) event.schema : null;
(event.schema == null || event.schema instanceof Schemaless) ? null : (DatasetSchema) event.schema;
// TODO: handle Schemaless separately
// create or update dataset // create or update dataset
DictDataset ds = _dictDatasetDao.insertUpdateDataset(identifier, changeAuditStamp, event.datasetProperty, dsSchema, final DictDataset ds =
event.deploymentInfo, toStringList(event.tags), event.capacity, event.partitionSpec); _dictDatasetDao.insertUpdateDataset(identifier, changeAuditStamp, event.datasetProperty, dsSchema,
event.deploymentInfo, toStringList(event.tags), event.capacity, event.partitionSpec);
// if schema is not null, insert or update schema // if schema is not null, insert or update schema
if (dsSchema != null) { if (dsSchema != null) { // if instanceof DatasetSchema
_fieldDetailDao.insertUpdateDatasetFields(identifier, ds.getId(), event.datasetProperty, changeAuditStamp, _fieldDetailDao.insertUpdateDatasetFields(identifier, ds.getId(), event.datasetProperty, changeAuditStamp,
dsSchema); dsSchema);
} else if (event.schema instanceof Schemaless) { // if instanceof Schemaless
_fieldDetailDao.insertUpdateSchemaless(identifier, ds.getId(), changeAuditStamp);
} }
// if owners are not null, insert or update owner // if owners are not null, insert or update owner
@ -128,10 +130,10 @@ public class MetadataChangeProcessor extends KafkaMessageProcessor {
} }
private FailedMetadataChangeEvent newFailedEvent(MetadataChangeEvent event, Throwable throwable) { private FailedMetadataChangeEvent newFailedEvent(MetadataChangeEvent event, Throwable throwable) {
FailedMetadataChangeEvent faileEvent = new FailedMetadataChangeEvent(); FailedMetadataChangeEvent failedEvent = new FailedMetadataChangeEvent();
faileEvent.time = System.currentTimeMillis(); failedEvent.time = System.currentTimeMillis();
faileEvent.error = ExceptionUtils.getStackTrace(throwable); failedEvent.error = ExceptionUtils.getStackTrace(throwable);
faileEvent.metadataChangeEvent = event; failedEvent.metadataChangeEvent = event;
return faileEvent; return failedEvent;
} }
} }