Sammle aufeinanderfolgende Paare aus einem Stream

Bei einem Stream wie { 0, 1, 2, 3, 4 },

Wie kann ich es am elegantesten in eine gegebene Form bringen:

{ new Pair(0, 1), new Pair(1, 2), new Pair(2, 3), new Pair(3, 4) }

(vorausgesetzt natürlich, ich habe das Klassenpaar definiert)?

Bearbeiten: Hier geht es nicht ausschließlich um Ints oder primitive Streams. Die Antwort sollte für einen Stream jeglicher Art allgemein sein.

Meine StreamEx- Bibliothek, die Standard-Streams erweitert, bietet eine pairMapMethode für alle Stream-Typen. Bei primitiven Streams wird der Stream-Typ nicht geändert, es können jedoch einige Berechnungen durchgeführt werden. Am häufigsten werden Unterschiede berechnet:

int[] pairwiseDiffs = IntStreamEx.of(input).pairMap((a, b) -> (b-a)).toArray();

Für den Objektstrom können Sie einen beliebigen anderen Objekttyp erstellen. Meine Bibliothek bietet keine neuen vom Benutzer sichtbaren Datenstrukturen wie Pair(das ist der Teil des Bibliothekskonzepts). Wenn Sie jedoch eine eigene PairKlasse haben und diese verwenden möchten, können Sie Folgendes tun:

Stream<Pair> pairs = IntStreamEx.of(input).boxed().pairMap(Pair::new);

Oder wenn Sie bereits welche haben Stream:

Stream<Pair> pairs = StreamEx.of(stream).pairMap(Pair::new);

Diese Funktionalität wird mithilfe eines benutzerdefinierten Spliterators implementiert . Es hat einen recht geringen Overhead und kann gut parallelisiert werden. Natürlich funktioniert es mit jeder Stream-Quelle, nicht nur mit einer Direktzugriffsliste / einem Array wie vielen anderen Lösungen. In vielen Tests funktioniert es sehr gut. Hier ist ein JMH-Benchmark, bei dem wir alle Eingabewerte vor einem größeren Wert mit unterschiedlichen Ansätzen finden (siehe diese Frage).

Die Java 8-Streams-Bibliothek ist in erster Linie darauf ausgerichtet, Streams für die parallele Verarbeitung in kleinere Blöcke aufzuteilen. Daher sind Stateful-Pipeline-Phasen sehr begrenzt, und beispielsweise das Abrufen des Index des aktuellen Stream-Elements und der Zugriff auf benachbarte Stream-Elemente werden nicht unterstützt.

Ein typischer Weg, um diese Probleme mit einigen Einschränkungen zu lösen, besteht natürlich darin, den Stream durch Indizes zu steuern und sich darauf zu verlassen, dass die Werte in einer Datenstruktur mit wahlfreiem Zugriff wie einer ArrayList verarbeitet werden, aus der die Elemente abgerufen werden können. Wenn die Werte in arrayListwären, könnte man die Paare wie gewünscht erzeugen, indem man so etwas macht:

    IntStream.range(1, arrayList.size())
             .mapToObj(i -> new Pair(arrayList.get(i-1), arrayList.get(i)))
             .forEach(System.out::println);

Die Einschränkung besteht natürlich darin, dass die Eingabe kein unendlicher Strom sein kann. Diese Pipeline kann jedoch parallel ausgeführt werden.

Dies ist nicht elegant, es ist eine hackige Lösung, funktioniert aber für unendliche Streams

Stream<Pair> pairStream = Stream.iterate(0, (i) -> i + 1).map( // natural numbers
    new Function<Integer, Pair>() {
        Integer previous;

        @Override
        public Pair apply(Integer integer) {
            Pair pair = null;
            if (previous != null) pair = new Pair(previous, integer);
            previous = integer;
            return pair;
        }
    }).skip(1); // drop first null

Jetzt können Sie Ihren Stream auf die gewünschte Länge beschränken

pairStream.limit(1_000_000).forEach(i -> System.out.println(i));

PS Ich hoffe, es gibt eine bessere Lösung, so etwas wie Clojure(partition 2 1 stream)

