跳到主要内容
版本：Next

Prometheus

Prometheus 数据接收器

引擎支持​

Spark
Flink
SeaTunnel Zeta

主要特性​

描述​

Prometheus 数据接收器把上游数据写入 Prometheus remote write API。它会从上游数据中取出 3 个字段来组成 Prometheus 采样点：

  • key_label：Prometheus 标签字段，通常是 map<string, string>。
  • key_value：指标数值字段。
  • key_timestamp：可选的时间戳字段。

接收器会按 Prometheus remote write 协议序列化数据，用 Snappy 压缩后，通过 HTTP POST 请求写入 Prometheus 兼容 remote write 地址，例如 http://prometheus:9090/api/v1/write 或 http://victoria-metrics:8428/api/v1/write。

如果样本时间太早，目标服务可能会按自己的保留策略或 remote write 规则拒绝写入。

支持的数据源信息​

使用 Prometheus 连接器时，需要安装以下依赖。可以通过 install-plugin.sh 安装，也可以从 Maven 中央仓库下载。

数据源支持版本依赖
Prometheusuniversal下载

接收器选项​

名称类型是否必填默认值描述
urlString是-Prometheus 兼容 remote write API 地址，例如 http://prometheus:9090/api/v1/write。
key_labelString是-上游数据中保存 Prometheus 标签的字段名。字段值建议为 map。
key_valueString是-上游数据中保存 Prometheus 指标值的字段名。推荐使用 double 类型字段。
key_timestampString否-上游数据中保存 Prometheus 指标时间戳的字段名。不配置时使用当前系统时间。
headersMap否-HTTP 请求头。
retryInt否-HTTP 请求出现 IOException 时的最大重试次数。
retry_backoff_multiplier_msInt否100重试退避时间倍数，单位毫秒。
retry_backoff_max_msInt否10000最大重试退避时间，单位毫秒。
batch_sizeInt否1024写入 Prometheus 前最多缓存的行数。
multi_table_sink_replicaInt否1多表写入时，每张表使用的写入器副本数。
common-optionsConfig否-接收器插件通用参数，详情请参考接收器通用选项。

key_label​

对应字段建议为 map<string, string>，会被转换为 Prometheus 标签。建议在 map 中包含 __name__，用来表示指标名。

Sink 会自动补充 remote write 需要的请求头：Content-type、Content-Encoding 和 X-Prometheus-Remote-Write-Version。

key_timestamp​

支持以下字段类型：

  • timestamp：按本地时区转换为毫秒级时间戳
  • bigint：按毫秒级时间戳处理
  • double：按 Unix 秒级时间戳处理，并转换为毫秒
  • string：按毫秒级时间戳解析

multi_table_sink_replica​

多表写入时，每张表使用的 Sink Writer 副本数。默认值为 1；只有当单张表需要更高写入并行度时才建议调大。

定时刷新​

即使上游数据空闲、缓存的行数还没达到 batch_size，接收器也可以按定时器刷新缓存，把已缓存的采样点发送出去。该定时器由引擎驱动，而不是由连接器驱动，目前仅 SeaTunnel Zeta 支持。

在作业的 env 中设置 sink.flush.interval（单位毫秒）即可启用：

env {
  sink.flush.interval = 10000
}

引擎会在正常的 Sink 数据处理线程上触发刷新，因此不需要连接器自己维护后台线程，也不会和写入、检查点、关闭等流程产生并发。刷新失败会被抛给引擎，而不会被静默丢弃。

在 Spark 和 Flink 上没有检查点之间的定时刷新：sink.flush.interval 是 Zeta 引擎的能力，Spark/Flink 的 Sink 写入器上下文并未实现它。在这两个引擎上，缓存会在达到 batch_size、检查点时（PrometheusWriter 在 prepareCommit() 中刷新）以及写入器关闭时被刷新。因此缓存的采样点最多保留一个检查点间隔，而不会一直保存到 batch_size 或关闭。如需降低 Spark 或 Flink 上检查点之间的延迟，请相应调整 batch_size。

检查点刷新在所有引擎上都会执行，包括 Zeta。因此在 Zeta 上，缓存会同时由 sink.flush.interval 和每个检查点触发刷新：如果检查点间隔短于 sink.flush.interval，刷新会比仅靠定时器时更频繁（每批更小）。这是预期行为；如果关注请求频率，请同时调整 sink.flush.interval 和检查点间隔。

检查点刷新与失败处理​

检查点刷新是一次 remote-write 请求，刷新失败会让检查点失败，而不会丢弃这批数据。有两点需要了解：

  • 瞬时失败会导致检查点失败。 网络抖动、接收端重启或 5xx 响应都会让当前检查点失败。Flink 的 tolerableCheckpointFailureNumber 默认是 0，因此一次失败就会重启作业；在 Spark 和 Flink 上，对于低吞吐作业你可能需要调高引擎的可容忍检查点失败次数。刷新内部的有界重试（带退避）作为后续项跟踪在 #11911。
  • 重放的安全性取决于接收端。 检查点失败后作业会重启，Source 从上一次成功的检查点重放，因此缓存的采样点会被重新发送。只有当 remote-write 接收端接受完全相同的重复样本（相同的 labels、timestamp 和 value）时，这才是安全的。如果接收端拒绝相同 timestamp 但 value 不同的样本，或拒绝乱序样本（Prometheus TSDB，以及 Cortex、Mimir、Thanos 等接收端对这些情况会返回 400），重放的刷新就会失败并持续让检查点失败。如果这对你的部署很重要，请启用接收端的乱序窗口，或确保重放的是完全相同的重复样本。

示例​

env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  FakeSource {
    schema = {
      fields {
        c_map = "map<string, string>"
        c_double = double
        c_timestamp = timestamp
      }
    }
    plugin_output = "fake"
    rows = [
      {
        kind = INSERT
        fields = [{"__name__" : "metric_1"}, 1.23, CURRENT_TIMESTAMP]
      },
      {
        kind = INSERT
        fields = [{"__name__" : "metric_2"}, 1.23, CURRENT_TIMESTAMP]
      }
    ]
  }
}

sink {
  Prometheus {
    plugin_input = "fake"
    url = "http://prometheus:9090/api/v1/write"
    key_label = "c_map"
    key_value = "c_double"
    key_timestamp = "c_timestamp"
    batch_size = 1
  }
}

Prometheus 兼容 Remote Write 示例​

sink {
  Prometheus {
    plugin_input = "fake"
    url = "http://victoria-metrics:8428/api/v1/write"
    key_label = "c_map"
    key_value = "c_double"
    key_timestamp = "c_timestamp"
    batch_size = 5
  }
}

变更日志​