Найти декартово произведение двух списков с помощью Apache Beam w java

Question

У меня 2 коллекции

      PCollection<List<String>> ListA = pipeline.apply("getListA", ParDo.of(new getListA()))
PCollection<List<String>> ListB = pipeline.apply("getListB", ParDo.of(new getListB()))

ListA содержит

      ["1","2","3"]

ListB содержит

      ["A","B","C"]

Как мне получить коллекцию PCollection, содержащую

      [
 ["A","1"],["A","2"],["A","3"],
 ["B","1"],["B","2"],["B","3"],
 ["C","1"],["C","2"],["C","3"],
]

Мой поиск указал мне на

Как сделать декартово произведение двух коллекций PCollections в потоке данных?

но это имеет дело с KV с использованием coGroupby с 2 выходами. Возможно, что coGroupby можно использовать для создания декартового произведения двух списков, но я этого не вижу.

Answer 1

Вы можете использовать Java 8 Stream.reduce метод следующим образом:

      List<String> listA = Arrays.asList("1", "2", "3");
List<String> listB = Arrays.asList("A", "B", "C");

      List<List<String>> cartesianProduct = Stream.of(listA, listB)
        // represent each list element as a singleton list
        .map(list -> list.stream().map(Collections::singletonList)
                .collect(Collectors.toList()))
        // summation of pairs of inner lists
        .reduce((list1, list2) -> list1.stream()
                // combinations of inner lists
                .flatMap(inner1 -> list2.stream()
                        // merge two inner lists into one
                        .map(inner2 -> new ArrayList<String>() {{
                            addAll(inner1);
                            addAll(inner2);
                        }}))
                // list of combinations
                .collect(Collectors.toList()))
        // List<List<String>>
        .orElse(Collections.emptyList());

      // output
System.out.println(cartesianProduct);
//[[1, A], [1, B], [1, C], [2, A], [2, B], [2, C], [3, A], [3, B], [3, C]]

---

^{См. Также: Как создать декартово произведение с потоками Java 8?}

Answer 2

Похоже, что у вас есть один элемент в каждой коллекции PCollection, поэтому вам просто нужно объединить эти элементы, а затем вы можете самостоятельно выполнить декартово произведение в DoFn

Что-то вроде

      Flatten.pcollections(ListA, List)
.apply(WithKeys.of(null))
.apply(GroupByKey.create())

После этого у вас будет PCollection с одним элементом, которым является KV (null, Iterable (ListA, ListB)), и вы можете сгенерировать декартово произведение с некоторыми циклами for.