@Test public void testAnalysisWithInvalidValues() { String dsId = UUID.randomUUID().toString(); final DataSetMetadata metadata = metadataBuilder.metadata().id(dsId).build(); dataSetMetadataRepository.save(metadata); contentStore.storeAsRaw(metadata, DataSetServiceTest.class.getResourceAsStream("../dataset_with_invalid_records.csv")); formatAnalysis.analyze(dsId); contentAnalysis.analyze(dsId); schemaAnalysis.analyze(dsId); qualityAnalysis.analyze(dsId); final DataSetMetadata actual = dataSetMetadataRepository.get(dsId); assertThat(actual.getLifecycle().qualityAnalyzed(), is(true)); assertThat(actual.getContent().getNbRecords(), is(9L)); assertThat(actual.getRowMetadata().getColumns().size(), is(2)); ColumnMetadata secondColumn = actual.getRowMetadata().getColumns().get(1); Quality quality = secondColumn.getQuality(); assertThat(quality.getValid(), is(6)); assertThat(quality.getInvalid(), is(2)); assertThat(quality.getEmpty(), is(1)); }
@Override public void analyze(String dataSetId) { if (StringUtils.isEmpty(dataSetId)) { throw new IllegalArgumentException("Data set id cannot be null or empty."); } DistributedLock datasetLock = repository.createDatasetMetadataLock(dataSetId); datasetLock.lock(); try { DataSetMetadata metadata = repository.get(dataSetId); if (metadata == null) { LOGGER.info("Unable to analyze quality of data set #{}: seems to be removed.", dataSetId); return; } if (!metadata.getLifecycle().isInProgress()) { LOGGER.debug("No need to recompute quality of data set #{} (statistics are completed).", dataSetId); return; } if (!metadata.getLifecycle().schemaAnalyzed()) { LOGGER.debug( "Schema information must be computed before quality analysis can be performed, ignoring message"); return; } try (Stream<DataSetRow> stream = store.stream(metadata)) { LOGGER.debug("Analyzing quality of dataset #{}...", metadata.getId()); final long dataSetSize = metadata.getContent().getNbRecords(); final boolean isNewDataSet = dataSetSize == 0; if (isNewDataSet || dataSetSize == maxRecord) { computeQuality(metadata, stream, dataSetSize == maxRecord ? -1 : maxRecord); } if (isNewDataSet && metadata.getContent().getNbRecords() >= maxRecord) { metadata.getLifecycle().setInProgress(true); } else { metadata.getLifecycle().setInProgress(false); } metadata.getLifecycle().qualityAnalyzed(true); final DataSetMetadata savedDataSetMetadata = repository.get(metadata.getId()); if (savedDataSetMetadata != null) { repository.save(metadata); LOGGER.debug("Analyzed quality of dataset #{}.", dataSetId); } } catch (Exception e) { LOGGER.warn("dataset '{}' generate an error, message: {} ", dataSetId, e.getMessage()); throw new TDPException(DataSetErrorCodes.UNABLE_TO_ANALYZE_DATASET_QUALITY, e); } } finally { datasetLock.unlock(); } }
QualityAnalysis implements SynchronousDataSetAnalyzer { @Override public void analyze(String dataSetId) { if (StringUtils.isEmpty(dataSetId)) { throw new IllegalArgumentException("Data set id cannot be null or empty."); } DistributedLock datasetLock = repository.createDatasetMetadataLock(dataSetId); datasetLock.lock(); try { DataSetMetadata metadata = repository.get(dataSetId); if (metadata == null) { LOGGER.info("Unable to analyze quality of data set #{}: seems to be removed.", dataSetId); return; } if (!metadata.getLifecycle().isInProgress()) { LOGGER.debug("No need to recompute quality of data set #{} (statistics are completed).", dataSetId); return; } if (!metadata.getLifecycle().schemaAnalyzed()) { LOGGER.debug( "Schema information must be computed before quality analysis can be performed, ignoring message"); return; } try (Stream<DataSetRow> stream = store.stream(metadata)) { LOGGER.debug("Analyzing quality of dataset #{}...", metadata.getId()); final long dataSetSize = metadata.getContent().getNbRecords(); final boolean isNewDataSet = dataSetSize == 0; if (isNewDataSet || dataSetSize == maxRecord) { computeQuality(metadata, stream, dataSetSize == maxRecord ? -1 : maxRecord); } if (isNewDataSet && metadata.getContent().getNbRecords() >= maxRecord) { metadata.getLifecycle().setInProgress(true); } else { metadata.getLifecycle().setInProgress(false); } metadata.getLifecycle().qualityAnalyzed(true); final DataSetMetadata savedDataSetMetadata = repository.get(metadata.getId()); if (savedDataSetMetadata != null) { repository.save(metadata); LOGGER.debug("Analyzed quality of dataset #{}.", dataSetId); } } catch (Exception e) { LOGGER.warn("dataset '{}' generate an error, message: {} ", dataSetId, e.getMessage()); throw new TDPException(DataSetErrorCodes.UNABLE_TO_ANALYZE_DATASET_QUALITY, e); } } finally { datasetLock.unlock(); } } }
QualityAnalysis implements SynchronousDataSetAnalyzer { @Override public void analyze(String dataSetId) { if (StringUtils.isEmpty(dataSetId)) { throw new IllegalArgumentException("Data set id cannot be null or empty."); } DistributedLock datasetLock = repository.createDatasetMetadataLock(dataSetId); datasetLock.lock(); try { DataSetMetadata metadata = repository.get(dataSetId); if (metadata == null) { LOGGER.info("Unable to analyze quality of data set #{}: seems to be removed.", dataSetId); return; } if (!metadata.getLifecycle().isInProgress()) { LOGGER.debug("No need to recompute quality of data set #{} (statistics are completed).", dataSetId); return; } if (!metadata.getLifecycle().schemaAnalyzed()) { LOGGER.debug( "Schema information must be computed before quality analysis can be performed, ignoring message"); return; } try (Stream<DataSetRow> stream = store.stream(metadata)) { LOGGER.debug("Analyzing quality of dataset #{}...", metadata.getId()); final long dataSetSize = metadata.getContent().getNbRecords(); final boolean isNewDataSet = dataSetSize == 0; if (isNewDataSet || dataSetSize == maxRecord) { computeQuality(metadata, stream, dataSetSize == maxRecord ? -1 : maxRecord); } if (isNewDataSet && metadata.getContent().getNbRecords() >= maxRecord) { metadata.getLifecycle().setInProgress(true); } else { metadata.getLifecycle().setInProgress(false); } metadata.getLifecycle().qualityAnalyzed(true); final DataSetMetadata savedDataSetMetadata = repository.get(metadata.getId()); if (savedDataSetMetadata != null) { repository.save(metadata); LOGGER.debug("Analyzed quality of dataset #{}.", dataSetId); } } catch (Exception e) { LOGGER.warn("dataset '{}' generate an error, message: {} ", dataSetId, e.getMessage()); throw new TDPException(DataSetErrorCodes.UNABLE_TO_ANALYZE_DATASET_QUALITY, e); } } finally { datasetLock.unlock(); } }  }
QualityAnalysis implements SynchronousDataSetAnalyzer { @Override public void analyze(String dataSetId) { if (StringUtils.isEmpty(dataSetId)) { throw new IllegalArgumentException("Data set id cannot be null or empty."); } DistributedLock datasetLock = repository.createDatasetMetadataLock(dataSetId); datasetLock.lock(); try { DataSetMetadata metadata = repository.get(dataSetId); if (metadata == null) { LOGGER.info("Unable to analyze quality of data set #{}: seems to be removed.", dataSetId); return; } if (!metadata.getLifecycle().isInProgress()) { LOGGER.debug("No need to recompute quality of data set #{} (statistics are completed).", dataSetId); return; } if (!metadata.getLifecycle().schemaAnalyzed()) { LOGGER.debug( "Schema information must be computed before quality analysis can be performed, ignoring message"); return; } try (Stream<DataSetRow> stream = store.stream(metadata)) { LOGGER.debug("Analyzing quality of dataset #{}...", metadata.getId()); final long dataSetSize = metadata.getContent().getNbRecords(); final boolean isNewDataSet = dataSetSize == 0; if (isNewDataSet || dataSetSize == maxRecord) { computeQuality(metadata, stream, dataSetSize == maxRecord ? -1 : maxRecord); } if (isNewDataSet && metadata.getContent().getNbRecords() >= maxRecord) { metadata.getLifecycle().setInProgress(true); } else { metadata.getLifecycle().setInProgress(false); } metadata.getLifecycle().qualityAnalyzed(true); final DataSetMetadata savedDataSetMetadata = repository.get(metadata.getId()); if (savedDataSetMetadata != null) { repository.save(metadata); LOGGER.debug("Analyzed quality of dataset #{}.", dataSetId); } } catch (Exception e) { LOGGER.warn("dataset '{}' generate an error, message: {} ", dataSetId, e.getMessage()); throw new TDPException(DataSetErrorCodes.UNABLE_TO_ANALYZE_DATASET_QUALITY, e); } } finally { datasetLock.unlock(); } }  @Override void analyze(String dataSetId); @Override int order(); void computeQuality(DataSetMetadata dataset, Stream<DataSetRow> records, long limit); }
QualityAnalysis implements SynchronousDataSetAnalyzer { @Override public void analyze(String dataSetId) { if (StringUtils.isEmpty(dataSetId)) { throw new IllegalArgumentException("Data set id cannot be null or empty."); } DistributedLock datasetLock = repository.createDatasetMetadataLock(dataSetId); datasetLock.lock(); try { DataSetMetadata metadata = repository.get(dataSetId); if (metadata == null) { LOGGER.info("Unable to analyze quality of data set #{}: seems to be removed.", dataSetId); return; } if (!metadata.getLifecycle().isInProgress()) { LOGGER.debug("No need to recompute quality of data set #{} (statistics are completed).", dataSetId); return; } if (!metadata.getLifecycle().schemaAnalyzed()) { LOGGER.debug( "Schema information must be computed before quality analysis can be performed, ignoring message"); return; } try (Stream<DataSetRow> stream = store.stream(metadata)) { LOGGER.debug("Analyzing quality of dataset #{}...", metadata.getId()); final long dataSetSize = metadata.getContent().getNbRecords(); final boolean isNewDataSet = dataSetSize == 0; if (isNewDataSet || dataSetSize == maxRecord) { computeQuality(metadata, stream, dataSetSize == maxRecord ? -1 : maxRecord); } if (isNewDataSet && metadata.getContent().getNbRecords() >= maxRecord) { metadata.getLifecycle().setInProgress(true); } else { metadata.getLifecycle().setInProgress(false); } metadata.getLifecycle().qualityAnalyzed(true); final DataSetMetadata savedDataSetMetadata = repository.get(metadata.getId()); if (savedDataSetMetadata != null) { repository.save(metadata); LOGGER.debug("Analyzed quality of dataset #{}.", dataSetId); } } catch (Exception e) { LOGGER.warn("dataset '{}' generate an error, message: {} ", dataSetId, e.getMessage()); throw new TDPException(DataSetErrorCodes.UNABLE_TO_ANALYZE_DATASET_QUALITY, e); } } finally { datasetLock.unlock(); } }  @Override void analyze(String dataSetId); @Override int order(); void computeQuality(DataSetMetadata dataset, Stream<DataSetRow> records, long limit);  }