收集到的隨時(shí)間變化的數(shù)據(jù)稱為時(shí)間序列數(shù)據(jù)。今天，它已經(jīng)成為每個(gè)行業(yè)和生態(tài)系統(tǒng)的一部分。它是不斷增長(zhǎng)的物聯(lián)網(wǎng)行業(yè)的一大組成部分，將成為人們?nèi)粘Ｉ畹闹匾糠?。但時(shí)間序列數(shù)據(jù)及其需求很難處理。這是因?yàn)闆]有專門為處理時(shí)間序列數(shù)據(jù)而構(gòu)建的工具。在這篇文章中，我將詳細(xì)介紹這些問題，以及過去 10 年來 InfluxData 如何解決這些問題。

InfluxData

InfluxData 是一個(gè)開源的時(shí)間序列數(shù)據(jù)庫(kù)平臺(tái)。你可能通過 ??InfluxDB?? 了解該公司，但你可能不知道它專門從事時(shí)間序列數(shù)據(jù)庫(kù)開發(fā)。這很重要，因?yàn)樵诠芾頃r(shí)間序列數(shù)據(jù)時(shí)，你要處理兩個(gè)問題：存儲(chǔ)生命周期和查詢。

在存儲(chǔ)生命周期中，開發(fā)人員通常首先收集和分析非常詳細(xì)的數(shù)據(jù)。但開發(fā)人員希望存儲(chǔ)較小的、降低采樣率的數(shù)據(jù)集，以描述其趨勢(shì)，而不占用太多的存儲(chǔ)空間。

查詢數(shù)據(jù)庫(kù)時(shí)，你不希望基于 ID 查詢數(shù)據(jù)，而是希望基于時(shí)間范圍進(jìn)行查詢。使用時(shí)間序列數(shù)據(jù)最常見的一件事是在一段時(shí)間內(nèi)對(duì)其進(jìn)行匯總。在典型的關(guān)系型數(shù)據(jù)庫(kù)中存儲(chǔ)數(shù)據(jù)時(shí)，這種查詢是很慢的，這種數(shù)據(jù)庫(kù)使用行和列來描述不同數(shù)據(jù)點(diǎn)的關(guān)系。專門為處理時(shí)間序列數(shù)據(jù)而設(shè)計(jì)的數(shù)據(jù)庫(kù)可以更快地處理這類查詢。InfluxDB 有自己的內(nèi)置查詢語(yǔ)言：Flux，這是專門為查詢時(shí)間序列數(shù)據(jù)集而構(gòu)建的。

Telegraf 如何工作的圖像

數(shù)據(jù)采集

數(shù)據(jù)采集和數(shù)據(jù)處理都有一些很棒的工具。InfluxData 有 12 個(gè)以上的客戶端庫(kù)，允許你使用自己選擇的編程語(yǔ)言編寫和查詢數(shù)據(jù)。這是自定義用法的一個(gè)很好的工具。開源攝取代理 Telegraf 包括 300 多個(gè)輸入和輸出插件。如果你是一個(gè)開發(fā)者，你也可以貢獻(xiàn)自己的插件。

InfluxDB 還可以接受上傳小體積歷史數(shù)據(jù)集的 CSV 文件，以及大數(shù)據(jù)集的批量導(dǎo)入。

    import math
    bicycles3 = from(bucket: "smartcity")
        |> range(start:2021-03-01T00:00:00z, stop: 2021-04-01T00:00:00z)
        |> filter(fn: (r) => r._measurement == "city_IoT")
        |> filter(fn: (r) => r._field == "counter")
        |> filter(fn: (r) => r.source == "bicycle")
        |> filter(fn: (r) => r.neighborhood_id == "3")
        |> aggregateWindow(every: 1h, fn: mean, createEmpty:false)
    bicycles4 = from(bucket: "smartcity")
        |> range(start:2021-03-01T00:00:00z, stop: 2021-04-01T00:00:00z)
        |> filter(fn: (r) => r._measurement == "city_IoT")
        |> filter(fn: (r) => r._field == "counter")
        |> filter(fn: (r) => r.source == "bicycle")
        |> filter(fn: (r) => r.neighborhood_id == "4")
        |> aggregateWindow(every: 1h, fn: mean, createEmpty:false)join(tables: {neighborhood_3: bicycles3, neighborhood_4: bicycles4}, on ["_time"], method: "inner")
        |> keep(columns: ["_time", "_value_neighborhood_3","_value_neighborhood_4"])
        |> map(fn: (r) => ({
            r with
            difference_value : math.abs(x: (r._value_neighborhood_3 - r._value_neighborhood_4))
        }))

