lambda - java stream map

Zipping streams usando JDK8 con lambda(java.util.stream.Streams.zip)

Question

En JDK 8 con lambda b93 había una clase java.util.stream.Streams.zip en b93 que se podía usar para comprimir las transmisiones (esto se ilustra en el tutorial Exploring Java8 Lambdas. Parte 1 de Dhananjay Nene ). Esta función :

Crea un flujo combinado perezoso y secuencial cuyos elementos son el resultado de combinar los elementos de dos flujos.

Sin embargo, en b98 esto ha desaparecido. De hecho, la clase Streams no es accesible en java.util.stream en b98 .

¿Se ha movido esta funcionalidad? En caso afirmativo, ¿cómo puedo comprimir las transmisiones de forma concisa utilizando b98?

La aplicación que tengo en mente es en esta implementación java de Shen , donde reemplacé la funcionalidad zip en el

• static <T> boolean every(Collection<T> c1, Collection<T> c2, BiPredicate<T, T> pred)
• static <T> T find(Collection<T> c1, Collection<T> c2, BiPredicate<T, T> pred)

funciona con un código bastante detallado (que no usa la funcionalidad de b98).

Answer 1

Como no puedo concebir el uso de compresión en colecciones distintas a las indexadas (listas) y soy un gran admirador de la simplicidad, esta sería mi solución:

<A,B,C>  Stream<C> zipped(List<A> lista, List<B> listb, BiFunction<A,B,C> zipper){
     int shortestLength = Math.min(lista.size(),listb.size());
     return IntStream.range(0,shortestLength).mapToObject( i -> {
          return zipper.apply(lista.get(i), listb.get(i));
     });        
}

Answer 2

Comprimir dos flujos usando JDK8 con lambda ( gist ).

public static <A, B, C> Stream<C> zip(Stream<A> streamA, Stream<B> streamB, BiFunction<A, B, C> zipper) {
    final Iterator<A> iteratorA = streamA.iterator();
    final Iterator<B> iteratorB = streamB.iterator();
    final Iterator<C> iteratorC = new Iterator<C>() {
        @Override
        public boolean hasNext() {
            return iteratorA.hasNext() && iteratorB.hasNext();
        }

        @Override
        public C next() {
            return zipper.apply(iteratorA.next(), iteratorB.next());
        }
    };
    final boolean parallel = streamA.isParallel() || streamB.isParallel();
    return iteratorToFiniteStream(iteratorC, parallel);
}

public static <T> Stream<T> iteratorToFiniteStream(Iterator<T> iterator, boolean parallel) {
    final Iterable<T> iterable = () -> iterator;
    return StreamSupport.stream(iterable.spliterator(), parallel);
}

Answer 3

El cyclops-react AOL, al cual contribuyo, también proporciona funcionalidad de compresión, tanto a través de una implementación extendida de Stream , que también implementa la interfaz de flujos reactivos ReactiveSeq, y a través de StreamUtils que ofrece gran parte de la misma funcionalidad a través de métodos estáticos a corrientes Java estándar.

 List<Tuple2<Integer,Integer>> list =  ReactiveSeq.of(1,2,3,4,5,6)
                                                  .zip(Stream.of(100,200,300,400));


  List<Tuple2<Integer,Integer>> list = StreamUtils.zip(Stream.of(1,2,3,4,5,6),
                                                  Stream.of(100,200,300,400));

También ofrece un zipping más generalizado basado en la aplicación. P.ej

   ReactiveSeq.of("a","b","c")
              .ap3(this::concat)
              .ap(of("1","2","3"))
              .ap(of(".","?","!"))
              .toList();

   //List("a1.","b2?","c3!");

   private String concat(String a, String b, String c){
    return a+b+c;
   }

E incluso la posibilidad de emparejar cada elemento en una secuencia con cada elemento en otra

   ReactiveSeq.of("a","b","c")
              .forEach2(str->Stream.of(str+"!","2"), a->b->a+"_"+b);

   //ReactiveSeq("a_a!","a_2","b_b!","b_2","c_c!","c2")

Answer 4

Esto es genial. Tuve que comprimir dos flujos en un mapa con una secuencia siendo la clave y otra siendo el valor

Stream<String> streamA = Stream.of("A", "B", "C");
Stream<String> streamB  = Stream.of("Apple", "Banana", "Carrot", "Doughnut");    
final Stream<Map.Entry<String, String>> s = StreamUtils.zip(streamA,
                    streamB,
                    (a, b) -> {
                        final Map.Entry<String, String> entry = new AbstractMap.SimpleEntry<String, String>(a, b);
                        return entry;
                    });

System.out.println(s.collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue())));

Salida: {A = Apple, B = Plátano, C = Zanahoria}

Answer 5

Humildemente sugiero esta implementación. La secuencia resultante se trunca al más corto de los dos flujos de entrada.

public static <L, R, T> Stream<T> zip(Stream<L> leftStream, Stream<R> rightStream, BiFunction<L, R, T> combiner) {
    Spliterator<L> lefts = leftStream.spliterator();
    Spliterator<R> rights = rightStream.spliterator();
    return StreamSupport.stream(new AbstractSpliterator<T>(Long.min(lefts.estimateSize(), rights.estimateSize()), lefts.characteristics() & rights.characteristics()) {
        @Override
        public boolean tryAdvance(Consumer<? super T> action) {
            return lefts.tryAdvance(left->rights.tryAdvance(right->action.accept(combiner.apply(left, right))));
        }
    }, leftStream.isParallel() || rightStream.isParallel());
}

