应用 GroupBy 然后在 Google Dataflow 中应用 Count答案

【问题标题】：Applying GroupBy and then apply Count in Google Dataflow应用 GroupBy 然后在 Google Dataflow 中应用 Count
【发布时间】：2015-05-25 04:43:31
【问题描述】：

我的 Google 云存储中有以下内容

Advertiser | Event
__________________
100 | Click

101 | Impression

100 | Impression

100 | Impression

101 | Impression

我的管道输出应该类似于

Advertiser | Clicks | Impressions

100 | 1 | 2

101 | 0 | 2

首先我使用了groupByKey，输出是这样的

100 次点击、展示、展示

101 印象，印象

现在可以计算KV的值了吗？

目前我只是使用比较字符串来计算点击次数和展示次数。

这里可以使用计数转换吗？

或者我们在这里使用任何其他转换？

或者我做的方式是唯一的方式？

谢谢，山姆。

【问题讨论】：

标签： google-cloud-dataflow

【解决方案1】：

我假设您的输入是 PCollection<KV<Long, EventType>> input，其中 Long 是广告商 ID，EventType 是 enum { CLICK, IMPRESSION, possibly something else }。

我还假设您希望输出为 PCollection> 其中 AdvertiserStats 是一个具有“numClicks”、“numImpressions”字段的类。

在这种情况下，实现您想要的一种方法是使用Combine - input.apply(Combine.<Long, AdvertiserStats>perKey(new ComputeAdvertiserStatsFn()))，其中ComputeAdvertiserStatsFn 的定义如下：

public class ComputeAdvertiserStatsFn
    extends CombineFn<EventType, AdvertiserStats, AdvertiserStats> {
  public AdvertiserStats createAccumulator() { return new AdvertiserStats(); }
  public void addInput(AdvertiserStats stats, EventType input) {
    switch (input) {
    case CLICK: stats.numClicks++; break;
    case IMPRESSION: stats.numImpressions++; break;
    default: (depending on your application?)
    }
  }
  public AdvertiserStats mergeAccumulators(Iterable<AdvertiserStats> stats) {
    AdvertiserStats merged = createAccumulator();
    for (AdvertiserStats item : stats) {
      merged.numClicks += item.numClicks;
      merged.numImpressions += item.numImpressions;
    }
    return merged;
  }
  public AdvertiserStats extractOutput(AdvertiserStats stats) { return stats; }
}

这应该表现得很好，因为大部分分组和计数将在本地发生。

目前，AFAIK，没有PTransform 可以为您完成ComputeAdvertiserStatsFn 的工作。我认为理想的界面应该类似于 input.apply(Combine.perKey(Count.perElement()))，但它不适用于当前定义的方式。

【讨论】：

我会试试这个解决方案。但这与我所做的或多或少相同。我通过编写一个 DoFn 来做到这一点，我将在其中计算每一次展示和点击。我认为这不是有效的方法。您的解决方案能否在处理大量数据时表现良好？
我想你的意思是一个解决方案，你通过广告商 ID 分组，然后在你的 DoFn 中计算不同的事件类型。建议的基于组合的解决方案应该表现得更好，因为组合在本地应用了大部分分组和组合，因此大大减少了混洗的数据量，从而大大减少了整体 IO。基本上，每次您可以将您的聚合表示为一个组合时，您都应该这样做。