企业官网建设流程全解析

1. PyFlink Metrics 的入口：UDF.open() + MetricGroup

在 Python UDF 里，指标注册通常写在open()：

• open()：每个并行子任务（subtask）初始化时调用一次
• eval()：每条数据调用（或每批数据调用，取决于 UDF 类型）

所以推荐模式是：

• open()里注册指标（Counter/Gauge/Distribution/Meter）
• eval()里更新指标

示例骨架：

frompyflink.table.udfimportScalarFunctionclassMyUDF(ScalarFunction):defopen(self,function_context):mg=function_context.get_metric_group()# register metrics heredefeval(self,x):# update metrics herereturnx

2. 四类指标类型：Counter / Gauge / Distribution / Meter

PyFlink 支持四种常用指标类型，各自适用场景不同。

2.1 Counter：计数器（最常用）

用途：统计处理条数、错误数、某类事件数等
更新方式：inc()/inc(n)/dec()/dec(n)

frompyflink.table.udfimportScalarFunctionclassMyUDF(ScalarFunction):def__init__(self):self.counter=Nonedefopen(self,function_context):self.counter=function_context.get_metric_group().counter("my_counter")defeval(self,i):self.counter.inc(i)# 示例里用 i 递增returni

工程建议（更贴近生产）：

• 用inc()统计条数
• 用独立 counter 统计异常：error_counter.inc()

2.2 Gauge：按需取值（只能是 int）

用途：展示“当前状态值”，例如当前缓存大小、最近一条数据长度、队列长度等
注册方式：gauge(name, Callable[[], int])
限制：Gauge 只支持整数

frompyflink.table.udfimportScalarFunctionclassMyUDF(ScalarFunction):def__init__(self):self.length=0defopen(self,function_context):function_context.get_metric_group().gauge("my_gauge",lambda:self.length)defeval(self,i):self.length=ireturni-1

工程建议：

• Gauge 的 callable 里不要做重计算，只返回当前值
• 如果你要报 float，通常做放大（比如乘 1000 转 int）或换别的指标类型/外部上报策略

2.3 Distribution：分布统计（sum/count/min/max/mean，只支持 int）

用途：统计某个值的分布特征，比如每条数据大小、处理耗时（毫秒）、某字段长度等
更新方式：update(n: int)

frompyflink.table.udfimportScalarFunctionclassMyUDF(ScalarFunction):def__init__(self):self.distribution=Nonedefopen(self,function_context):self.distribution=function_context.get_metric_group().distribution("my_distribution")defeval(self,i):self.distribution.update(i)returni-1

工程建议：

• 用毫秒/字节/长度这种自然 int 的指标最合适
• 如果是耗时，尽量在 Python 内用轻量计时（避免每条数据记录太重）

2.4 Meter：吞吐率（事件/秒，滑动时间窗）

用途：看吞吐趋势，比如每秒处理记录数、某类事件速率
更新方式：mark_event()/mark_event(n)
可配置统计时间窗：默认 60s，可指定 120s 等

frompyflink.table.udfimportScalarFunctionclassMyUDF(ScalarFunction):def__init__(self):self.meter=Nonedefopen(self,function_context):self.meter=function_context.get_metric_group().meter("my_meter",time_span_in_seconds=120)defeval(self,i):self.meter.mark_event(i)returni-1

工程建议：

• 一般用mark_event(1)表示处理 1 条
• 不要把业务字段值当成 event 数随便塞进去，除非它就是“事件个数”

3. 指标分组：add_group() 做业务维度聚合

你可以通过MetricGroup.add_group(key, value=None)做分组，形成更清晰的指标层级。

3.1 普通分组（类似 namespace）

function_context \.get_metric_group()\.add_group("my_metrics")\.counter("my_counter")

效果：指标会挂在my_metrics分组下，避免所有指标挤在一个层级。

3.2 key-value 分组（定义 user variable）

function_context \.get_metric_group()\.add_group("my_metrics_key","my_metrics_value")\.counter("my_counter")

注意点（文档强调）：

• 这种写法会创建“用户变量（user variable）”
• 用户变量不能用在 scope formats（也就是不能指望它出现在 scope 格式化模板里）

4. 生产最佳实践：怎么埋点才有用、不拖垮性能？

下面这些是“埋了之后真的能救命”的指标组合（建议你直接套用）：

• processed_records（Counter）：处理总条数
• error_records（Counter）：异常条数（try/except 里 inc）
• current_cache_size（Gauge）：当前缓存/字典大小（如果你在 open 里加载了东西）
• latency_ms（Distribution）：单条处理耗时或某阶段耗时（整数毫秒）
• throughput_rps（Meter）：记录速率（每秒条数）

性能注意：

• 指标更新要轻：Counter/Meter 很轻，Distribution/Gauge callable 也尽量轻
• 不要在 Gauge 的 lambda 里做昂贵计算
• 分组不要做高基数维度（例如把 user_id 当 group value），会导致指标爆炸

5. 一段“可直接用于生产 UDF”的埋点模板

你可以把这段作为自己的标准模板（结构清晰，扩展方便）：

• open：注册 metrics
• eval：更新 metrics + 业务处理

（如果你需要，我也可以按你现有的 UDF 样式，给你写一个“带异常计数 + 耗时分布 + 吞吐 meter + 分组”的完整类）