Ich habe einen spliterator Wrapper implementiert , die alle nimmt nElemente Taus dem ursprünglichen spliterator und produziert List<T>:

public class ConsecutiveSpliterator<T> implements Spliterator<List<T>> {

    private final Spliterator<T> wrappedSpliterator;

    private final int n;

    private final Deque<T> deque;

    private final Consumer<T> dequeConsumer;

    public ConsecutiveSpliterator(Spliterator<T> wrappedSpliterator, int n) {
        this.wrappedSpliterator = wrappedSpliterator;
        this.n = n;
        this.deque = new ArrayDeque<>();
        this.dequeConsumer = deque::addLast;
    }

    @Override
    public boolean tryAdvance(Consumer<? super List<T>> action) {
        deque.pollFirst();
        fillDeque();
        if (deque.size() == n) {
            List<T> list = new ArrayList<>(deque);
            action.accept(list);
            return true;
        } else {
            return false;
        }
    }

    private void fillDeque() {
        while (deque.size() < n && wrappedSpliterator.tryAdvance(dequeConsumer))
            ;
    }

    @Override
    public Spliterator<List<T>> trySplit() {
        return null;
    }

    @Override
    public long estimateSize() {
        return wrappedSpliterator.estimateSize();
    }

    @Override
    public int characteristics() {
        return wrappedSpliterator.characteristics();
    }
}

Die folgende Methode kann verwendet werden, um einen aufeinanderfolgenden Stream zu erstellen:

public <E> Stream<List<E>> consecutiveStream(Stream<E> stream, int n) {
    Spliterator<E> spliterator = stream.spliterator();
    Spliterator<List<E>> wrapper = new ConsecutiveSpliterator<>(spliterator, n);
    return StreamSupport.stream(wrapper, false);
}

Beispielnutzung:

consecutiveStream(Stream.of(0, 1, 2, 3, 4, 5), 2)
    .map(list -> new Pair(list.get(0), list.get(1)))
    .forEach(System.out::println);

Sie können dies mit der Stream.reduce () -Methode tun (ich habe keine anderen Antworten mit dieser Technik gesehen).

public static <T> List<Pair<T, T>> consecutive(List<T> list) {
    List<Pair<T, T>> pairs = new LinkedList<>();
    list.stream().reduce((a, b) -> {
        pairs.add(new Pair<>(a, b));
        return b;
    });
    return pairs;
}

Sie können dies in Cyclops-React (ich trage zu dieser Bibliothek bei) mit dem Sliding-Operator tun .

  LazyFutureStream.of( 0, 1, 2, 3, 4 )
                  .sliding(2)
                  .map(Pair::new);

Oder

   ReactiveSeq.of( 0, 1, 2, 3, 4 )
                  .sliding(2)
                  .map(Pair::new);

Angenommen, der Pair-Konstruktor kann eine Sammlung mit 2 Elementen akzeptieren.

Wenn Sie um 4 gruppieren und um 2 erhöhen möchten, wird dies ebenfalls unterstützt.

     ReactiveSeq.rangeLong( 0L,Long.MAX_VALUE)
                .sliding(4,2)
                .forEach(System.out::println);

Entsprechende statische Methoden zum Erstellen einer verschiebbaren Ansicht über java.util.stream.Stream werden auch in der StreamUtils- Klasse von cyclops-Streams bereitgestellt .

       StreamUtils.sliding(Stream.of(1,2,3,4),2)
                  .map(Pair::new);

Hinweis: - Für den Single-Thread-Betrieb wäre ReactiveSeq besser geeignet. LazyFutureStream erweitert ReactiveSeq, ist jedoch hauptsächlich auf die gleichzeitige / parallele Verwendung ausgerichtet (es handelt sich um einen Stream of Futures).

LazyFutureStream erweitert ReactiveSeq, wodurch Seq vom fantastischen jOOλ (das java.util.stream.Stream erweitert) erweitert wird, sodass die von Lukas vorgestellten Lösungen auch mit beiden Stream-Typen funktionieren würden. Für alle Interessierten sind die Hauptunterschiede zwischen den Fenster- / Gleitoperatoren der offensichtliche relative Kompromiss zwischen Leistung und Komplexität und die Eignung für die Verwendung mit unendlichen Streams (das Gleiten verbraucht nicht den Stream, sondern puffert, während er fließt).

