定义函数

协处理器注释

@coprocesssor 注释指定一个 python 函数作为 GreptimeDB 的协处理器,并为其设置一些属性。

该引擎允许一个且仅有一个用 @coprocesssor 注释的函数,不能在一个脚本中拥有一个以上的协处理器。

ParameterDescriptionExample
sqlOptional. The SQL statement that the coprocessor function will query data from the database and assign them to input args.@copr(sql=”select * from cpu”, ..)
argsOptional. The argument names that the coprocessor function will be taken as input, which are the columns in query results by sql.@copr(args=[“cpu”, “mem”], ..)
returnsThe column names that the coprocessor function will return. The Coprocessor Engine uses it to generate the output schema.@copr(returns=[“add”, “sub”, “mul”, “div”], ..)
backendOptional. The coprocessor function will run on available engines like rspy and pyo3, which are associated with RustPython Backend and CPython Backend respectively. The default engine is set to rspy.@copr(backend=”rspy”, ..)

sqlargs 都是可选的;它们必须都可用,或都不可用,并通常在后查询处理中被使用,详情请阅读下文。

returns 是每个 coprocessor 都需要的,因为输出模式必然存在。

因为 RustPython 不能支持 C 语言 API,当尝试使用 pyo3 后端来使用只支持 C 语言 API 的第三方 python 库时,例如 numpypandas 等,backend 则是必要的。

协处理器函数的输入

该协处理器也接受之前已经看到的参数:

python

  1. @coprocessor(args=["number"], sql="select number from numbers limit 20", returns=["value"])
  2. def normalize(v) -> vector[i64]:
  3. return [normalize0(x) for x in v]

参数 v 是执行 sql 返回的查询结果中的 number 列(由 args 属性指定)。

当然,也可以有多个参数:

python

  1. @coprocessor(args=["number", "number", "number"],
  2. sql="select number from numbers limit 5",
  3. returns=["value"])
  4. def normalize(n1, n2, n3) -> vector[i64]:
  5. # returns [0,1,8,27,64]
  6. return n1 * n2 * n3

除了 args,还可以向协处理器传递用户定义的参数:

python

  1. @coprocessor(returns=['value'])
  2. def add(**params) -> vector[i64]:
  3. a = params['a']
  4. b = params['b']
  5. return int(a) + int(b)

然后从 HTTP API 传递 ab

sh

  1. curl -XPOST \
  2. "http://localhost:4000/v1/run-script?name=add&db=public&a=42&b=99"

json

  1. {
  2. "code": 0,
  3. "output": [
  4. {
  5. "records": {
  6. "schema": {
  7. "column_schemas": [
  8. {
  9. "name": "value",
  10. "data_type": "Int64"
  11. }
  12. ]
  13. },
  14. "rows": [
  15. [
  16. 141
  17. ]
  18. ]
  19. }
  20. }
  21. ],
  22. "execution_time_ms": 0
  23. }

a=42&b=99 作为查询参数传入HTTP API,返回结果 141

用户定义的参数必须由协处理器中的 **kwargs 来完成,其类型都是字符串。可以传递任何想要的东西,如在协处理器中运行的 SQL。

协处理器函数的输出

正如前面的例子所展示的那样,协处理器函数的输出必须是向量。

We can return multi vectors:

python

  1. from greptime import vector
  2. @coprocessor(returns=["a", "b", "c"])
  3. def return_vectors() -> (vector[i64], vector[str], vector[f64]):
  4. a = vector([1, 2, 3])
  5. b = vector(["a", "b", "c"])
  6. c = vector([42.0, 43.0, 44.0])
  7. return a, b, c

函数 return_vectors 的返回类型是 (vector[i64], vector[str], vector[f64])

但必须确保所有这些由函数返回的向量具有相同的长度,因为当它们被转换为行时,每一行必须呈现所有的列值。

当然,可以返回字面值,它们也会被转化为向量:

python

  1. from greptime import vector
  2. @coprocessor(returns=["a", "b", "c"])
  3. def return_vectors() -> (vector[i64], vector[str], vector[i64]):
  4. a = 1
  5. b = "Hello, GreptimeDB!"
  6. c = 42
  7. return a, b, c

HTTP API

/scripts 提交一个 Python 脚本到 GreptimeDB。

保存一个 Python 脚本,如 test.py

python

  1. @coprocessor(args = ["number"],
  2. returns = [ "number" ],
  3. sql = "select number from numbers limit 5")
  4. def square(number) -> vector[i64]:
  5. return number * 2

将其提交到数据库:

shell

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

json

  1. {"code":0}

Python 脚本被插入到 scripts 表中并被自动编译:

shell

  1. curl -G http://localhost:4000/v1/sql --data-urlencode "sql=select * from scripts"

json

  1. {
  2. "code": 0,
  3. "output": [{
  4. "records": {
  5. "schema": {
  6. "column_schemas": [
  7. {
  8. "name": "schema",
  9. "data_type": "String"
  10. },
  11. {
  12. "name": "name",
  13. "data_type": "String"
  14. },
  15. {
  16. "name": "script",
  17. "data_type": "String"
  18. },
  19. {
  20. "name": "engine",
  21. "data_type": "String"
  22. },
  23. {
  24. "name": "timestamp",
  25. "data_type": "TimestampMillisecond"
  26. },
  27. {
  28. "name": "gmt_created",
  29. "data_type": "TimestampMillisecond"
  30. },
  31. {
  32. "name": "gmt_modified",
  33. "data_type": "TimestampMillisecond"
  34. }
  35. ]
  36. },
  37. "rows": [
  38. [
  39. "default",
  40. "square",
  41. "@coprocessor(args = [\"number\"],\n returns = [ \"number\" ],\n sql = \"select number from numbers\")\ndef square(number):\n return number * 2\n",
  42. "python",
  43. 0,
  44. 1676032587204,
  45. 1676032587204
  46. ]
  47. ]
  48. }
  49. }],
  50. "execution_time_ms": 4
  51. }

也可以通过 /run-script 执行脚本:

shell

  1. curl -XPOST -G "http://localhost:4000/v1/run-script?db=default&name=square"

json

  1. {
  2. "code": 0,
  3. "output": [{
  4. "records": {
  5. "schema": {
  6. "column_schemas": [
  7. {
  8. "name": "number",
  9. "data_type": "Float64"
  10. }
  11. ]
  12. },
  13. "rows": [
  14. [
  15. 0
  16. ],
  17. [
  18. 2
  19. ],
  20. [
  21. 4
  22. ],
  23. [
  24. 6
  25. ],
  26. [
  27. 8
  28. ]
  29. ]
  30. }
  31. }],
  32. "execution_time_ms": 8
  33. }

Python 脚本的参数和结果

/scripts 接受指定数据库的查询参数 db,以及命名脚本的 name/scripts 处理 POST 方法主体来作为脚本文件内容。

/run-script 通过 dbname 运行编译好的脚本,然后返回输出,这与 /sql API 中的查询结果相同。

/run-script 也接收其他查询参数,作为传递到协处理器的用户参数,参考输入和输出