InfluxDB v2.0如何将field转换为tag？Flux语言实现方案求助

核心实现逻辑依赖Flux to() 函数的tagColumns参数，该参数可以手动指定哪些列作为tag写入新measurement，其他非内置列会自动识别为field写入，具体修改方案如下：

如果你只需要转换单个field为tag，可直接在原有查询基础上调整：

from(bucket: "bucket_name")
    |> range(start: -10y)
    |> filter(fn: (r) => r._measurement == "cu_om")
    |> aggregateWindow(every: 5s, fn: last, createEmpty: false)
    // 将需要转成tag的field值复制为独立列，as后填写你期望的tag名
    |> duplicate(column: "_value", as: "your_tag_name")
    |> set(key: "_measurement", value: "cu_om_new1")
    // 写入时指定该列为tag列
    |> to(
        org: "org_name", 
        bucket: "bucket_name",
        tagColumns: ["your_tag_name"]
    )

如果你有多个field需要同时处理，需要先通过pivot把field展开为同级列再操作：

from(bucket: "bucket_name")
    |> range(start: -10y)
    |> filter(fn: (r) => r._measurement == "cu_om")
    |> aggregateWindow(every: 5s, fn: last, createEmpty: false)
    // 把所有field展开为同级列
    |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value")
    |> set(key: "_measurement", value: "cu_om_new1")
    // 所有要转成tag的列都放在tagColumns数组中
    |> to(
        org: "org_name", 
        bucket: "bucket_name",
        tagColumns: ["tag1", "tag2"]
    )

注意事项 #

• 数据量较大时不要一次性拉取全量历史数据，建议按时间分片执行查询写入，避免内存溢出
• 写入完成后可先查询新measurement的schema确认tag配置正确，再清理旧measurement的数据

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