Die Protonenpack-Bibliothek bietet die Fensterfunktionalität. Bei einer Pair-Klasse und einem Stream können Sie dies folgendermaßen tun:

Stream<Integer> st = Stream.iterate(0 , x -> x + 1);
Stream<Pair<Integer, Integer>> pairs = StreamUtils.windowed(st, 2, 1)
                                                  .map(l -> new Pair<>(l.get(0), l.get(1)))
                                                  .moreStreamOps(...);

Jetzt pairsenthält der Stream:

(0, 1)
(1, 2)
(2, 3)
(3, 4)
(4, ...) and so on

Aufeinanderfolgende Paare finden

Wenn Sie bereit sind, eine Bibliothek eines Drittanbieters zu verwenden und keine Parallelität benötigen, bietet jOOλ folgende Fensterfunktionen im SQL-Stil

System.out.println(
Seq.of(0, 1, 2, 3, 4)
   .window()
   .filter(w -> w.lead().isPresent())
   .map(w -> tuple(w.value(), w.lead().get())) // alternatively, use your new Pair() class
   .toList()
);

Nachgeben

[(0, 1), (1, 2), (2, 3), (3, 4)]

Die lead()Funktion greift über das Fenster auf den nächsten Wert in Durchlaufreihenfolge zu.

Aufeinanderfolgende Tripel / Vierfache / n-Tupel finden

Eine Frage in den Kommentaren war die Frage nach einer allgemeineren Lösung, bei der nicht Paare, sondern n-Tupel (oder möglicherweise Listen) gesammelt werden sollten. Hier ist also ein alternativer Ansatz:

int n = 3;

System.out.println(
Seq.of(0, 1, 2, 3, 4)
   .window(0, n - 1)
   .filter(w -> w.count() == n)
   .map(w -> w.window().toList())
   .toList()
);

Eine Liste von Listen erstellen

[[0, 1, 2], [1, 2, 3], [2, 3, 4]]

Ohne das filter(w -> w.count() == n)wäre das Ergebnis

[[0, 1, 2], [1, 2, 3], [2, 3, 4], [3, 4], [4]]

Haftungsausschluss: Ich arbeite für die Firma hinter jOOλ

Streams.zip(..)ist in Guave für diejenigen verfügbar , die davon abhängig sind.

Beispiel:

Streams.zip(list.stream(),
            list.stream().skip(1),
            (a, b) -> System.out.printf("%s %s\n", a, b));

Wir können RxJava (sehr leistungsfähige reaktive Erweiterungsbibliothek ) verwenden.

IntStream intStream  = IntStream.iterate(1, n -> n + 1);

Observable<List<Integer>> pairObservable = Observable.from(intStream::iterator).buffer(2,1);

pairObservable.take(10).forEach(b -> {
            b.forEach(n -> System.out.println(n));
            System.out.println();
        });

Der Puffer Operator wandelt ein beobachtbares dass aussendet Artikel in einer beobachtbaren dass aussendet Sammlungen dieser Elemente gepuffert ..

Die Operation ist im Wesentlichen zustandsbehaftet, daher nicht wirklich, welche Streams gelöst werden sollen - siehe Abschnitt "Statusloses Verhalten" im Javadoc :

Der beste Ansatz besteht darin, statusbehaftete Verhaltensparameter zu vermeiden, um Operationen vollständig zu streamen

Eine Lösung besteht darin, den Status in Ihren Stream über einen externen Zähler einzuführen, obwohl dies nur mit einem sequentiellen Stream funktioniert.

public static void main(String[] args) {
    Stream<String> strings = Stream.of("a", "b", "c", "c");
    AtomicReference<String> previous = new AtomicReference<>();
    List<Pair> collect = strings.map(n -> {
                            String p = previous.getAndSet(n);
                            return p == null ? null : new Pair(p, n);
                        })
                        .filter(p -> p != null)
                        .collect(toList());
    System.out.println(collect);
}


