So teilen Sie Iterable in Chunks mit Lambda-Ausdrücken

Question

Ich habe Iterable<CSVRecord> = CSVFormat.RFC4180.withFirstRecordAsHeader().parse(in) (Apache Commons) Datensätze, die> 10.000.000.000 Zeilen ist. Zuvor hatte ich eine for-Schleife mit Zähler und nach jeder x-Zeile verarbeitete ich Daten. Jetzt versuche ich einen ähnlichen Effekt mit Java 8 Lambda-Ausdrücken zu erreichen.

Bisher komme ich mit diesem, aber es läuft nicht genügend Arbeitsspeicher, da ich nicht in der Lage bin richtigen Weg zu finden, wie das spalten in subList

Iterable<List<?>> params = new ArrayList<>(StreamSupport 
       .stream(records.spliterator(), true) 
       .map(r -> Arrays.asList(
         r.get("name"), 
         r.get("surname"), 
         r.get("something"), 
       )) 
       .collect(Collectors.toList()).subList(0, 20000)); 

subList am Ende nicht funktioniert :(

ich brauche nur einen Beweis Konzept wie Iterable aufzuspalten - zB der richtigen Ort, wo subList

Answer 1

ich bin nicht sicher zu stellen, wenn Sie es mit einem einzigen Lambda Expressen tun können, aber Sie können .skip() und 01 verwenden?:

int maxSize = 20000; 
int currentOffset = 0; // Increment by maxSize each iteration  
Iterable<List<?>> params = new ArrayList<>(StreamSupport 
          .stream(records.spliterator(), true) 
          .skip(currentOffset) 
          .limit(maxSize) 
          .map(r -> Arrays.asList(
            r.get("name"), 
            r.get("surname"), 
            r.get("something"), 
          )) 
          .collect(Collectors.toList()) 

Answer 2

Ich denke, die beste Lösung für Ihren Fall ist die Datentransformationsstufe und Chunking zu trennen. Für Datentransformation (Datensätze -> Arrays) können Sie Streams oder Parrallstream verwenden. Sie scheinen hier. Aber Chunking ist kein gutes Szenario für die Streams. Hören Sie, es wäre besser, einfache Schleife oder einige Bibliothek API zu verwenden (wie RC empfohlen).

Answer 3

Dies beantwortet nicht die Frage des Chunking, aber ... Denken Sie daran, mehr Speicher zu kaufen. Wenn dies die typische Größe des Problemsatzes ist. Der Arbeitsspeicher kann billiger sein als die Kosten für die zusätzliche Denkzeit, um ständig speichereffiziente Programme zu schreiben.

Answer 4

Hallo ich bin mir nicht sicher, ob es gut aussieht, aber das ist eine andere Möglichkeit, Dinge zu handhaben.

//that can be CsvParser not List 
List<Integer> collection = Arrays.asList(1, 2, 4, 5, 2, 1, 2, 4, 5); 

int limit = 2; 
int size = collection.size(); 

long maxPartIndex = (long) Math.ceil((double) size/ limit); 

LongStream.range(0, maxPartIndex) 
    .mapToObj(partIndex -> getPart(collection.spliterator(), partIndex, limit)) 
    .forEach(System.out::println); 

....

private static <T> List<T> getPart(Spliterator<T> stream, long index, long size) { 
    return StreamSupport.stream(stream, false) 
    .skip(index * size) 
    .limit(size) 
    .collect(Collectors.toList()); 
} 

output:

(1, 2) (4, 5) (2, 1) (2, 4) (5)