Colete pares sucessivos de um riacho

Dado um fluxo como { 0, 1, 2, 3, 4 },

como posso transformá-lo da forma mais elegante em determinada forma:

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

(presumindo, é claro, que eu tenha definido um par de classe)?

Edit: Isso não é estritamente sobre ints ou fluxos primitivos. A resposta deve ser geral para um fluxo de qualquer tipo.

Minha biblioteca StreamEx que estende fluxos padrão fornece um pairMapmétodo para todos os tipos de fluxo. Para fluxos primitivos, não altera o tipo de fluxo, mas pode ser usado para fazer alguns cálculos. O uso mais comum é calcular diferenças:

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

Para fluxo de objeto, você pode criar qualquer outro tipo de objeto. Minha biblioteca não fornece nenhuma nova estrutura de dados visível ao usuário Pair(essa é a parte do conceito de biblioteca). No entanto, se você tem sua própria Pairaula e deseja usá-la, pode fazer o seguinte:

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

Ou se você já tem algum Stream:

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

Esta funcionalidade é implementada usando um divisor personalizado . Ele tem um overhead bastante baixo e pode paralelizar bem. É claro que funciona com qualquer fonte de fluxo, não apenas lista / array de acesso aleatório como muitas outras soluções. Em muitos testes, ele tem um desempenho muito bom. Aqui está um benchmark JMH onde encontramos todos os valores de entrada que precedem um valor maior usando diferentes abordagens (consulte esta questão).

A biblioteca de streams Java 8 é principalmente voltada para a divisão de streams em pedaços menores para processamento paralelo, portanto, estágios de pipeline com estado são bastante limitados e tarefas como obter o índice do elemento stream atual e acessar elementos de stream adjacentes não são suportadas.

Uma maneira típica de resolver esses problemas, com algumas limitações, é claro, é conduzir o fluxo por índices e confiar em ter os valores sendo processados ​​em alguma estrutura de dados de acesso aleatório, como um ArrayList do qual os elementos podem ser recuperados. Se os valores estivessem em arrayList, seria possível gerar os pares conforme solicitado fazendo algo assim:

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

Claro que a limitação é que a entrada não pode ser um fluxo infinito. No entanto, esse pipeline pode ser executado em paralelo.

Isso não é elegante, é uma solução hackeada, mas funciona para fluxos infinitos

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

Agora você pode limitar seu stream ao comprimento que desejar

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

PS Espero que haja uma solução melhor, algo como clojure(partition 2 1 stream)

Implementei um wrapper divisor que pega todos os nelementos Tdo divisor original e produz 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();
    }
}

O método a seguir pode ser usado para criar um fluxo consecutivo:

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

Uso de amostra:

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

Você pode fazer isso com o método Stream.reduce () (não vi nenhuma outra resposta usando essa técnica).

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

Você pode fazer isso em cyclops-react (eu contribuo para esta biblioteca), usando o operador deslizante.

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

Ou

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

Supondo que o construtor de par pode aceitar uma coleção com 2 elementos.

Se você quiser agrupar por 4 e incrementar por 2, isso também é compatível.

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

Métodos estáticos equivalentes para criar uma visualização deslizante sobre java.util.stream.Stream também são fornecidos na classe Cyclops-streams StreamUtils .

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

Nota: - para operação de thread único, o ReactiveSeq seria mais apropriado. LazyFutureStream estende ReactiveSeq, mas é principalmente voltado para uso simultâneo / paralelo (é um Fluxo de Futuros).

LazyFutureStream estende ReactiveSeq, que estende Seq do incrível jOOλ (que estende java.util.stream.Stream), então as soluções apresentadas por Lukas também funcionariam com qualquer tipo de fluxo. Para qualquer pessoa interessada, as principais diferenças entre os operadores de janela / deslizante são a relação óbvia de troca de poder / complexidade e adequação para uso com fluxos infinitos (deslizar não consome o fluxo, mas armazena enquanto ele flui).

