如何在 Dataflow 中做两个 PCollection 的笛卡尔积？答案

【问题标题】：How to do a cartesian product of two PCollections in Dataflow?如何在 Dataflow 中做两个 PCollection 的笛卡尔积？
【发布时间】：2016-12-08 23:13:26
【问题描述】：

我想做两个 PCollection 的笛卡尔积。 PCollection 都不能放入内存，因此进行侧面输入是不可行的。

我的目标是：我有两个数据集。一种是小尺寸的许多元素。另一个是非常大的少数（~10）。我想取这两个元素的乘积，然后生成键值对象。

【问题讨论】：

标签： java google-cloud-dataflow gcloud

【解决方案1】：

我认为 CoGroupByKey 可能适用于您的情况：

https://cloud.google.com/dataflow/model/group-by-key#join

这就是我为类似的用例所做的。虽然我的可能没有受到内存的限制（您是否尝试过使用更大机器的更大集群？）：

PCollection<KV<String, TableRow>> inputClassifiedKeyed = inputClassified
        .apply(ParDo.named("Actuals : Keys").of(new ActualsRowToKeyedRow()));

PCollection<KV<String, Iterable<Map<String, String>>>> groupedCategories = p
[...]
.apply(GroupByKey.create());

所以这些集合是由同一个键作为键的。

然后我声明了标签：

final TupleTag<Iterable<Map<String, String>>> categoryTag = new TupleTag<>();
final TupleTag<TableRow> actualsTag = new TupleTag<>();

合并它们：

PCollection<KV<String, CoGbkResult>> actualCategoriesCombined =
        KeyedPCollectionTuple.of(actualsTag, inputClassifiedKeyed)
                .and(categoryTag, groupedCategories)
                .apply(CoGroupByKey.create());

在我的情况下，最后一步 - 重新格式化结果（来自连续流中的标记组：

actualCategoriesCombined.apply(ParDo.named("Actuals : Formatting").of(
    new DoFn<KV<String, CoGbkResult>, TableRow>() {
        @Override
        public void processElement(ProcessContext c) throws Exception {
            KV<String, CoGbkResult> e = c.element();

            Iterable<TableRow> actualTableRows =
                    e.getValue().getAll(actualsTag);
            Iterable<Iterable<Map<String, String>>> categoriesAll =
                    e.getValue().getAll(categoryTag);

            for (TableRow row : actualTableRows) {
                // Some of the actuals do not have categories
                if (categoriesAll.iterator().hasNext()) {
                    row.put("advertiser", categoriesAll.iterator().next());
                }
                c.output(row);
            }
        }
    }))

希望这会有所帮助。再次 - 不确定内存限制。如果您尝试这种方法，请务必告诉结果。

【讨论】：

由于我缺乏Java和数据流方面的经验，我需要很长时间才能解析这个。如果可能的话，你能在 python 中发布一个简单的例子吗？我认为扩展这个问题而不是我提出一个新问题会更好。
作为参考，我来自 Spark，这很简单：collection_a.cartesian(collection_b)
我很乐意这样做，你介意私下联系吗？

【解决方案2】：

创建笛卡尔积使用Apache Beam extension Join

import org.apache.beam.sdk.extensions.joinlibrary.Join;

...

// Use function Join.fullOuterJoin(final PCollection<KV<K, V1>> leftCollection, final PCollection<KV<K, V2>> rightCollection, final V1 leftNullValue, final V2 rightNullValue)
// and the same key for all rows to create cartesian product as it is shown below:

    public static void process(Pipeline pipeline, DataInputOptions options) {
        PCollection<KV<Integer, CpuItem>> cpuList = pipeline
                .apply("ReadCPUs", TextIO.read().from(options.getInputCpuFile()))
                .apply("Creating Cpu Objects", new CpuItem()).apply("Preprocess Cpu",
                        MapElements
                                .into(TypeDescriptors.kvs(TypeDescriptors.integers(), TypeDescriptor.of(CpuItem.class)))
                                .via((CpuItem e) -> KV.of(0, e)));

        PCollection<KV<Integer, GpuItem>> gpuList = pipeline
                .apply("ReadGPUs", TextIO.read().from(options.getInputGpuFile()))
                .apply("Creating Gpu Objects", new GpuItem()).apply("Preprocess Gpu",
                        MapElements
                                .into(TypeDescriptors.kvs(TypeDescriptors.integers(), TypeDescriptor.of(GpuItem.class)))
                                .via((GpuItem e) -> KV.of(0, e)));

        PCollection<KV<Integer,KV<CpuItem,GpuItem>>>  cartesianProduct = Join.fullOuterJoin(cpuList, gpuList, new CpuItem(), new GpuItem());
        PCollection<String> finalResultCollection = cartesianProduct.apply("Format results", MapElements.into(TypeDescriptors.strings())
                .via((KV<Integer, KV<CpuItem,GpuItem>> e) -> e.getValue().toString()));
        finalResultCollection.apply("Output the results",
                TextIO.write().to("fps.batchproc\\parsed_cpus").withSuffix(".log"));
        pipeline.run();
    }

在上面的代码中这一行

...
        .via((CpuItem e) -> KV.of(0, e)));
...

我为输入数据中可用的所有行创建键等于 0 的 Map。结果，所有行都匹配。即等于没有 WHERE 子句的 SQL 表达式 JOIN

【讨论】：