Victoria Metrics quick start（使用docker +python操作）

VictoriaMetrics的介绍机器技术特性；与mysql对比，docker+python操作，使用理解，示例插入查询python脚本和查询数据讲解

Invictus path

1841人浏览 · 2024-01-08 16:06:56

Invictus path · 2024-01-08 16:06:56 发布

1.Victoria Metric数据库介绍

（这一部分主要来自官方文档，但是经过我的筛选，主要是面向使用Victoria Metric，和一些使用它必知必会的概念）

VictoriaMetrics 是一种快速、可扩展的监控解决方案和时间序列数据库（tsdb）
它可以用作 Grafana 中 Prometheus 的直接替代品，因为它支持Prometheus 查询 API
它实现了类似 PromQL 的查询语言 - MetricsQL，它在 PromQL 之上提供了改进的功能（就是一种查询语言，类似于mysql和sql的关系），但是语法简单、自然编程语言很像、没有像sql有各种关系和语法需要记忆，所以想要使用Victoria Metric，需要学习promql（学习难度很低）
它为数据摄取和数据查询提供高性能和良好的垂直和水平可扩展性

2.VM优势与MySql优势对比

VictoriaMetrics（VM）和MySQL，作为不同类型的数据库系统，各自具有独特的优势，主要体现在它们的设计目标和使用场景上。下表是具体各方面对比及原理：

	vm	mysql
数据	时间序列数据（如监控数据或指标）	通用数据和复杂数据关系
特性	1. 针对时间序列数据的查询，VM 提供了高效的数据聚合和计算能力，特别适合于快速检索和分析这类数据 2. VM 支持水平扩展，能够通过增加更多节点来处理更多的数据和查询，适合大规模部署。 3. 支持promql的metricql，简单，和自然编程语言很像但更简单，适合快速上手，甚至不需要系统学习	1. MySQL 提供完整的事务支持，包括原子性、一致性、隔离性和持久性（ACID），适合需要强事务处理的应用。 2. MySQL 提供完整的事务支持，包括原子性、一致性、隔离性和持久性（ACID），适合需要强事务处理的应用。 3. sql本身的灵活性、强大性、通用性、mysql社区的丰富性
C	特别优化用于快速插入大量的时间序列数据。它可以高效处理高频率的数据点插入，监控和指标收集场景中有大量的数据需要记录	支持多种数据类型和复杂的数据结构的插入，但在处理大量插入操作时，不如专门的tsdb高效。
R	优秀的读取性能，特别是对于时间序列数据的聚合和分析。它对 Prometheus 查询语言的原生支持使得查询时间序列数据变得非常高效。	提供了复杂的查询能力，包括联接、子查询等。这在需要进行复杂数据关联和分析的应用中非常有用。
U	时间序列数据的更新不是这类数据库的主要用途。	强大的更新功能，支持通过 SQL 语句进行复杂的、条件化的数据更新。这使得 MySQL 在需要频繁更新数据的应用中更为合适。
D	时间序列数据库通常不用于单条数据的删除操作，因为这类数据库的主要用途是长期存储数据用于分析。VM 也是如此，一般不需要也没必要删除。	提供强大的删除操作能力，可以精确控制要删除的数据。SQL 语言使得对数据的删除操作既灵活又强大。
技术原理	1. 作为时间序列数据库，VM 主要处理的是时间戳和相关数据值的序列。它使用列式存储，这种结构特别适合于快速读取大量时间序列数据。列式存储优化了数据的压缩率和读取效率，但并不适合频繁的单条数据修改或删除。 2. 它的索引和查询系统特别设计来处理按时间顺序存储的数据。VM 的查询优化主要集中在时间范围查询和数据聚合	1. 作为关系型数据库，MySQL 使用行式存储模型，每行存储一条记录的所有数据。这种结构便于增加、更新和删除单条记录。行式存储提供了更好的通用性和灵活性，但在处理大规模数据聚合查询时，不如列式存储高效。 2. 使用通用的索引机制，如 B-Tree，适用于各种类型的查询，包括基于非时间属性的查询。MySQL 的查询优化更为全面，支持复杂的连接（Join）和子查询。

