Collectez des paires successives à partir d'un flux

Étant donné un flux tel que { 0, 1, 2, 3, 4 },

comment puis-je le transformer le plus élégamment en une forme donnée:

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

(en supposant, bien sûr, que j'ai défini la classe Pair)?

Edit: Il ne s'agit pas strictement d'ints ou de flux primitifs. La réponse doit être générale pour un flux de tout type.

Ma bibliothèque StreamEx qui étend les flux standard fournit une pairMapméthode pour tous les types de flux. Pour les flux primitifs, cela ne change pas le type de flux, mais peut être utilisé pour effectuer certains calculs. L'utilisation la plus courante consiste à calculer les différences:

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

Pour le flux d'objets, vous pouvez créer tout autre type d'objet. Ma bibliothèque ne fournit aucune nouvelle structure de données visible par l'utilisateur comme Pair(c'est la partie du concept de bibliothèque). Cependant, si vous avez votre propre Pairclasse et que vous souhaitez l'utiliser, vous pouvez effectuer les opérations suivantes:

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

Ou si vous en avez déjà Stream:

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

Cette fonctionnalité est implémentée à l'aide d' un séparateur personnalisé . Il a une surcharge assez faible et peut bien se paralléliser. Bien sûr, cela fonctionne avec n'importe quelle source de flux, pas seulement une liste / un tableau d'accès aléatoire comme beaucoup d'autres solutions. Dans de nombreux tests, il fonctionne très bien. Voici un benchmark JMH où nous trouvons toutes les valeurs d'entrée précédant une valeur plus grande en utilisant différentes approches (voir cette question).

La bibliothèque de flux Java 8 est principalement destinée à diviser les flux en plus petits morceaux pour un traitement parallèle, de sorte que les étapes de pipeline avec état sont assez limitées, et les actions telles que l'obtention de l'index de l'élément de flux actuel et l'accès aux éléments de flux adjacents ne sont pas prises en charge.

Un moyen typique de résoudre ces problèmes, avec certaines limitations, bien sûr, consiste à piloter le flux par des index et à compter sur le traitement des valeurs dans une structure de données à accès aléatoire comme une ArrayList à partir de laquelle les éléments peuvent être récupérés. Si les valeurs étaient dans arrayList, on pourrait générer les paires comme demandé en faisant quelque chose comme ceci:

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

Bien sûr, la limitation est que l'entrée ne peut pas être un flux infini. Ce pipeline peut cependant être exécuté en parallèle.

Ce n'est pas élégant, c'est une solution hackish, mais fonctionne pour des flux infinis

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

Vous pouvez désormais limiter votre diffusion à la longueur souhaitée

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

PS j'espère qu'il y a une meilleure solution, quelque chose comme clojure(partition 2 1 stream)

J'ai implémenté un wrapper de séparateur qui prend tous les néléments Tdu séparateur d'origine et produit 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();
    }
}

La méthode suivante peut être utilisée pour créer un flux consécutif:

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

Exemple d'utilisation:

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