static class Pair<T> {
    private T left, right;
    Pair(T left, T right) { this.left = left; this.right = right; }
    @Override public String toString() { return "{" + left + "," + right + '}'; }
}

In Ihrem Fall würde ich meine benutzerdefinierte IntFunction schreiben, die den zuletzt übergebenen int verfolgt, und diesen verwenden, um den ursprünglichen IntStream zuzuordnen.

import java.util.function.IntFunction;
import java.util.stream.IntStream;

public class PairFunction implements IntFunction<PairFunction.Pair> {

  public static class Pair {

    private final int first;
    private final int second;

    public Pair(int first, int second) {
      this.first = first;
      this.second = second;
    }

    @Override
    public String toString() {
      return "[" + first + "|" + second + "]";
    }
  }

  private int last;
  private boolean first = true;

  @Override
  public Pair apply(int value) {
    Pair pair = !first ? new Pair(last, value) : null;
    last = value;
    first = false;
    return pair;
  }

  public static void main(String[] args) {

    IntStream intStream = IntStream.of(0, 1, 2, 3, 4);
    final PairFunction pairFunction = new PairFunction();
    intStream.mapToObj(pairFunction)
        .filter(p -> p != null) // filter out the null
        .forEach(System.out::println); // display each Pair

  }

}

Für aufeinanderfolgende Unterschiede in der Zeit (x-Werte) ein zeitseriellen Berechnung verwende ich die stream‚s - collect(...)Methode:

final List< Long > intervals = timeSeries.data().stream()
                    .map( TimeSeries.Datum::x )
                    .collect( DifferenceCollector::new, DifferenceCollector::accept, DifferenceCollector::combine )
                    .intervals();

Wo der DifferenceCollector so etwas ist:

public class DifferenceCollector implements LongConsumer
{
    private final List< Long > intervals = new ArrayList<>();
    private Long lastTime;

    @Override
    public void accept( final long time )
    {
        if( Objects.isNull( lastTime ) )
        {
            lastTime = time;
        }
        else
        {
            intervals.add( time - lastTime );
            lastTime = time;
        }
    }

    public void combine( final DifferenceCollector other )
    {
        intervals.addAll( other.intervals );
        lastTime = other.lastTime;
    }

    public List< Long > intervals()
    {
        return intervals;
    }
}

Sie könnten dies wahrscheinlich an Ihre Bedürfnisse anpassen.

Ich habe endlich einen Weg gefunden, den Stream.reduce auszutricksen, um ordentlich mit Wertepaaren umgehen zu können. Es gibt eine Vielzahl von Anwendungsfällen, für die diese Funktion erforderlich ist, die in JDK 8 nicht selbstverständlich vorkommen :

public static int ArithGeo(int[] arr) {
    //Geometric
    List<Integer> diffList = new ArrayList<>();
    List<Integer> divList = new ArrayList<>();
    Arrays.stream(arr).reduce((left, right) -> {
        diffList.add(right-left);
        divList.add(right/left);
        return right;
    });
    //Arithmetic
    if(diffList.stream().distinct().count() == 1) {
        return 1;
    }
    //Geometric
    if(divList.stream().distinct().count() == 1) {
        return 2;
    }
    return -1;
}

Der Trick, den ich benutze, ist das Rückgaberecht; Aussage.

Eine elegante Lösung wäre die Verwendung eines Reißverschlusses . Etwas wie:

List<Integer> input = Arrays.asList(0, 1, 2, 3, 4);
Stream<Pair> pairStream = Streams.zip(input.stream(),
                                      input.stream().substream(1),
                                      (a, b) -> new Pair(a, b)
);

Dies ist ziemlich prägnant und elegant, verwendet jedoch eine Liste als Eingabe. Eine unendliche Stream-Quelle kann auf diese Weise nicht verarbeitet werden.

Ein weiteres (viel problematischeres) Problem ist, dass zip zusammen mit der gesamten Streams-Klasse kürzlich aus der API entfernt wurde. Der obige Code funktioniert nur mit b95 oder älteren Versionen. Mit dem neuesten JDK würde ich also sagen, dass es keine elegante Lösung im FP-Stil gibt, und im Moment können wir nur hoffen, dass zip auf irgendeine Weise wieder in die API eingeführt wird.

