Metrics in Beam+Spark

[Yushu Yao]

Hi Team,
Does anyone have a working example of a beam job running on top of spark?
So that I can use the beam metric syntax and the metrics will be shipped
out via spark's infra?

The only thing I achieved is to be able to queryMetrics() every half second
and copy all the metrics into the spark metrics.
Wondering if there is a better way?

Thanks!
-Yushu

MetricQueryResults metrics =
    pipelineResult
        .metrics()
        .queryMetrics(
            MetricsFilter.builder()
                .addNameFilter(MetricNameFilter.inNamespace(namespace))
                .build());

for (MetricResult<GaugeResult> cc : metrics.getGauges()) {
    LOGGER.info("Adding Gauge: {} : {}", cc.getName(), cc.getAttempted());
    com.codahale.metrics.Gauge gauge =
        new SimpleBeamGauge(cc.getAttempted().getValue());
    try {
        String name = metricName(cc.getKey(), addStepName, addNamespace);
        if (registry.getNames().contains(name)) {
            LOGGER.info("Removing metric {}", name);
            registry.remove(name);
        }
        registry.register(name, gauge);
    } catch (IllegalArgumentException e) {
        LOGGER.warn("Duplicated metrics found. Try turning on
addStepName=true.", e);
    }
}