Flux

Flux 是我們的內(nèi)部查詢語(yǔ)言，從零開始建立，用于處理時(shí)間序列數(shù)據(jù)。它也是我們一些工具的基礎(chǔ)動(dòng)力，包括 任務(wù)task、警報(bào)alert 和 通知notification。要剖析上面的 Flux 查詢，需要定義一些東西。首先，“桶bucket”就是我們所說的數(shù)據(jù)庫(kù)。你可以配置存儲(chǔ)桶，然后將數(shù)據(jù)流添加到其中。查詢會(huì)調(diào)用 ??smartcity?? 存儲(chǔ)桶，其范圍為特定的一天（準(zhǔn)確地說是 24 小時(shí)）。你可以從存儲(chǔ)桶中獲取所有數(shù)據(jù)，但大多數(shù)用戶都包含一個(gè)數(shù)據(jù)范圍。這是你能做的最基本的 Flux 查詢。

接下來，我添加過濾器，將數(shù)據(jù)過濾到更精確、更易于管理的地方。例如，我過濾分配給 id 為 3 的社區(qū)中的自行車數(shù)量。從那里，我使用 ??aggregateWindow?? 獲取每小時(shí)的平均值。這意味著我希望收到一個(gè)包含 24 列的表，每小時(shí)一列。我也對(duì) id 為 4 的社區(qū)進(jìn)行同樣的查詢。最后，我將這兩張表相疊加，得出這兩個(gè)社區(qū)自行車使用量的差異。

如果你想知道什么時(shí)候是交通高峰，這是不錯(cuò)的選擇。顯然，這只是 Flux 查詢功能的一個(gè)小例子。但它提供了一個(gè)很好的例子，使用了 Flux 附帶的一些工具。我還有很多的數(shù)據(jù)分析和統(tǒng)計(jì)功能。但對(duì)于這一點(diǎn)，我建議查看 Flux 文檔。

    import "influxdata/influxdb/tasks"
    option task = {name: PB_downsample, every: 1h, offset: 10s}
    from(bucket: "plantbuddy")
        |>range(start: tasks.lastSuccess(orTime: -task.every))
        |>filter(fn: (r) => r["_measurement"] == "sensor_data")
        |>aggregateWindow(every: 10m, fn:last, createEmpty:false)
        |>yield(name: "last")
        |>to(bucket: "downsampled")

任務(wù)

InfluxDB 任務(wù)task 是一個(gè)定時(shí) Flux 腳本，它接收輸入數(shù)據(jù)流并以某種方式修改或分析它。然后，它將修改后的數(shù)據(jù)存儲(chǔ)在新的存儲(chǔ)桶中或執(zhí)行其他操作。將較小的數(shù)據(jù)集存儲(chǔ)到新的存儲(chǔ)桶中，稱為“降采樣downsampling”，這是數(shù)據(jù)庫(kù)的核心功能，也是時(shí)間序列數(shù)據(jù)生命周期的核心部分。

你可以在當(dāng)前任務(wù)示例中看到，我已經(jīng)對(duì)數(shù)據(jù)進(jìn)行了降采樣。我得到每 10 分鐘增量的最后一個(gè)值，并將該值存儲(chǔ)在降采樣桶中。原始數(shù)據(jù)集在這 10 分鐘內(nèi)可能有數(shù)千個(gè)數(shù)據(jù)點(diǎn)，但現(xiàn)在降采樣桶只有 60 個(gè)新值。需要注意的一點(diǎn)是，我還使用了范圍內(nèi)的 ??lastSuccess?? 函數(shù)。這會(huì)告訴 InfluxDB 從上次成功運(yùn)行的時(shí)間開始運(yùn)行此任務(wù)，以防它在過去 2 小時(shí)內(nèi)失敗，在這種情況下，它可以追溯 3 個(gè)小時(shí)內(nèi)的最后一次成功運(yùn)行。這對(duì)于內(nèi)置錯(cuò)誤處理非常有用。

