Можете ли вы восстановить баланс несбалансированного Spliterator неизвестного размера?

Я хочу использовать Streamдля распараллеливания обработки разнородного набора JSON-файлов с удаленным хранением неизвестного числа (количество файлов не известно заранее). Размер файлов может варьироваться от 1 записи JSON на файл до 100 000 записей в некоторых других файлах. Запись JSON в этом случае означает автономный объект JSON, представленный одной строкой в ​​файле.

Я действительно хочу использовать Streams для этого, и поэтому я реализовал это Spliterator:

public abstract class JsonStreamSpliterator<METADATA, RECORD> extends AbstractSpliterator<RECORD> {

    abstract protected JsonStreamSupport<METADATA> openInputStream(String path);

    abstract protected RECORD parse(METADATA metadata, Map<String, Object> json);

    private static final int ADDITIONAL_CHARACTERISTICS = Spliterator.IMMUTABLE | Spliterator.DISTINCT | Spliterator.NONNULL;
    private static final int MAX_BUFFER = 100;
    private final Iterator<String> paths;
    private JsonStreamSupport<METADATA> reader = null;

    public JsonStreamSpliterator(Iterator<String> paths) {
        this(Long.MAX_VALUE, ADDITIONAL_CHARACTERISTICS, paths);
    }

    private JsonStreamSpliterator(long est, int additionalCharacteristics, Iterator<String> paths) {
        super(est, additionalCharacteristics);
        this.paths = paths;
    }

    private JsonStreamSpliterator(long est, int additionalCharacteristics, Iterator<String> paths, String nextPath) {
        this(est, additionalCharacteristics, paths);
        open(nextPath);
    }

    @Override
    public boolean tryAdvance(Consumer<? super RECORD> action) {
        if(reader == null) {
            String path = takeNextPath();
            if(path != null) {
                open(path);
            }
            else {
                return false;
            }
        }
        Map<String, Object> json = reader.readJsonLine();
        if(json != null) {
            RECORD item = parse(reader.getMetadata(), json);
            action.accept(item);
            return true;
        }
        else {
            reader.close();
            reader = null;
            return tryAdvance(action);
        }
    }

    private void open(String path) {
        reader = openInputStream(path);
    }

    private String takeNextPath() {
        synchronized(paths) {
            if(paths.hasNext()) {
                return paths.next();
            }
        }
        return null;
    }

    @Override
    public Spliterator<RECORD> trySplit() {
        String nextPath = takeNextPath();
        if(nextPath != null) {
            return new JsonStreamSpliterator<METADATA,RECORD>(Long.MAX_VALUE, ADDITIONAL_CHARACTERISTICS, paths, nextPath) {
                @Override
                protected JsonStreamSupport<METADATA> openInputStream(String path) {
                    return JsonStreamSpliterator.this.openInputStream(path);
                }
                @Override
                protected RECORD parse(METADATA metaData, Map<String,Object> json) {
                    return JsonStreamSpliterator.this.parse(metaData, json);
                }
            };              
        }
        else {
            List<RECORD> records = new ArrayList<RECORD>();
            while(tryAdvance(records::add) && records.size() < MAX_BUFFER) {
                // loop
            }
            if(records.size() != 0) {
                return records.spliterator();
            }
            else {
                return null;
            }
        }
    }
}

Проблема, с которой я столкнулся, заключается в том, что, хотя Stream сначала прекрасно распараллеливается, в конечном итоге самый большой файл остается обработанным в одном потоке. Я полагаю, что проксимальная причина хорошо задокументирована: сплитератор "несбалансирован".

Более конкретно, кажется, что trySplitметод не вызывается после определенной точки Stream.forEachжизненного цикла, поэтому дополнительная логика для распределения небольших пакетов в конце trySplitредко выполняется.

Обратите внимание, что все сплитераторы, возвращаемые из trySplit, используют один и тот же pathsитератор. Я думал, что это действительно умный способ сбалансировать работу всех сплитераторов, но этого было недостаточно для достижения полного параллелизма.

Я хотел бы, чтобы параллельная обработка сначала выполнялась для файлов, а затем, когда несколько больших файлов все еще остаются сплитерирующими, я хочу распараллелить фрагменты оставшихся файлов. Это было намерение elseблока в конце trySplit.

Есть ли простой / простой / канонический способ обойти эту проблему?

Вы trySplitдолжны вывести сплиты одинакового размера, независимо от размера базовых файлов. Вы должны рассматривать все файлы как один блок и ArrayListкаждый раз заполнять сплитератор с обратной связью одинаковым количеством объектов JSON. Число объектов должно быть таким, чтобы обработка одного разбиения занимала от 1 до 10 миллисекунд: менее 1 мс, и вы начинаете приближаться к затратам на передачу пакета в рабочий поток, выше этого, и вы начинаете рисковать неравномерной загрузкой ЦП из-за задачи, которые слишком грубы.