Answer 6

La biblioteca Lazy-Seq proporciona funcionalidad zip.

https://github.com/nurkiewicz/LazySeq

Esta biblioteca está fuertemente inspirada en scala.collection.immutable.Stream y su objetivo es proporcionar una implementación de secuencia perezosa inmutable, segura para subprocesos y fácil de usar, posiblemente infinita.

Answer 7

Los métodos de la clase que mencionó se han movido a la interfaz de Stream en sí a los métodos predeterminados. Pero parece que el método zip ha sido eliminado. Tal vez porque no está claro cuál debería ser el comportamiento predeterminado para las diferentes corrientes de tamaño. Pero implementar el comportamiento deseado es sencillo:

static <T> boolean every(
  Collection<T> c1, Collection<T> c2, BiPredicate<T, T> pred) {
    Iterator<T> it=c2.iterator();
    return c1.stream().allMatch(x->!it.hasNext()||pred.test(x, it.next()));
}
static <T> T find(Collection<T> c1, Collection<T> c2, BiPredicate<T, T> pred) {
    Iterator<T> it=c2.iterator();
    return c1.stream().filter(x->it.hasNext()&&pred.test(x, it.next()))
      .findFirst().orElse(null);
}

Answer 8

Necesitaba esto también, así que tomé el código fuente de b93 y lo puse en una clase "util". Tuve que modificarlo ligeramente para trabajar con la API actual.

Como referencia, aquí está el código de trabajo (tómelo bajo su propio riesgo ...):

public static<A, B, C> Stream<C> zip(Stream<? extends A> a,
                                     Stream<? extends B> b,
                                     BiFunction<? super A, ? super B, ? extends C> zipper) {
    Objects.requireNonNull(zipper);
    Spliterator<? extends A> aSpliterator = Objects.requireNonNull(a).spliterator();
    Spliterator<? extends B> bSpliterator = Objects.requireNonNull(b).spliterator();

    // Zipping looses DISTINCT and SORTED characteristics
    int characteristics = aSpliterator.characteristics() & bSpliterator.characteristics() &
            ~(Spliterator.DISTINCT | Spliterator.SORTED);

    long zipSize = ((characteristics & Spliterator.SIZED) != 0)
            ? Math.min(aSpliterator.getExactSizeIfKnown(), bSpliterator.getExactSizeIfKnown())
            : -1;

    Iterator<A> aIterator = Spliterators.iterator(aSpliterator);
    Iterator<B> bIterator = Spliterators.iterator(bSpliterator);
    Iterator<C> cIterator = new Iterator<C>() {
        @Override
        public boolean hasNext() {
            return aIterator.hasNext() && bIterator.hasNext();
        }

        @Override
        public C next() {
            return zipper.apply(aIterator.next(), bIterator.next());
        }
    };

    Spliterator<C> split = Spliterators.spliterator(cIterator, zipSize, characteristics);
    return (a.isParallel() || b.isParallel())
           ? StreamSupport.stream(split, true)
           : StreamSupport.stream(split, false);
}

Answer 9

Si tiene Guayaba en su proyecto, puede usar el método Streams.zip (se agregó en Guava 21):

Devuelve una secuencia en la que cada elemento es el resultado de pasar el elemento correspondiente de cada una de streamA y streamB para funcionar. La secuencia resultante solo será tan larga como la más corta de las dos secuencias de entrada; si una secuencia es más larga, sus elementos adicionales serán ignorados. La secuencia resultante no es eficientemente divisible. Esto puede dañar el rendimiento paralelo.

 public class Streams {
     ...

     public static <A, B, R> Stream<R> zip(Stream<A> streamA,
             Stream<B> streamB, BiFunction<? super A, ? super B, R> function) {
         ...
     }
 }

Answer 10

zip es una de las funciones proporcionadas por la biblioteca protonpack .

Stream<String> streamA = Stream.of("A", "B", "C");
Stream<String> streamB  = Stream.of("Apple", "Banana", "Carrot", "Doughnut");

List<String> zipped = StreamUtils.zip(streamA,
                                      streamB,
                                      (a, b) -> a + " is for " + b)
                                 .collect(Collectors.toList());

assertThat(zipped,
           contains("A is for Apple", "B is for Banana", "C is for Carrot"));

Answer 11

public class Tuple<S,T> {
    private final S object1;
    private final T object2;

    public Tuple(S object1, T object2) {
        this.object1 = object1;
        this.object2 = object2;
    }

    public S getObject1() {
        return object1;
    }

    public T getObject2() {
        return object2;
    }
}


public class StreamUtils {

    private StreamUtils() {
    }

    public static <T> Stream<Tuple<Integer,T>> zipWithIndex(Stream<T> stream) {
        Stream<Integer> integerStream = IntStream.range(0, Integer.MAX_VALUE).boxed();
        Iterator<Integer> integerIterator = integerStream.iterator();
        return stream.map(x -> new Tuple<>(integerIterator.next(), x));
    }
}