Bisakah Anda menyeimbangkan kembali Spliterator yang tidak seimbang dengan ukuran yang tidak diketahui?

Saya ingin menggunakan Streamuntuk memparalelkan pemrosesan set heterogen file JSON yang disimpan dari nomor yang tidak dikenal (jumlah file tidak diketahui dimuka). Ukuran file dapat sangat bervariasi, mulai dari 1 catatan JSON per file hingga 100.000 catatan di beberapa file lainnya. Sebuah JSON catatan dalam hal ini berarti mandiri JSON objek direpresentasikan sebagai satu baris dalam file.

Saya benar-benar ingin menggunakan Streaming untuk ini dan jadi saya menerapkan ini 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;
            }
        }
    }
}

Masalah yang saya alami adalah ketika Stream memparalelkan dengan indah pada awalnya, akhirnya file terbesar dibiarkan diproses dalam satu utas. Saya percaya penyebab proksimal didokumentasikan dengan baik: spliterator "tidak seimbang".

Lebih konkret, tampak bahwa trySplitmetode ini tidak dipanggil setelah titik tertentu dalam Stream.forEachsiklus hidup, sehingga logika ekstra untuk mendistribusikan batch kecil pada akhir trySplitjarang dieksekusi.

Perhatikan bagaimana semua spliterator yang dikembalikan dari trySplit berbagi pathsiterator yang sama . Saya pikir ini adalah cara yang sangat pintar untuk menyeimbangkan pekerjaan di semua pembagi, tapi itu belum cukup untuk mencapai paralelisme penuh.

Saya ingin pemrosesan paralel untuk melanjutkan pertama di file, dan kemudian ketika beberapa file besar masih dibiarkan membelah, saya ingin memparalelkan antar potongan file yang tersisa. Itulah maksud dari elseblok di akhir trySplit.

Apakah ada cara yang mudah / sederhana / kanonik untuk mengatasi masalah ini?

trySplitKeluaran Anda harus berukuran sama, terlepas dari ukuran file yang mendasarinya. Anda harus memperlakukan semua file sebagai satu unit dan mengisi ArrayListspliterator yang dikembalikan dengan jumlah objek JSON yang sama setiap kali. Jumlah objek harus sedemikian rupa sehingga pemrosesan satu split membutuhkan waktu antara 1 dan 10 milidetik: lebih rendah dari 1 ms dan Anda mulai mendekati biaya menyerahkan batch ke thread pekerja, lebih tinggi dari itu dan Anda mulai berisiko beban CPU yang tidak merata karena tugas-tugas yang terlalu kasar.

Pemisah tidak wajib melaporkan perkiraan ukuran, dan Anda sudah melakukan ini dengan benar: perkiraan Anda adalah Long.MAX_VALUE, yang merupakan nilai khusus yang berarti "tidak terikat". Namun, jika Anda memiliki banyak file dengan objek JSON tunggal, menghasilkan kumpulan ukuran 1, ini akan merusak kinerja Anda dalam dua cara: overhead pembukaan-membaca-menutup file mungkin menjadi hambatan dan, jika Anda berhasil melarikan diri bahwa, biaya handoff benang mungkin signifikan dibandingkan dengan biaya pemrosesan satu item, sekali lagi menyebabkan kemacetan.

Lima tahun yang lalu saya memecahkan masalah yang sama, Anda dapat melihat solusi saya .

Setelah banyak percobaan, saya masih tidak bisa mendapatkan paralelisme tambahan dengan bermain dengan perkiraan ukuran. Pada dasarnya, nilai apa pun selain Long.MAX_VALUEakan cenderung menyebabkan spliterator berakhir terlalu dini (dan tanpa pemisahan apa pun), sementara di sisi lain Long.MAX_VALUEperkiraan akan menyebabkan trySplitdipanggil tanpa henti hingga kembali null.

Solusi yang saya temukan adalah berbagi sumber daya secara internal di antara para pembagi dan membiarkan mereka menyeimbangkan kembali di antara mereka sendiri.

Kode kerja:

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;
            }
        }
    }
}