Сплитератор не обязан сообщать оценку размера, и вы уже делаете это правильно: ваша оценка Long.MAX_VALUE- это специальное значение, означающее «неограниченный». Однако, если у вас много файлов с одним объектом JSON, что приводит к пакетам размером 1, это ухудшит вашу производительность двумя способами: издержки открытия-чтения-закрытия файла могут стать узким местом и, если вам удастся сбежать что стоимость передачи потока может быть значительной по сравнению со стоимостью обработки одного элемента, что снова вызывает узкое место.

Пять лет назад я решал похожую проблему, вы можете взглянуть на мое решение .

После долгих экспериментов я все еще не смог получить никакого дополнительного параллелизма, играя с оценками размера. По сути, любое значение, отличное от, Long.MAX_VALUEможет привести к преждевременному завершению работы сплитератора (и без какого-либо разделения), тогда как, с другой стороны, Long.MAX_VALUEоценка будет вызываться trySplitбезостановочно, пока не вернется null.

Решение, которое я нашел, состоит в том, чтобы внутренне распределить ресурсы между сплитераторами и позволить им восстановить баланс между собой.

Рабочий код:

public class AwsS3LineSpliterator<LINE> extends AbstractSpliterator<AwsS3LineInput<LINE>> {

    public final static class AwsS3LineInput<LINE> {
        final public S3ObjectSummary s3ObjectSummary;
        final public LINE lineItem;
        public AwsS3LineInput(S3ObjectSummary s3ObjectSummary, LINE lineItem) {
            this.s3ObjectSummary = s3ObjectSummary;
            this.lineItem = lineItem;
        }
    }

    private final class InputStreamHandler {
        final S3ObjectSummary file;
        final InputStream inputStream;
        InputStreamHandler(S3ObjectSummary file, InputStream is) {
            this.file = file;
            this.inputStream = is;
        }
    }

    private final Iterator<S3ObjectSummary> incomingFiles;

    private final Function<S3ObjectSummary, InputStream> fileOpener;

    private final Function<InputStream, LINE> lineReader;

    private final Deque<S3ObjectSummary> unopenedFiles;

    private final Deque<InputStreamHandler> openedFiles;

    private final Deque<AwsS3LineInput<LINE>> sharedBuffer;

    private final int maxBuffer;

    private AwsS3LineSpliterator(Iterator<S3ObjectSummary> incomingFiles, Function<S3ObjectSummary, InputStream> fileOpener,
            Function<InputStream, LINE> lineReader,
            Deque<S3ObjectSummary> unopenedFiles, Deque<InputStreamHandler> openedFiles, Deque<AwsS3LineInput<LINE>> sharedBuffer,
            int maxBuffer) {
        super(Long.MAX_VALUE, 0);
        this.incomingFiles = incomingFiles;
        this.fileOpener = fileOpener;
        this.lineReader = lineReader;
        this.unopenedFiles = unopenedFiles;
        this.openedFiles = openedFiles;
        this.sharedBuffer = sharedBuffer;
        this.maxBuffer = maxBuffer;
    }

    public AwsS3LineSpliterator(Iterator<S3ObjectSummary> incomingFiles, Function<S3ObjectSummary, InputStream> fileOpener, Function<InputStream, LINE> lineReader, int maxBuffer) {
        this(incomingFiles, fileOpener, lineReader, new ConcurrentLinkedDeque<>(), new ConcurrentLinkedDeque<>(), new ArrayDeque<>(maxBuffer), maxBuffer);
    }

    @Override
    public boolean tryAdvance(Consumer<? super AwsS3LineInput<LINE>> action) {
        AwsS3LineInput<LINE> lineInput;
        synchronized(sharedBuffer) {
            lineInput=sharedBuffer.poll();
        }
        if(lineInput != null) {
            action.accept(lineInput);
            return true;
        }
        InputStreamHandler handle = openedFiles.poll();
        if(handle == null) {
            S3ObjectSummary unopenedFile = unopenedFiles.poll();
            if(unopenedFile == null) {
                return false;
            }
            handle = new InputStreamHandler(unopenedFile, fileOpener.apply(unopenedFile));
        }
        for(int i=0; i < maxBuffer; ++i) {
            LINE line = lineReader.apply(handle.inputStream);
            if(line != null) {
                synchronized(sharedBuffer) {
                    sharedBuffer.add(new AwsS3LineInput<LINE>(handle.file, line));
                }
            }
            else {
                return tryAdvance(action);
            }
        }
        openedFiles.addFirst(handle);
        return tryAdvance(action);
    }

    @Override
    public Spliterator<AwsS3LineInput<LINE>> trySplit() {
        synchronized(incomingFiles) {
            if (incomingFiles.hasNext()) {
                unopenedFiles.add(incomingFiles.next());
                return new AwsS3LineSpliterator<LINE>(incomingFiles, fileOpener, lineReader, unopenedFiles, openedFiles, sharedBuffer, maxBuffer);
            } else {
                return null;
            }
        }
    }
}