Skip to content

Commit

Permalink
EVA-3639 File stats job (#197)
Browse files Browse the repository at this point in the history
file stats job
  • Loading branch information
nitin-ebi authored Aug 29, 2024
1 parent 022ca8f commit 2cf0c11
Show file tree
Hide file tree
Showing 10 changed files with 487 additions and 34 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -189,4 +189,8 @@ public String getFileId() {
public String[] getAlternates() {
return alternates;
}

public BasicDBObject getAttrs() {
return attrs;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,8 @@ public class BeanNames {
public static final String DROP_FILES_BY_STUDY_STEP = "drop-files-by-study-step";
public static final String LOAD_ANNOTATION_METADATA_STEP = "annotation-metadata-step";
public static final String ACCESSION_IMPORT_STEP = "accession-import-step";
public static final String CALCULATE_AND_LOAD_STATISTICS_STEP = "calculate-load-statistics-step";
public static final String VARIANT_STATS_STEP = "variant-stats-step";
public static final String FILE_STATS_STEP = "file-stats-step";

public static final String AGGREGATED_VCF_JOB = "aggregated-vcf-job";
public static final String ANNOTATE_VARIANTS_JOB = "annotate-variants-job";
Expand All @@ -73,5 +74,6 @@ public class BeanNames {
public static final String CALCULATE_STATISTICS_JOB = "calculate-statistics-job";
public static final String DROP_STUDY_JOB = "drop-study-job";
public static final String ACCESSION_IMPORT_JOB = "accession-import-job";
public static final String CALCULATE_AND_LOAD_STATISTICS_JOB = "calculate-load-statistics-job";
public static final String VARIANT_STATS_JOB = "variant-stats-job";
public static final String FILE_STATS_JOB = "file-stats-job";
}
Original file line number Diff line number Diff line change
Expand Up @@ -27,37 +27,32 @@
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.Scope;
import uk.ac.ebi.eva.pipeline.configuration.jobs.steps.CalculateAndLoadStatisticsStepConfiguration;
import uk.ac.ebi.eva.pipeline.configuration.jobs.steps.FileStatsStepConfiguration;
import uk.ac.ebi.eva.pipeline.parameters.NewJobIncrementer;

import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.CALCULATE_AND_LOAD_STATISTICS_JOB;
import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.CALCULATE_AND_LOAD_STATISTICS_STEP;
import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.FILE_STATS_JOB;
import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.FILE_STATS_STEP;

/**
* Configuration to run a full Statistics job: variantStatsFlow: statsCreate --> statsLoad
* <p>
* TODO add a new PopulationStatisticsJobParametersValidator
*/
@Configuration
@EnableBatchProcessing
@Import({CalculateAndLoadStatisticsStepConfiguration.class})
public class CalculateAndLoadStatisticsJobConfiguration {
@Import({FileStatsStepConfiguration.class})
public class FileStatsJobConfiguration {

private static final Logger logger = LoggerFactory.getLogger(CalculateAndLoadStatisticsJobConfiguration.class);
private static final Logger logger = LoggerFactory.getLogger(FileStatsJobConfiguration.class);

@Autowired
@Qualifier(CALCULATE_AND_LOAD_STATISTICS_STEP)
private Step calculateAndLoadStatisticsStep;
@Qualifier(FILE_STATS_STEP)
private Step fileStatsStep;

@Bean(CALCULATE_AND_LOAD_STATISTICS_JOB)
@Bean(FILE_STATS_JOB)
@Scope("prototype")
public Job calculateAndLoadStatisticsJob(JobBuilderFactory jobBuilderFactory) {
logger.debug("Building '" + CALCULATE_AND_LOAD_STATISTICS_JOB + "'");
public Job fileStatsJob(JobBuilderFactory jobBuilderFactory) {
logger.debug("Building '" + FILE_STATS_JOB + "'");

return jobBuilderFactory
.get(CALCULATE_AND_LOAD_STATISTICS_JOB)
.get(FILE_STATS_JOB)
.incrementer(new NewJobIncrementer())
.start(calculateAndLoadStatisticsStep)
.start(fileStatsStep)
.build();
}

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
/*
* Copyright 2024 EMBL - European Bioinformatics Institute
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package uk.ac.ebi.eva.pipeline.configuration.jobs;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.Scope;
import uk.ac.ebi.eva.pipeline.configuration.jobs.steps.VariantStatsStepConfiguration;
import uk.ac.ebi.eva.pipeline.parameters.NewJobIncrementer;

import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.VARIANT_STATS_JOB;
import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.VARIANT_STATS_STEP;

@Configuration
@EnableBatchProcessing
@Import({VariantStatsStepConfiguration.class})
public class VariantStatsJobConfiguration {

private static final Logger logger = LoggerFactory.getLogger(VariantStatsJobConfiguration.class);

@Autowired
@Qualifier(VARIANT_STATS_STEP)
private Step variantStatsStep;

@Bean(VARIANT_STATS_JOB)
@Scope("prototype")
public Job variantStatsJob(JobBuilderFactory jobBuilderFactory) {
logger.debug("Building '" + VARIANT_STATS_JOB + "'");

return jobBuilderFactory
.get(VARIANT_STATS_JOB)
.incrementer(new NewJobIncrementer())
.start(variantStatsStep)
.build();
}

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
/*
* Copyright 2024 EMBL - European Bioinformatics Institute
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package uk.ac.ebi.eva.pipeline.configuration.jobs.steps;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.core.MongoTemplate;
import uk.ac.ebi.eva.pipeline.jobs.steps.tasklets.FileStatsTasklet;
import uk.ac.ebi.eva.pipeline.parameters.ChunkSizeParameters;
import uk.ac.ebi.eva.pipeline.parameters.DatabaseParameters;
import uk.ac.ebi.eva.pipeline.parameters.InputParameters;
import uk.ac.ebi.eva.pipeline.parameters.JobOptions;
import uk.ac.ebi.eva.utils.TaskletUtils;

import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.FILE_STATS_STEP;


@Configuration
@EnableBatchProcessing
public class FileStatsStepConfiguration {
private static final Logger logger = LoggerFactory.getLogger(FileStatsStepConfiguration.class);

@Bean
@StepScope
public FileStatsTasklet fileStatsTasklet(DatabaseParameters databaseParameters,
MongoTemplate mongoTemplate,
InputParameters inputParameters,
ChunkSizeParameters chunkSizeParameters) {
return new FileStatsTasklet(databaseParameters, mongoTemplate, inputParameters.getStudyId(),
chunkSizeParameters.getChunkSize());
}

@Bean(FILE_STATS_STEP)
public TaskletStep fileStatsStep(DatabaseParameters databaseParameters,
MongoTemplate mongoTemplate,
InputParameters inputParameters,
ChunkSizeParameters chunkSizeParameters,
StepBuilderFactory stepBuilderFactory,
JobOptions jobOptions) {
logger.debug("Building '" + FILE_STATS_STEP + "'");

return TaskletUtils.generateStep(stepBuilderFactory, FILE_STATS_STEP,
fileStatsTasklet(databaseParameters, mongoTemplate, inputParameters, chunkSizeParameters),
jobOptions.isAllowStartIfComplete());
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -33,26 +33,26 @@
import uk.ac.ebi.eva.pipeline.configuration.io.writers.VariantStatsWriterConfiguration;
import uk.ac.ebi.eva.pipeline.configuration.jobs.steps.processors.VariantStatsProcessorConfiguration;

import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.CALCULATE_AND_LOAD_STATISTICS_STEP;
import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.VARIANT_STATS_PROCESSOR;
import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.VARIANT_STATS_READER;
import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.VARIANT_STATS_STEP;
import static uk.ac.ebi.eva.pipeline.configuration.BeanNames.VARIANT_STATS_WRITER;


@Configuration
@EnableBatchProcessing
@Import({VariantStatsReaderConfiguration.class, VariantStatsWriterConfiguration.class,
VariantStatsProcessorConfiguration.class, ChunkSizeCompletionPolicyConfiguration.class})
public class CalculateAndLoadStatisticsStepConfiguration {
public class VariantStatsStepConfiguration {

@Bean(CALCULATE_AND_LOAD_STATISTICS_STEP)
public Step calculateAndLoadStatisticsStep(
@Bean(VARIANT_STATS_STEP)
public Step variantStatsStep(
@Qualifier(VARIANT_STATS_READER) ItemStreamReader<VariantDocument> variantStatsReader,
@Qualifier(VARIANT_STATS_PROCESSOR) ItemProcessor<VariantDocument, VariantDocument> variantStatsProcessor,
@Qualifier(VARIANT_STATS_WRITER) ItemWriter<VariantDocument> variantStatsWriter,
StepBuilderFactory stepBuilderFactory,
SimpleCompletionPolicy chunkSizeCompletionPolicy) {
TaskletStep step = stepBuilderFactory.get(CALCULATE_AND_LOAD_STATISTICS_STEP)
TaskletStep step = stepBuilderFactory.get(VARIANT_STATS_STEP)
.<VariantDocument, VariantDocument>chunk(chunkSizeCompletionPolicy)
.reader(variantStatsReader)
.processor(variantStatsProcessor)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@
import java.util.Map;
import java.util.stream.Collectors;

import static com.mongodb.client.model.Accumulators.sum;
import static com.mongodb.client.model.Aggregates.group;
import static com.mongodb.client.model.Aggregates.match;
import static com.mongodb.client.model.Aggregates.project;
import static com.mongodb.client.model.Projections.computed;
Expand Down Expand Up @@ -92,11 +94,13 @@ private void populateFilesIdAndNumberOfSamplesMap() {
computed("fid", "$fid"),
computed("numOfSamples", new Document("$size", new Document("$objectToArray", "$samp")))
));
Bson groupStage = group("$fid", sum("totalNumOfSamples", "$numOfSamples"));

filesIdNumberOfSamplesMap = mongoTemplate.getCollection(databaseParameters.getCollectionFilesName())
.aggregate(asList(matchStage, projectStage))
.aggregate(asList(matchStage, projectStage, groupStage))
.into(new ArrayList<>())
.stream()
.collect(Collectors.toMap(doc -> doc.getString("fid"), doc -> doc.getInteger("numOfSamples")));
.collect(Collectors.toMap(doc -> doc.getString("_id"), doc -> doc.getInteger("totalNumOfSamples")));
}

public static Map<String, Integer> getFilesIdAndNumberOfSamplesMap() {
Expand Down
Loading

0 comments on commit 2cf0c11

Please sign in to comment.