A biblioteca proton-pack fornece a funcionalidade em janela. Dada uma classe Pair e um Stream, você pode fazer assim:

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

Agora o pairsstream contém:

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

Encontrando pares sucessivos

Se você está disposto a usar uma biblioteca de terceiros e não precisa de paralelismo, então jOOλ oferece funções de janela no estilo SQL como segue

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

Produzindo

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

A lead()função acessa o próximo valor na ordem de passagem da janela.

Encontrar sucessivos triplos / quádruplos / n-tuplas

Uma pergunta nos comentários estava pedindo uma solução mais geral, onde não pares, mas n-tuplas (ou possivelmente listas) deveriam ser coletados. Aqui está, portanto, uma abordagem alternativa:

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

Produzindo uma lista de listas

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

Sem o filter(w -> w.count() == n), o resultado seria

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

Isenção de responsabilidade: eu trabalho para a empresa por trás do jOOλ

Streams.zip(..)está disponível na Goiaba , para quem depende dela.

Exemplo:

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

Podemos usar RxJava ( biblioteca de extensão reativa muito poderosa )

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

O operador buffer transforma um Observable que emite itens em um Observable que emite coleções armazenadas em buffer desses itens.

A operação é essencialmente com estado, então não é realmente o que os streams pretendem resolver - consulte a seção "Comportamentos sem estado" no javadoc :

A melhor abordagem é evitar parâmetros comportamentais com estado para transmitir operações inteiramente

Uma solução aqui é introduzir o estado em seu stream por meio de um contador externo, embora funcione apenas com um stream sequencial.

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

No seu caso, eu escreveria meu IntFunction personalizado que mantém o controle do último int passado e o usaria para mapear o IntStream original.

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

  }

}

Para calcular diferenças sucessivas no tempo (x-valores) de uma série de tempo, eu utilizar o stream's collect(...)método:

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

Onde o DifferenceCollector é algo assim:

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

Você provavelmente pode modificar isso para atender às suas necessidades.

Finalmente descobri uma maneira de enganar o Stream.reduce para poder lidar ordenadamente com pares de valores; há uma infinidade de casos de uso que requerem esse recurso que não aparece naturalmente no 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;
}

O truque que uso é o direito de retorno; declaração.

Uma solução elegante seria usar zip . Algo como:

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

Isso é muito conciso e elegante, no entanto, usa uma lista como entrada. Uma fonte de fluxo infinito não pode ser processada dessa maneira.

Outro problema (muito mais problemático) é que o zip junto com a classe Streams inteira foi recentemente removido da API. O código acima funciona apenas com b95 ou versões anteriores. Portanto, com o JDK mais recente, eu diria que não há uma solução elegante no estilo FP e, no momento, podemos apenas esperar que, de alguma forma, o zip seja reintroduzido na API.

Este é um problema interessante. Minha tentativa híbrida é inferior a alguma boa?

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

Eu acredito que não se presta ao processamento paralelo e, portanto, pode ser desqualificado.

Como outros observaram, devido à natureza do problema, existe alguma condição necessária.

Eu me deparei com um problema semelhante, no qual eu queria o que era essencialmente a função LEAD do Oracle SQL. Minha tentativa de implementar isso está abaixo.

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

Você pode conseguir isso usando uma fila limitada para armazenar os elementos que fluem pelo fluxo (que se baseia na ideia que descrevi em detalhes aqui: É possível obter o próximo elemento no fluxo? )

O exemplo abaixo define primeiro a instância da classe BoundedQueue que armazenará elementos que passam pelo fluxo (se você não gosta da ideia de estender LinkedList, consulte o link mencionado acima para uma abordagem alternativa e mais genérica). Mais tarde, você apenas combina dois elementos subsequentes em uma instância de 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);
  }
}

Eu concordo com @aepurniet, mas em vez disso, você deve usar mapToObj

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

Execute um forloop que vai de 0 a length-1de seu stream

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
}