入门指南

安装

创建 Python 环境

使用 Conda 来创建一个 Python 3 环境。Conda 是一个管理 Python 环境的实用工具，参见 official docs 以获得更多信息。

shell

  1. conda create --name Greptime python=3
  2. conda activate Greptime

安装 GreptimeDB

请参考 安装 GreptimeDB。

Hello world 实例

让我们从 hello world 实例开始入手：

python

  1. @coprocessor(returns=['msg'])
  2. def hello() -> vector[str]:
  3.    return "Hello, GreptimeDB"

将其保存为 hello.py，然后通过 HTTP API 发布：

提交 Python 脚本到 GreptimeDB

sh

  1. curl --data-binary "@hello.py" -XPOST "http://localhost:4000/v1/scripts?name=hello&db=public"

然后在 SQL 中调用：

sql

  1. select hello();

sql

  1. +-------------------+
  2. | hello()           |
  3. +-------------------+
  4. | Hello, GreptimeDB |
  5. +-------------------+
  6. 1 row in set (1.77 sec)

或者通过 HTTP API 进行调用：

sh

  1. curl -XPOST "http://localhost:4000/v1/run-script?name=hello&db=public"

json

  1. {
  2.   "code": 0,
  3.   "output": [
  4.     {
  5.       "records": {
  6.         "schema": {
  7.           "column_schemas": [
  8.             {
  9.               "name": "msg",
  10.               "data_type": "String"
  11.             }
  12.           ]
  13.         },
  14.         "rows": [["Hello, GreptimeDB"]]
  15.       }
  16.     }
  17.   ],
  18.   "execution_time_ms": 1917
  19. }

函数 hello 带有 @coprocessor 注解。

@coprocessor 中的 returns 指定了返回的列名，以及整体的返回格式：

json

  1.        "schema": {
  2.           "column_schemas": [
  3.             {
  4.               "name": "msg",
  5.               "data_type": "String"
  6.             }
  7.           ]
  8.         }

参数列表后面的 -> vector[str] 指定了函数的返回类型，都是具有具体类型的 vector。返回类型是生成 coprocessor 函数的输出所必需的。

hello 的函数主体返回一个字面字符串 "Hello, GreptimeDB"。Coprocessor 引擎将把它转换成一个常量字符串的 vector 并返回它。

总的来说一个协处理器包含三个主要部分：

  • @coprocessor 注解
  • 函数的输入和输出
  • 函数主体

我们可以像 SQL UDF(User Defined Function) 一样在 SQL 中调用协处理器，或者通过 HTTP API 调用。

SQL 实例

将复杂的分析用的 Python 代码（比如下面这个通过 cpu/mem/disk 使用率来确定负载状态的代码）保存到一个文件中（这里命名为 system_status.py）：

python

  1. @coprocessor(args=["host", "idc", "cpu_util", "memory_util", "disk_util"],
  2.              returns = ["host", "idc", "status"],
  3.              sql = "SELECT * FROM system_metrics")
  4. def system_status(hosts, idcs, cpus, memories, disks)\
  5.     -> (vector[str], vector[str], vector[str]):
  6.     statuses = []
  7.     for host, cpu, memory, disk in zip(hosts, cpus, memories, disks):
  8.         if cpu > 80 or memory > 80 or disk > 80:
  9.             statuses.append("red")
  10.             continue
  12.         status = cpu * 0.4 + memory * 0.4 + disk * 0.2
  14.         if status > 80:
  15.             statuses.append("red")
  16.         elif status > 50:
  17.             statuses.append("yello")
  18.         else:
  19.             statuses.append("green")
  22.     return hosts, idcs, statuses

上述代码根据 cpu/memory/disk 的使用情况来评估主机状态。参数来自于查询 system_metrics 的数据，由 @coprocessor 注释中的参数 sql 指定（这里是="SELECT * FROM system_metrics"）。查询结果被分配给 args=[...] 中的每个位置参数，然后函数返回三个变量，这些变量被转换为三个列 returns = ["host", "idc", "status"]。

提交 Python 脚本到 GreptimeDB

可以用 system_status 将文件提交给 GreptimeDB，这样以后就可以用这个名称来引用并执行它：

shell

  1. curl  --data-binary "@system_status.py" \
  2.    -XPOST "http://localhost:4000/v1/scripts?name=system_status&db=greptime.public"

运行该脚本：

shell

  1. curl  -XPOST \
  2.  "http://localhost:4000/v1/run-script?name=system_status&db=public"

以 json 格式获取结果：

json

  1. {
  2.   "code": 0,
  3.   "output": {
  4.     "records": {
  5.       "schema": {
  6.         "column_schemas": [
  7.           {
  8.             "name": "host",
  9.             "data_type": "String"
  10.           },
  11.           {
  12.             "name": "idc",
  13.             "data_type": "String"
  14.           },
  15.           {
  16.             "name": "status",
  17.             "data_type": "String"
  18.           }
  19.         ]
  20.       },
  21.       "rows": [
  22.         ["host1", "idc_a", "green"],
  23.         ["host1", "idc_b", "yello"],
  24.         ["host2", "idc_a", "red"]
  25.       ]
  26.     }
  27.   }
  28. }

更多有关 Python 协处理器的信息，请参考函数文档。