Dies ist ein interessantes Problem. Ist mein Hybridversuch unter irgendetwas gut?

public static void main(String[] args) {
    List<Integer> list = Arrays.asList(1, 2, 3);
    Iterator<Integer> first = list.iterator();
    first.next();
    if (first.hasNext())
        list.stream()
        .skip(1)
        .map(v -> new Pair(first.next(), v))
        .forEach(System.out::println);
}

Ich glaube, es eignet sich nicht für die parallele Verarbeitung und kann daher disqualifiziert werden.

Wie andere beobachtet haben, ist aufgrund der Art des Problems eine gewisse Staatlichkeit erforderlich.

Ich war mit einem ähnlichen Problem konfrontiert, bei dem ich im Wesentlichen die Oracle SQL-Funktion LEAD haben wollte. Mein Versuch, das umzusetzen, ist unten.

/**
 * Stream that pairs each element in the stream with the next subsequent element.
 * The final pair will have only the first item, the second will be null.
 */
<T> Spliterator<Pair<T>> lead(final Stream<T> stream)
{
    final Iterator<T> input = stream.sequential().iterator();

    final Iterable<Pair<T>> iterable = () ->
    {
        return new Iterator<Pair<T>>()
        {
            Optional<T> current = getOptionalNext(input);

            @Override
            public boolean hasNext()
            {
                return current.isPresent();
            }

            @Override
            public Pair<T> next()
            {
                Optional<T> next = getOptionalNext(input);
                final Pair<T> pair = next.isPresent()
                    ? new Pair(current.get(), next.get())
                    : new Pair(current.get(), null);
                current = next;

                return pair;
            }
        };
    };

    return iterable.spliterator();
}

private <T> Optional<T> getOptionalNext(final Iterator<T> iterator)
{
    return iterator.hasNext()
        ? Optional.of(iterator.next())
        : Optional.empty();
}

Sie können dies erreichen, indem Sie eine begrenzte Warteschlange verwenden, um Elemente zu speichern, die durch den Stream fließen (was auf der Idee basiert, die ich hier ausführlich beschrieben habe: Ist es möglich, das nächste Element im Stream abzurufen? )

Das folgende Beispiel definiert zuerst die Instanz der BoundedQueue-Klasse, in der Elemente gespeichert werden, die durch den Stream gehen (wenn Sie die Idee zur Erweiterung der LinkedList nicht mögen, lesen Sie den oben genannten Link für einen alternativen und allgemeineren Ansatz). Später kombinieren Sie einfach zwei aufeinanderfolgende Elemente zu einer Instanz von Pair:

public class TwoSubsequentElems {
  public static void main(String[] args) {
    List<Integer> input = new ArrayList<Integer>(asList(0, 1, 2, 3, 4));

    class BoundedQueue<T> extends LinkedList<T> {
      public BoundedQueue<T> save(T curElem) {
        if (size() == 2) { // we need to know only two subsequent elements
          pollLast(); // remove last to keep only requested number of elements
        }

        offerFirst(curElem);

        return this;
      }

      public T getPrevious() {
        return (size() < 2) ? null : getLast();
      }

      public T getCurrent() {
        return (size() == 0) ? null : getFirst();
      }
    }

    BoundedQueue<Integer> streamHistory = new BoundedQueue<Integer>();

    final List<Pair<Integer>> answer = input.stream()
      .map(i -> streamHistory.save(i))
      .filter(e -> e.getPrevious() != null)
      .map(e -> new Pair<Integer>(e.getPrevious(), e.getCurrent()))
      .collect(Collectors.toList());

    answer.forEach(System.out::println);
  }
}

Ich stimme @aepurniet zu, aber stattdessen müssen Sie mapToObj verwenden

range(0, 100).mapToObj((i) -> new Pair(i, i+1)).forEach(System.out::println);

Führen Sie eine forSchleife aus, die von 0 bis length-1zu Ihrem Stream läuft

for(int i = 0 ; i < stream.length-1 ; i++)
{
    Pair pair = new Pair(stream[i], stream[i+1]);
    // then add your pair to an array
}