class Normalizer {

    public void normalize(){

        logger.info(() -> "Normalization started...");

        List<VortexData> source = new ArrayList<>();
        sourceVariables.forEach(variable -> source.addAll(
                DataReader.builder()
                        .path(pathToSource)
                        .variable(variable)
                        .build()
                        .getDtos()));

        List<VortexData> normals = new ArrayList<>();
        normalsVariables.forEach(variable -> normals.addAll(
                DataReader.builder()
                        .path(pathToNormals)
                        .variable(variable)
                        .build()
                        .getDtos()));

        AtomicReference<ZonedDateTime> intervalStart = new AtomicReference<>();
        intervalStart.set(startTime);

        AtomicInteger count = new AtomicInteger();
        while (intervalStart.get().isBefore(endTime)){
            ZonedDateTime start = intervalStart.get();
            ZonedDateTime end = intervalStart.get().plus(interval);

            List<VortexGrid> sourceFiltered = source.stream().filter(grid -> ((grid.startTime().isEqual(start) || grid.startTime().isAfter(start))
                    && (grid.endTime().isEqual(end) || grid.endTime().isBefore(end))))
                    .map(grid -> (VortexGrid)grid)
                    .collect(Collectors.toList());

            if (sourceFiltered.isEmpty()) {
                logger.warning(() -> format("No source grids for period %s to %s", formatter.format(start), formatter.format(end)));
            }

            List<VortexGrid> normalsFiltered = normals.stream().filter(grid -> ((grid.startTime().equals(start) || grid.startTime().isAfter(start))
                    && (grid.endTime().isEqual(end) || grid.endTime().isBefore(end))))
                    .map(grid -> (VortexGrid)grid)
                    .collect(Collectors.toList());

            if (normalsFiltered.isEmpty()){
                logger.warning(() -> format("No normals grids for period %s to %s", formatter.format(start), formatter.format(end)));
            }

            List<VortexGrid> output = normalize(sourceFiltered, normalsFiltered);

            output.forEach(grid -> {
                List<VortexData> data = new ArrayList<>();
                data.add(grid);

                DataWriter writer = DataWriter.builder()
                        .data(data)
                        .destination(destination)
                        .options(writeOptions)
                        .build();

                writer.write();
            });

            intervalStart.set(end);
            count.addAndGet(output.size());
        }
        logger.info(() -> format("Normalization complete: %s grids normalized.", count.get()));
    }

    private  Normalizer(NormalizerBuilder builder);

}