Отсутствующие метрики для конвейера Apache Beam (через SparkRunner / Dataproc)

В настоящее время я добавляю некоторые метрики в существующий конвейер, который работает в Google Dataproc через Spark Runner, и я пытаюсь определить, как получить доступ к этим метрикам и, в конечном итоге, предоставить их Stackdriver (для последующего использования в панелях мониторинга Grafana. ).

Сами метрики довольно просты (серия счетчиков) и определяются как таковые (и доступны в DoFns по всему конвейеру):

object Metrics {
   val exampleMetric: Counter = Metrics.counter(ExamplePipeline::class.qualifiedName, "count")

   // Others omitted for brevity
}

Эта метрика (и другие) увеличиваются на протяжении всего конвейера в различных DoFn вызовах, и несколько модульных тестов подтверждают, что объект MetricQueryResults из конвейера правильно заполнен после выполнения через DirectRunner.

Основная проблема заключается в том, что я не вижу никаких указаний ни в Dataproc, ни в каком-либо из связанных пользовательских интерфейсов, представленных в GCP (YARN ResourceManager, Spark History Server, YARN Application Timeline и т. д.), что эти метрики генерируются. Я пробовал просматривать журналы и все, что можно, но я не вижу никаких признаков этих пользовательских метрик (или вообще каких-либо метрик, исходящих из Spark и / или в Stackdriver).

Конфигурация работы

Само задание Spark настраивается с помощью следующей команды в сценарии (при условии, что соответствующий файл .jar был скопирован в соответствующую корзину в GCP:

gcloud dataproc jobs submit spark --jar $bucket/deployment/example-pipeline.jar \
       --project $project_name \
       --cluster $cluster_name \
       --region $region  \
       --id pipeline-$timestamp \
       --driver-log-levels $lots_of_things_here \
       --properties=spark.dynamicAllocation.enabled=false \
       --labels="type"="example-pipeline","namespace"="$namespace" \
       --async \
       -- \
         --runner=SparkRunner \
         --streaming

Конфигурация кластера

В самом кластере, похоже, включены все свойства, связанные с метрикой, которые я мог придумать, например:

dataproc:dataproc.logging.stackdriver.enable=true
dataproc:dataproc.logging.stackdriver.job.driver.enable=true
dataproc:dataproc.monitoring.stackdriver.enable=true
dataproc:spark.submit.deployMode=cluster
spark:spark.eventLog.dir=hdfs:/https://var/log/spark/apps
spark:spark.eventLog.enabled=true
yarn:yarn.log-aggregation-enable=true
yarn:yarn.log-aggregation.retain-seconds=-1

Это всего лишь несколько свойств в кластере, однако существует бесчисленное множество других, поэтому, если какое-то из них кажется отсутствующим или неправильным (что касается истории метрик), не стесняйтесь спрашивать.

Вопросов

• Не похоже, что эти метрики автоматически генерируются или отображаются из Spark (или в Stackdriver), есть ли какая-то недостающая конфигурация на уровне кластера / задания? Или что-то вроде интерфейса MetricsOptions?
• Как только у нас будут фактически выдаваемые метрики, я бы предположил, что у Stackdriver есть механизм для обработки их потребления из DataProc (что, похоже, похоже на то, что будет обрабатывать dataproc:dataproc.monitoring.stackdriver.enable=true). Так ли это?

Я должен представить, что это довольно распространенный вариант использования (для Spark / Dataproc / Beam), но я не уверен, какие части головоломки конфигурации отсутствуют, а документация / статьи, связанные с этим процессом, кажутся довольно скудными.

Заранее спасибо!