我有一个 Apache Beam 作业，它从 PubSub 注入数据，然后加载到 BigQuery， 我将 PubSub 消息转换为带有字段的 pojo
身份证， 姓名，人数
Count 是指非唯一元素在单个摄取中的计数。
如果我从 PubSub 3 个元素加载，其中两个是相同的，那么我需要加载到 BigQuery 2 个元素中，其中一个的计数为 2。
我想知道在 Apache Beam 中实现它是多么容易。 我试图通过 DoFn 或 MapElements 制作它，但我只能处理单个元素。 我也尝试将元素转换为 KV，然后计数，但我有非确定性编码器。
在通常的 java 应用程序中，我可以简单地使用 equals 或通过 Map，但在 Apache Beam 中，一切都不同。
【问题讨论】：
标签： java etl apache-beam
简单而正确的方法是使用Count.<T>perElement()，像这样：
Pipeline p = ...;
PCollection<T> elements = p.apply(...); // read elements
PCollection<KV<T, Long>> elementsCounts =
elements.apply(Count.<T>perElement());
PCollection<TableRow> results = elementsCounts.apply(ParDo.of(
new FormatOutputFn()));
尽管如此，您需要有一个确定性元素编码器来实现这一点。因此，如果不是这种情况（正如我从您上面所说的那样），您需要在 Count 之前添加一个步骤，以将元素转换为不同的表示形式，这样就有可能拥有确定性编码器（例如 AvroCoder )。
如果由于某些原因不可能，那么另一种解决方法可能是为每个元素计算一个 uniq 哈希（但哈希值也必须是确定性的），为每个元素创建一个 KV，并将新哈希作为 @ 987654327@ 和元素作为Value 并在下游使用GroupByKey 具有相同值的分组元组。
另外，请注意，由于PubSub 是一个无限源，您需要通过任何类型的Windows 策略“窗口化”您的输入（Global 一个除外），因为您的所有组/组合操作都应该是在窗口内完成。看看WindowedWordCount 作为类似问题的解决方案示例。
【讨论】：