3.使用docker拉起vm

这是docker-compose.yml文件中vm服务的配置，vm默认的端口号是8428，本人的机器上8428端口有正式项目的vm服务在占用，所以使用8429端口进行测试

使用docker-compose up命令拉起服务

这里对于vm的初学者（mysql的非初学者）需要做出解释，像关系型数据库，有服务之后还需要创建连接，还需要建库，建表，然后不管用原生sql还是什么其它封装了sql的框架组件进行查询；

但是vm拉起服务后就结束了，它是非关系型数据库没有表的概念，用个不完全贴切的比喻，就像一锅“大杂烩”；那么数据怎么插？怎么查？

想要回答这个问题，就必须要明白几个概念（只是quickstart,还有很多概念和语法请自行参阅官方文档）：

指标：这是vm中数据的标识，查vm就是查vm中的指标值

标签：指标可以有多个标签，主要用来按照某种逻辑筛选同类型指标

那么，指标名+标签 => 唯一确定了一个时间序列，也就是一个时间序列上的指标值,下面这个例子的指标名就是CPU_OCCUPIED，代表cpu占用率，{machine="2",process_id="1"}就是标签，代表了机器id和进程id，也就是说这个指标记录了id为2的机器上进程id为1的进程在一段时间序列中每分钟的cpu占用率

 # CPU_OCCUPIED{machine="2",process_id="1"}
timestamp        value
12:01             66
12:02             65   
12:03             78

所以数据是存在由指标和标签确定的一个时间序列中的，只有你向vm中插入指标值，这个指标才会存在！

查询则需要用vm提供的ui客户端中（）写metricql/promql查询和插入数据：

一旦容器启动，可以通过访问http://localhost:8428/vmui/?#/?g0.range_input=30m&g0.end_input=2023-11-14T09%3A44%3A40&g0.tab=0&g0.relative_time=last_30_minutes来访问VictoriaMetrics的Web界面。可以使用该界面来管理和监控你的指标数据。

或者使用编程语言，构造promql查询和发送http请求插入数据

下面用python编写脚本进行操作，当然，在正式的项目中，是需要高级工程师将各种查询方式封装成工具类来使用的（本blog暂时只提供python脚本方式操作，有工具类需求的朋友可以转发私信，后续可以考虑更新python使用vm工具类操作vm服务的具体实现）

4.使用python脚本插入数据

直接上脚本：

import time
import json
import requests
import urllib3

def insert_into_vm2(business, data_source, ad_task, ts, rows, product_type=1):
    # VictoriaMetrics 的地址和端口
    url = 'http://127.0.0.1:8428/api/v1/import'
    # json数据
    data = json.dumps({
        "metric": {
            "__name__": "analysis_data_volume_rows",    # {__name__="ask" or __name__="bid"} 查询ask或bid的数据
            "business": str(business),
            "data_source": str(data_source),
            "ad_task": str(ad_task),
            "product_type": str(product_type)
        },
        "values": [rows],
        "timestamps": [ts]    # 需要int
    })

    # 配置并发请求
    headers = {"Content-Type": "application/json"}
    http_pool = urllib3.PoolManager()
    response = http_pool.request('POST', url, body=data, headers=headers)
    print(response.status, f"data is {response.data}")

看懵了吧？哈哈，上解释！

既然是指标监控，那么就需要application以推送的模式发送到服务器，至于发送的时间和数据格式由应用程序决定，这也就是push模式以及vm支持的几种传输协议：

显然，json是我们最最常用的方式，我们也是用json来进行发送，（官方文档：VictoriaMetrics · VictoriaMetrics）