Vous pouvez le faire avec la méthode Stream.reduce () (je n'ai vu aucune autre réponse utilisant cette technique).

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

Vous pouvez le faire dans cyclops-react (je contribue à cette bibliothèque), en utilisant l'opérateur glissant.

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

En supposant que le constructeur Pair peut accepter une Collection avec 2 éléments.

Si vous souhaitez grouper par 4 et incrémenter de 2, cela est également pris en charge.

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

Des méthodes statiques équivalentes pour créer une vue glissante sur java.util.stream.Stream sont également fournies dans la classe Cyclops -streams StreamUtils .

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

Remarque: - pour une opération monothread, ReactiveSeq serait plus approprié. LazyFutureStream étend ReactiveSeq mais est principalement conçu pour une utilisation simultanée / parallèle (c'est un Stream of Futures).

LazyFutureStream étend ReactiveSeq qui étend Seq du génial jOOλ (qui étend java.util.stream.Stream), de sorte que les solutions présentées par Lukas fonctionneraient également avec l'un ou l'autre type de Stream. Pour toute personne intéressée, les principales différences entre les opérateurs fenêtre / glissement sont le compromis évident puissance / complexité relative et l'aptitude à être utilisé avec des flux infinis (le glissement ne consomme pas le flux, mais des tampons lorsqu'il s'écoule).

La bibliothèque proton-pack fournit la fonctionnalité fenêtrée. Étant donné une classe Pair et un Stream, vous pouvez le faire comme ceci:

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

Maintenant, le pairsflux contient:

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

Trouver des paires successives

Si vous êtes prêt à utiliser une bibliothèque tierce et n'avez pas besoin de parallélisme, alors jOOλ propose des fonctions de fenêtre de style SQL comme suit

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

Céder

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

La lead()fonction accède à la valeur suivante dans l'ordre de parcours à partir de la fenêtre.

Recherche de triples / quadruples / n-tuples successifs

Une question dans les commentaires demandait une solution plus générale, où non pas des paires mais des n-uplets (ou éventuellement des listes) devraient être collectés. Voici donc une approche alternative:

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

Produire une liste de listes

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

Sans le filter(w -> w.count() == n), le résultat serait

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

Clause de non-responsabilité: je travaille pour l'entreprise derrière jOOλ

Streams.zip(..)est disponible à Guava , pour ceux qui en dépendent.

Exemple:

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

Nous pouvons utiliser RxJava ( bibliothèque d' extension réactive très puissante )

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

L' opérateur de tampon transforme un Observable qui émet des éléments en un Observable qui émet des collections tamponnées de ces éléments.

L'opération est essentiellement avec état donc pas vraiment ce que les flux sont censés résoudre - voir la section "Comportements sans état" dans le javadoc :

La meilleure approche consiste à éviter les paramètres comportementaux avec état pour diffuser entièrement les opérations

Une solution ici est d'introduire un état dans votre flux via un compteur externe, bien que cela ne fonctionnera qu'avec un flux séquentiel.

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

Dans votre cas, j'écrirais mon IntFunction personnalisé qui garde la trace du dernier int passé et l'utiliser pour mapper l'IntStream d'origine.

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

  }

}

Pour le calcul de différences successives dans le temps (valeurs x) d'une série chronologique, j'utilise la streams » collect(...)procédé:

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

Où le DifferenceCollector est quelque chose comme ceci:

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

Vous pouvez probablement le modifier en fonction de vos besoins.

J'ai finalement trouvé un moyen de tromper le Stream.reduce pour pouvoir gérer proprement des paires de valeurs; il existe une multitude de cas d'utilisation qui nécessitent cette fonctionnalité qui n'apparaît pas naturellement dans 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;
}

L'astuce que j'utilise est le bon retour; déclaration.

Une solution élégante serait d'utiliser zip . Quelque chose comme:

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

C'est assez concis et élégant, mais il utilise une liste comme entrée. Une source de flux infinie ne peut pas être traitée de cette manière.

Un autre problème (beaucoup plus gênant) est que le zip avec toute la classe Streams a été récemment supprimé de l'API. Le code ci-dessus ne fonctionne qu'avec b95 ou versions antérieures. Donc, avec le dernier JDK, je dirais qu'il n'y a pas de solution élégante de style FP et pour le moment, nous pouvons juste espérer que d'une manière ou d'une autre, zip sera réintroduit dans l'API.

C'est un problème intéressant. Ma tentative hybride ci-dessous est-elle bonne?

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

Je pense qu'il ne se prête pas à un traitement parallèle et peut donc être disqualifié.

Comme d'autres l'ont observé, il y a, en raison de la nature du problème, un certain état requis.

J'ai été confronté à un problème similaire, dans lequel je voulais ce qui était essentiellement la fonction Oracle SQL LEAD. Ma tentative de mise en œuvre est ci-dessous.

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

Vous pouvez y parvenir en utilisant une file d'attente limitée pour stocker les éléments qui circulent dans le flux (qui repose sur l'idée que j'ai décrite en détail ici: est-il possible d'obtenir l'élément suivant dans le flux? )

L'exemple Belows définit d'abord l'instance de la classe BoundedQueue qui stockera les éléments passant par le flux (si vous n'aimez pas l'idée d'étendre la LinkedList, reportez-vous au lien mentionné ci-dessus pour une approche alternative et plus générique). Plus tard, vous combinez simplement deux éléments suivants dans l'instance 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);
  }
}

Je suis d'accord avec @aepurniet mais à la place, vous devez utiliser mapToObj

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

Exécutez une forboucle qui va de 0 à length-1de votre flux

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
}