Zbieraj kolejne pary ze strumienia

Biorąc pod uwagę strumień, taki jak { 0, 1, 2, 3, 4 },

jak mogę najbardziej elegancko nadać mu daną formę:

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

(zakładając oczywiście, że zdefiniowałem parę klas)?

Edycja: nie dotyczy to wyłącznie ints ani strumieni pierwotnych. Odpowiedź powinna być ogólna dla dowolnego typu strumienia.

Moja biblioteka StreamEx, która rozszerza standardowe strumienie, zapewnia pairMapmetodę dla wszystkich typów strumieni. W przypadku strumieni pierwotnych nie zmienia typu strumienia, ale może służyć do wykonywania pewnych obliczeń. Najczęstszym zastosowaniem jest obliczanie różnic:

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

Dla strumienia obiektów można utworzyć dowolny inny typ obiektu. Moja biblioteka nie dostarcza żadnych nowych struktur danych widocznych dla użytkownika, takich jak Pair(to jest część koncepcji biblioteki). Jeśli jednak masz własną Pairklasę i chcesz z niej korzystać, możesz wykonać następujące czynności:

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

Lub jeśli już masz Stream:

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

Ta funkcja jest realizowana przy użyciu niestandardowego rozdzielacza . Ma dość niski narzut i może ładnie pracować równolegle. Oczywiście działa z każdym źródłem strumienia, a nie tylko z listami / tablicami o swobodnym dostępie, jak wiele innych rozwiązań. W wielu testach spisuje się naprawdę dobrze. Oto test porównawczy JMH, w którym wszystkie wartości wejściowe poprzedzające większą wartość przy użyciu różnych podejść (zobacz to pytanie).

Biblioteka strumieni Java 8 jest głównie nastawiona na dzielenie strumieni na mniejsze fragmenty w celu przetwarzania równoległego, więc etapy stanowe potoku są dość ograniczone, a wykonywanie takich czynności, jak uzyskiwanie indeksu bieżącego elementu strumienia i uzyskiwanie dostępu do sąsiednich elementów strumienia, nie jest obsługiwane.

Typowym sposobem rozwiązania tych problemów, oczywiście z pewnymi ograniczeniami, jest kierowanie strumienia za pomocą indeksów i poleganie na przetwarzaniu wartości w jakiejś strukturze danych o swobodnym dostępie, takiej jak ArrayList, z której można pobrać elementy. Gdyby wartości były w arrayList, można by wygenerować pary zgodnie z żądaniem, wykonując coś takiego:

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

Oczywiście ograniczeniem jest to, że wejście nie może być nieskończonym strumieniem. Ten potok może jednak przebiegać równolegle.

To nie jest eleganckie, to hakerskie rozwiązanie, ale działa dla nieskończonych strumieni

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

Teraz możesz ograniczyć swój strumień do żądanej długości

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

PS Mam nadzieję, że jest lepsze rozwiązanie, coś w rodzaju clojure(partition 2 1 stream)

Zaimplementowałem opakowanie spliteratora, które pobiera wszystkie nelementy Tz oryginalnego spliteratora i produkuje 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();
    }
}

Do stworzenia kolejnego strumienia można użyć następującej metody:

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

Przykładowe użycie:

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

Możesz to zrobić za pomocą metody Stream.reduce () (nie widziałem żadnych innych odpowiedzi wykorzystujących tę technikę).

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

Możesz to zrobić w Cyclops-React (współtworzę tę bibliotekę), używając przesuwanego operatora.

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

Lub

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

Zakładając, że konstruktor Pair może zaakceptować Collection z 2 elementami.

Jeśli chcesz pogrupować o 4 i zwiększyć o 2, to również jest obsługiwane.

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

Równoważne metody statyczne do tworzenia przesuwanego widoku w java.util.stream.Stream są również dostępne w klasie StreamUtils cyclops-streams .

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

Uwaga: - dla operacji jednowątkowych bardziej odpowiednie byłoby ReactiveSeq. LazyFutureStream rozszerza ReactiveSeq, ale jest przede wszystkim nastawiony na jednoczesne / równoległe użycie (jest to strumień kontraktów futures).

LazyFutureStream rozszerza ReactiveSeq, który rozszerza Seq z niesamowitego jOOλ (który rozszerza java.util.stream.Stream), więc rozwiązania prezentowane przez Lukasa działałyby również z każdym typem Stream. Dla każdego zainteresowanego podstawowymi różnicami między operatorami okien / przesuwnych są oczywisty względny kompromis między mocą a złożonością oraz przydatność do użycia z nieskończonymi strumieniami (przesuwanie nie zużywa strumienia, ale buforuje podczas jego przepływu).