檢查和警報(bào)通知系統(tǒng)的圖像

檢查和警報(bào)

InfluxDB 包含一個(gè) 警報(bào)Alert 或 檢查Check 和 通知notification

許多人選擇設(shè)置通知。為此，你需要定義一個(gè) 通知端點(diǎn)notification endpoint。例如，聊天應(yīng)用程序可以進(jìn)行 HTTP 調(diào)用以接收通知。然后你定義何時(shí)接收通知，例如，你可以每小時(shí)運(yùn)行一次檢查。你可以每 24 小時(shí)運(yùn)行一次通知。你可以讓通知響應(yīng)值更改，例如，“WARN”更改為“CRITICAL”，或者當(dāng)值為“CRITICAL”時(shí)，無論如何都從“OK”更改為“WARN”。這是一個(gè)高度可定制的系統(tǒng)。從這個(gè)系統(tǒng)創(chuàng)建的 Flux 代碼也可以編輯。

新 Edge 功能的圖像

邊緣

最后，我想把所有的核心功能放在一起，包括最近發(fā)布的一個(gè)非常特別的新功能。“Edge to cloud” 是一個(gè)非常強(qiáng)大的工具，允許你運(yùn)行開源 InfluxDB，并在出現(xiàn)連接問題時(shí)在本地存儲(chǔ)數(shù)據(jù)。連接修復(fù)后，它會(huì)將數(shù)據(jù)流傳輸?shù)?InfluxData 云平臺(tái)。

這對(duì)于邊緣設(shè)備和重要數(shù)據(jù)非常重要，因?yàn)槿魏螖?shù)據(jù)丟失都是有害的。你定義一個(gè)要復(fù)制到云的存儲(chǔ)桶，然后該存儲(chǔ)桶有一個(gè)磁盤支持的隊(duì)列來本地存儲(chǔ)數(shù)據(jù)。然后定義云存儲(chǔ)桶應(yīng)該復(fù)制到的內(nèi)容。在連接到云端之前，數(shù)據(jù)都存儲(chǔ)在本地。

InfluxDB 和物聯(lián)網(wǎng)邊緣

假設(shè)你有一個(gè)項(xiàng)目，你想使用連接到植物上的物聯(lián)網(wǎng)傳感器 ??監(jiān)測(cè)家里植物的健康狀況??。該項(xiàng)目是使用你的筆記本電腦作為邊緣設(shè)備設(shè)置的。當(dāng)你的筆記本電腦合上或關(guān)閉時(shí)，它會(huì)在本地存儲(chǔ)數(shù)據(jù)，然后在重新連接時(shí)將數(shù)據(jù)流傳到我的云存儲(chǔ)桶。

圖片展示了 Plant buddy 的工作方式

需要注意的一點(diǎn)是，在將數(shù)據(jù)存儲(chǔ)到復(fù)制桶之前，這會(huì)對(duì)本地設(shè)備上的數(shù)據(jù)進(jìn)行降采樣。你的植物傳感器每秒提供一個(gè)數(shù)據(jù)點(diǎn)。但它將數(shù)據(jù)壓縮為一分鐘的平均數(shù)，因此存儲(chǔ)的數(shù)據(jù)更少了。在云賬戶中，你可以添加一些警報(bào)和通知，讓你知道植物的水分何時(shí)低于某個(gè)水平，需要澆水。也可以在網(wǎng)站上使用視覺效果來告訴用戶植物的健康狀況。

數(shù)據(jù)庫(kù)是許多應(yīng)用程序的主干。在像 InfluxDB 的時(shí)間序列數(shù)據(jù)庫(kù)平臺(tái)中使用帶有時(shí)間戳的數(shù)據(jù)可以節(jié)省開發(fā)人員的時(shí)間，并使他們能夠訪問各種工具和服務(wù)。InfluxDB 的維護(hù)者喜歡看到人們?cè)谖覀兊拈_源社區(qū)中構(gòu)建什么，所以請(qǐng)與我們聯(lián)系，并與其他人共享你的項(xiàng)目和代碼！