VictoriaMetrics 在/api/v1/import接受 JSON 行格式的数据，并在/api/v1/export导出此格式的数据。

格式遵循JSON 流概念，例如每行包含 JSON 对象，其指标数据采用以下格式（）：

{
  # metric contans metric name plus labels for a particular time series
  "metric":{
    "__name__": "metric_name",  # this is metric name

    # Other labels for the time series

    "label1": "value1",
    "label2": "value2",
    ...
    "labelN": "valueN"
  },

  # values contains raw sample values for the given time series
  "values": [1, 2.345, -678],

  # timestamps contains raw sample UNIX timestamps in milliseconds for the given time series
  # every timestamp is associated with the value at the corresponding position
  "timestamps": [1549891472010,1549891487724,1549891503438]
}

所以我们插入数据的url和准备的数据才会是那样写的，__name__就是指标名，label就是标签名，value就是指标值，timestamp就是时间戳，只需要把字典json.dumps变成json串就可以了

5.使用python查询数据

vm主要提供了两种方式的查询：即使查询和范围查询。

即时查询

即使查询在给定时间戳执行查询表达式：

GET | POST /api/v1/query?query=...&time=...&step=...

参数：

query- MetricSQL表达式。
time- 可选，以秒精度计算at 的时间戳query。如果省略，time则设置为now()（当前时间戳）。可以用多种允许的格式指定time参数。
step- 可选，执行时搜索过去的原始样本的最大间隔query。

查询的结果样例：

# json
{
  "status": "success",
  "data": {
    "resultType": "vector",
    "result": [
      {
        "metric": {
          "__name__": "foo_bar"
        },
        "value": [
          1652169780,
          "3"
        ]
      }
    ]
  }
}

所以将结果查询到之后就可以，json.loads成字典来操作了，下面的就是拿到了metrics这层的数据：

    def query_data(self, query=None, time=None, step=None):
        params = {
            "query": query
        }
        if time:
            params["time"] = time

        if step:
            params["step"] = step

        url = self.query_url + urlencode(params)
        try:
            # res = requests.get(url)
            res = self.http_pool.request('GET', url)
            if res.status == 200:
                data = json.loads(res.data).get("data").get("result")
                # logger.info(f"VictoriaMetrics query success, and url is {url}, and data is {data}")
                if data is None:
                    logger.info(f"VictoriaMetrics query null, and url is {url}")
                    return None
                return data
            else:
                logger.error(
                    f"VictoriaMetrics query failed, and url is {url}, response code is {res.status}, "
                    f"and error is {res.text}")
                return None
        except Exception as e:
            logger.error(f"VictoriaMetrics query failed, and url is {url}, exception is {e}")
            return None

但是有一点需要注意，那就是，无论插入数据的json中timestamp和指标值的先后数据，查询出来的数据（即value列表）都是timestamp在前，指标值在后：
返回数据的value列表是 [timestamp,指标值]

value = int(query_data[0].get("value", [])[1])

范围查询

范围查询以给定的步长在给定的时间范围内执行查询表达式：

GET | POST /api/v1/query_range?query=...&start=...&end=...&step=...

参数：

query- MetricSQL表达式。
start- 评估时间范围的起始时间戳query。
end- 评估时间范围的结束时间戳query。如果end未设置，则end自动设置为当前时间。
step-数据点之间的间隔，必须从范围查询返回。

查询的结果样例：

# json
{
  "status": "success",
  "data": {
    "resultType": "matrix",
    "result": [
      {
        "metric": {
          "__name__": "foo_bar"
        },
        "values": [
          [
            1652169600,
            "1"
          ],
          [
            1652169660,
            "2"
          ],
          [
            1652169720,
            "3"
          ],
          [
            1652169780,
            "3"
          ],
          [
            1652169840,
            "7"
          ],
          [
            1652169900,
            "7"
          ]
        ]
      }
    ]
  }
}

剩下的拿结果的方式后及时查询同理了。