Biblioteka proton-pack zapewnia okienkiem functionnality. Biorąc pod uwagę klasę pary i strumień, możesz to zrobić w następujący sposób:

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(...);

Teraz pairsstrumień zawiera:

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

Znajdowanie kolejnych par

Jeśli chcesz skorzystać z biblioteki innej firmy i nie potrzebujesz równoległości, jOOλ oferuje następujące funkcje okna w stylu SQL

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()
);

Wydajność

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

lead()Funkcja uzyskuje dostęp do kolejnej wartości w porządku przemierzania z okna.

Znajdowanie kolejnych potrójnych / poczwórnych / n-krotek

Pytanie w komentarzach dotyczyło bardziej ogólnego rozwiązania, w którym nie należy gromadzić par, ale n-krotek (lub ewentualnie list). Oto więc alternatywne podejście:

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()
);

Udostępnianie listy list

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

Bez tego filter(w -> w.count() == n)wynik byłby

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

Zastrzeżenie: pracuję dla firmy stojącej za jOOλ

Streams.zip(..)jest dostępny w guawie , dla tych, którzy na nim polegają.

Przykład:

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

Możemy użyć RxJava (bardzo potężna reaktywna biblioteka rozszerzeń )

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

Bufor operator zmienia zauważalnemu emitujący elementów do możliwego do zaobserwowania emitujący buforowane zbiory tych elementów ..

Operacja jest zasadniczo stanowa, więc nie do końca, jakie strumienie mają rozwiązać - zobacz sekcję „Zachowania bezstanowe” w javadoc :

Najlepszym podejściem jest unikanie stanowych parametrów behawioralnych w celu całkowitego przesyłania strumieniowego operacji

Jednym z rozwiązań jest wprowadzenie stanu do strumienia przez zewnętrzny licznik, chociaż będzie to działać tylko ze strumieniem sekwencyjnym.

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

W twoim przypadku napisałbym moją niestandardową funkcję IntFunction, która śledzi ostatni przekazany int i użyję jej do zmapowania oryginalnego IntStream.

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

  }

}

Do obliczenia kolejnych różnic w czasie (wartości x) z szeregów czasowych, używać stream„S collect(...)sposób:

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

Gdzie DifferenceCollector wygląda mniej więcej tak:

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

Prawdopodobnie możesz to zmodyfikować, aby dopasować je do swoich potrzeb.

W końcu znalazłem sposób na oszukanie Stream.reduce, aby móc starannie radzić sobie z parami wartości; istnieje wiele przypadków użycia, które wymagają tej funkcji, która nie pojawia się naturalnie w JDK 8:

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

Sztuczka, której używam, to właściwy powrót; komunikat.

Eleganckim rozwiązaniem byłoby użycie zamka błyskawicznego . Coś jak:

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

Jest to dość zwięzłe i eleganckie, jednak używa listy jako danych wejściowych. Nieskończone źródło strumienia nie może być przetwarzane w ten sposób.

Innym (dużo bardziej kłopotliwym) problemem jest to, że zip wraz z całą klasą Streams został ostatnio usunięty z API. Powyższy kod działa tylko z wersjami b95 lub starszymi. Więc w przypadku najnowszego JDK powiedziałbym, że nie ma eleganckiego rozwiązania w stylu FP i teraz możemy mieć tylko nadzieję, że w jakiś sposób zip zostanie ponownie wprowadzony do API.

To ciekawy problem. Czy moja próba hybrydy poniżej jest dobra?

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

Uważam, że nie nadaje się do przetwarzania równoległego, a zatem może zostać zdyskwalifikowany.

Jak zauważyli inni, ze względu na naturę problemu wymagana jest pewna stanowość.

Miałem podobny problem, w którym chciałem, aby w istocie była to funkcja Oracle SQL LEAD. Moja próba wdrożenia tego znajduje się poniżej.

/**
 * 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();
}

Możesz to osiągnąć, używając ograniczonej kolejki do przechowywania elementów, które przepływają przez strumień (co opiera się na pomyśle, który szczegółowo opisałem tutaj: Czy można pobrać następny element w Stream? )

Poniższy przykład najpierw definiuje instancję klasy BoundedQueue, która będzie przechowywać elementy przechodzące przez strumień (jeśli nie podoba ci się pomysł rozszerzenia LinkedList, odwołaj się do wspomnianego powyżej linku, aby uzyskać alternatywne i bardziej ogólne podejście). Później wystarczy połączyć dwa kolejne elementy w instancję 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);
  }
}

Zgadzam się z @aepurniet, ale zamiast map musisz użyć mapToObj

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

Uruchom forpętlę, która biegnie od 0 do length-1strumienia

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
}