InfluxDB Flux关联MySQL表实现非匹配行返回unknown的方法

方案1：使用Flux的dict.get()字典映射（推荐，性能更优） #

该方案适合model映射表数据量不大的场景，不需要关联操作，执行效率远高于join。

// 1. 从MySQL读取全量model映射数据
mysqlModelData = sql.from(
  driverName: "mysql",
  dataSourceName: "你的MySQL连接串，格式为<用户名>:<密码>@tcp(<MySQL地址>:<端口>)/<库名>?parseTime=true",
  query: "SELECT model_name, model_desc FROM 你的MySQL表名"
)

// 2. 把查询结果转为 [model_name:model_desc] 格式的字典
modelMap = dict.fromList(
  pairs: mysqlModelData |> map(fn: (r) => ({key: r.model_name, value: r.model_desc}))
)

// 3. 查询InfluxDB时序数据
influxEventData = from(bucket: "你的InfluxDB Bucket名")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r._measurement == "你的测量名")
  // 按实际存储结构调整，把字段转为列
  |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
  |> keep(columns: ["_time", "username", "model_name", "last_connected"])

// 4. 匹配model_desc，未命中自动填充unknown
result = influxEventData
  |> map(fn: (r) => ({ r with
    model_desc: dict.get(dict: modelMap, key: r.model_name, default: "unknown")
  }))

result |> yield(name: "final_result")

方案2：通过join实现左外连接 #

如果你的model表数据量很大，不适合加载为全量字典，可以用该方案实现左外连接效果。

// 前置查询和方案1一致，分别拉取MySQL和InfluxDB数据
mysqlModelData = sql.from(
  driverName: "mysql",
  dataSourceName: "你的MySQL连接串",
  query: "SELECT model_name, model_desc FROM 你的MySQL表名"
) |> keep(columns: ["model_name", "model_desc"])

influxEventData = from(bucket: "你的InfluxDB Bucket名")
  |> range(start: v.timeRangeStart, stop: v.timeRangeStop)
  |> filter(fn: (r) => r._measurement == "你的测量名")
  |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
  |> keep(columns: ["_time", "username", "model_name", "last_connected"])

// 关联两张表，保留InfluxDB全量数据
joinResult = join(
  tables: {left: influxEventData, right: mysqlModelData},
  on: (r) => r.model_name,
  fn: (left, right) => ({left with model_desc: right.model_desc})
)

// 未匹配到的model_desc会为null，统一替换为unknown
result = joinResult |> fill(column: "model_desc", value: "unknown")

result |> yield(name: "final_result")

注意事项 #

• 需提前在InfluxDB的数据源配置中添加MySQL数据源，才可以通过sql.from直接读取MySQL数据
• 如果model表更新频率较高，可以给MySQL查询增加缓存配置，降低数据库访问压力

内容的提问来源于stack exchange，提问作者user17126302