Prometheus的Pyspark UDF监视

发布于 2025-01-24 12:52:11 字数 1297 浏览 2 评论 0 原文

我正在尝试使用计数器在UDF中监视某些逻辑。

即

counter = Counter(...).labels("value")

@ufd
def do_smthng(col):
  if col:
    counter.label("not_null").inc()
  else:
    counter.label("null").inc()
  return col

这不是实际情况，但是您应该明白这个主意。我遵循了这篇文章： https://kb.databricks.com/metrics/metrics/spark-metrics/spark-metrics.html

到目前为止，我已经尝试过：

使用全局Propetheus计数器（锁定失败是不可拾取的），
使用PY4J创建自定义源：


# noinspection PyPep8Naming
class CustomMetrics:
    def __init__(self, sourceName, metricRegistry):
        self.metricRegistry = metricRegistry
        self.sourceName = sourceName

    class Java:
        implements = ["org.apache.spark.metrics.source.Source"]

py_4j_gateway = spark_session.sparkContext._gateway
metric_registry = py_4j_gateway.jvm.com.codahale.metrics.MetricRegistry()
SparkEnv = py_4j_gateway.jvm.org.apache.spark.SparkEnv
custom_metrics_provider = CustomMetrics("spark.ingest.custom", metric_registry)

它在同一错误中失败。我也无法获得 sparkenv.get.metricssystem ，因此无论如何我都无法注册自定义指标客户端。

我无法从Python访问内部度量注册表吗？我开始想知道人们如何使用自定义指标监视火花管道。

火花3.1.2 Python 3.8 x86 Mackbook Pro M1 Pro

原文

I am am trying to monitor some logic in a udf using counters.

i.e.

counter = Counter(...).labels("value")

@ufd
def do_smthng(col):
  if col:
    counter.label("not_null").inc()
  else:
    counter.label("null").inc()
  return col

This is not the real case, but you should get the idea.
I have followed this article:
https://kb.databricks.com/metrics/spark-metrics.html

I have so far tried:

Using a global prometheus counter (Failed with Lock is not picklable)
Creating a custom source using py4j:


# noinspection PyPep8Naming
class CustomMetrics:
    def __init__(self, sourceName, metricRegistry):
        self.metricRegistry = metricRegistry
        self.sourceName = sourceName

    class Java:
        implements = ["org.apache.spark.metrics.source.Source"]

py_4j_gateway = spark_session.sparkContext._gateway
metric_registry = py_4j_gateway.jvm.com.codahale.metrics.MetricRegistry()
SparkEnv = py_4j_gateway.jvm.org.apache.spark.SparkEnv
custom_metrics_provider = CustomMetrics("spark.ingest.custom", metric_registry)

Which failed with the same error.
I also can't get SparkEnv.get.metricsSystem so I can't register the custom metrics client in any case.

Is there no way for me to access the internal metric registry from python?
I am starting to wonder how people do monitor spark pipelines with custom metrics.

Spark 3.1.2
Python 3.8 x86
MackBook Pro M1 Pro