使用 Flux 连接数据

memUsed = from(bucket: "db/rp")
    |> range(start: -5m)
    |> filter(fn: (r) => r._measurement == "mem" and r._field == "used")

procTotal = from(bucket: "db/rp")
    |> range(start: -5m)
    |> filter(fn: (r) => r._measurement == "processes" and r._field == "total")

join(tables: {mem: memUsed, proc: procTotal}, on: ["_time", "_stop", "_start", "host"])

join(tables: {mem: memUsed, proc: procTotal}, on: ["_time", "_stop", "_start", "host"])
    |> map(fn: (r) => ({_time: r._time, _value: r._value_mem / r._value_proc}))

batchSize = (cluster_id, start=-1m, interval=10s) => {
    httpd = from(bucket: "telegraf")
        |> range(start: start)
        |> filter(fn: (r) => r._measurement == "influxdb_httpd" and r._field == "writeReq" and r.cluster_id == cluster_id)
        |> aggregateWindow(every: interval, fn: mean)
        |> derivative(nonNegative: true, unit: 60s)

    write = from(bucket: "telegraf")
        |> range(start: start)
        |> filter(fn: (r) => r._measurement == "influxdb_write" and r._field == "pointReq" and r.cluster_id == cluster_id)
        |> aggregateWindow(every: interval, fn: max)
        |> derivative(nonNegative: true, unit: 60s)

    return join(tables: {httpd: httpd, write: write}, on: ["_time", "_stop", "_start", "host"])
        |> map(fn: (r) => ({_time: r._time, _value: r._value_httpd / r._value_write}))
        |> group(columns: cluster_id)
}

batchSize(cluster_id: "enter cluster id here")