在 influxdb 中为聚合数据添加标签

Posted 2023-02-15

【中文标题】在 influxdb 中为聚合数据添加标签

【英文标题】：Adding a tag to aggregated data in influxdb

【发布时间】：2020-07-16 01:33:45

【问题描述】：

我在处理SELECT INTO 查询时无法理解如何向数据系列添加标签。我有一个 NYTimes COVID dataset 的 Influxdb，我使用 cases 和 deaths 字段作为 fields 和 state 和 county 信息作为标签。

我可以在这样的查询中汇总来自邻近县的数据：

SELECT sum("cases") AS "cases" FROM "ny_covid"."autogen"."value" WHERE ("state"='Pennsylvania') AND ("county"='Philadelphia' OR "county"='Delaware') GROUP BY time(1d) FILL(null)

这非常有效。但我想将这些聚合数据保存到一个新数据库中以进行其他查询。我可以这样做：

SELECT sum("cases") AS "cases" INTO "new_covid"."autogen"."value" FROM "ny_covid"."autogen"."value" WHERE ("state"='Pennsylvania') AND ("county"='Philadelphia' OR "county"='Delaware') GROUP BY time(1d) FILL(null)

我的问题是，如何将location=Philly 之类的标签添加到我刚刚插入new_covid 数据库的数据中？因为，我想做一些其他位置级别的聚合，似乎tag 是保持这些值不同的方法。

SELECT sum("cases") AS "cases" INTO "new_covid"."autogen"."value" FROM "ny_covid"."autogen"."value" WHERE ("state"='Pennsylvania') AND ("county"='Dauphin' OR "county"='Lancaster') GROUP BY time(1d) FILL(null)

我所做的所有搜索都是关于在查询中使用标签或在跨数据库复制时保留它们。但是我找不到任何关于在SELECT INTO 类型语句中附加标签的信息。

【问题讨论】：

您真的需要创建新标签吗？为什么不能重复使用现有标签state,county？

@JanGaraj 因为我想将 6 个县的 sum 的标签添加为新数据库中的新区域。这样，我可以通过标签将它们从new_covid 数据库中拉出，并将其与匹兹堡周围的 7 个县 sum 或哈里斯堡周围的 8 个县区分开来。我也不知道如何将标签从以前的数据库转移到新的数据库中。这也是一个普遍的问题：“如何为使用SELECT INTO 语句插入的数据添加标签”？

好的，那么您需要在应用程序/自己的代码级别执行此操作：select data; add new tag to data; insert data;。使用您喜欢的具有 InfluxDB 支持的语言（例如 python）并对其进行编码。

在做SELECT INTO语句时无法添加标签真的是答案吗？添加标签的唯一方法是从一个数据库中提取数据，然后重新插入。相反，在SELECT INTO 语句中区分聚合的唯一方法是将它们发送到不同的fields?

不，您还有其他选择。如：加入InfluxDB，说服团队这是一个重要的特性，所以会实现/fork源码，自己实现。 InfluxQL 不是 SQL，我不认为这个特性在时间序列世界中被广泛使用。使用更好的数据库，这将满足您的需求。也许 InfluxDB 2.0（仍然是测试版）会更好——它支持多数据存储世界——你也可以连接关系数据库 = 也许你可以将它用作数据转换的“临时”存储，在那里你可以使用旧的SQL。

【参考方案1】：

您可以使用TICKscript language 轻松获得所需的功能，这是 TICK 堆栈中 InfluxDB 的免费部分。 但是，除了 InfluxDB 本身之外，这还需要使用 Kapacitor 和 Chronograf。

Kapacitor 将执行用 TICKscript 语言编写的任务来操作数据。 Chronograf 提供了一个 Web UI 来配置 InfluxDB 和 Kapacitor。

从您的问题中查找 InfluxQL 查询的 TICKscript 任务示例：

var b1 = batch
    |query('''
    SELECT sum("cases") AS "cases" 
    FROM "ny_covid"."autogen"."value" 
    WHERE ("state"='Pennsylvania') AND ("county"='Philadelphia' OR "county"='Delaware') )
    ''')
        .offset(1d)
        .period(1d)
        .cron('@midnight')
        .groupBy(time(1d))
    // Write the transformed data to InfluxDB
    |influxDBOut()
        .database('new_covid')
        .retentionPolicy('autogen')
        .measurement('value')
        .tag('location', 'Philly')

var b2 = batch
    |query('''
    SELECT sum("cases") AS "cases" 
    FROM "ny_covid"."autogen"."value" 
    WHERE ("state"='Pennsylvania') AND ("county"='Dauphin' OR "county"='Lancaster')
    ''')
        .offset(1d)
        .period(1d)
        .cron('@midnight')
        .groupBy(time(1d))
    // Write the transformed data to InfluxDB
    |influxDBOut()
        .database('new_covid')
        .retentionPolicy('autogen')
        .measurement('value')
        .tag('location', 'Dauphy')

这也不是获得这种结果的一种方法。您也可以检查stream approach 而不是上面的batch used。

运行此代码所需要做的就是为 kapacitor 创建新任务：

Chronograf-> Alerting tab -> Manage tasks -> TICKscripts 部分中的“+Write TICKscript”按钮。

要获得完整的 TICKstack 环境，您可以使用 offitial sandbox on dockers