如何在Apache Beam中为特定消息属性的每个取值创建自定义指标

方案1：使用带标签的Counter（推荐，适配Cloud Monitoring原生功能） #

Apache Beam 2.20及以上版本的Counter指标支持附加自定义标签，你可以将eventType作为标签值绑定到计数器上，Dataflow运行时会自动将带标签的指标同步到Google Cloud Monitoring，天然支持按标签拆分统计：

// 定义计数器的命名空间、名称，以及标签键
private static final String METRIC_NAMESPACE = "your-namespace";
private static final String EVENT_COUNTER_NAME = "event_count_by_type";
private static final String EVENT_TYPE_LABEL = "event_type";

@ProcessElement
public void processElement(ProcessContext context) {
  // 解析JSON获取eventType值
  String eventType = context.element().get("eventType").getAsString();
  // 构造带标签的计数器并递增
  Metrics.counter(METRIC_NAMESPACE, EVENT_COUNTER_NAME)
    .withTag(EVENT_TYPE_LABEL, eventType)
    .inc();
  // 原有业务逻辑
  context.output(context.element());
}

注意事项：如果你的eventType取值量级超过100，建议先做值的合法性校验和收敛，避免生成过多时间序列超出Cloud Monitoring配额
后续在Cloud Monitoring中选择custom.googleapis.com/your-namespace/event_count_by_type指标，按event_type标签分组，选择堆叠柱状图展示类型即可实现需求。

方案2：低版本Beam兼容方案 #

如果你使用的Beam版本低于2.20不支持标签功能，可以在DoFn中维护Map存储不同eventType对应的Counter实例：

private static final String METRIC_NAMESPACE = "your-namespace";
// 存储eventType与对应Counter的映射
private final Map<String, Counter> eventCounterMap = new HashMap<>();

@ProcessElement
public void processElement(ProcessContext context) {
  String eventType = context.element().get("eventType").getAsString();
  // 不存在则新建Counter，注意Counter命名要包含eventType标识
  Counter counter = eventCounterMap.computeIfAbsent(eventType, 
    k -> Metrics.counter(METRIC_NAMESPACE, "event_count_" + eventType)
  );
  counter.inc();
  // 原有业务逻辑
  context.output(context.element());
}

该方案仅适合eventType取值固定且数量较少的场景，否则会生成大量独立的指标项，增加监控配置成本。

内容的提问来源于stack exchange，提问作者jamiet