InfluxDB 2.0 Alpha 14 中引入了多数据源 Flux

作者：Nate Isley / 产品，开发者
2019年7月18日

正如我们的联合创始人 Paul 在InfluxDB 2.0 Alpha 版本发布公告中广泛讨论的那样，我们相信 Flux 将在将时间序列数据与许多不同数据源结合方面发挥重要作用。Flux 的第一个官方多数据存储步骤是 InfluxDB 2.0 Alpha 14 的 from SQL 发布。鉴于数据存储越来越多以及需要满足这些需求，这一 Flux 添加功能具有重要意义。

这是一个多数据存储的世界

几年前，在 APM 行业爆炸性增长的初期，我是 Wily Technology 的顾问，处理了数百个处于不同困境中的应用程序。一次又一次，我看到应用程序忽略了它们特定数据的本质，这不可避免地导致数据存储被滥用，而这些数据存储根本不适合这种方式。

随着不同的应用程序数据集的扩展，它们被赋予了使其适合特定数据存储处理器的特征。使用为它们专门构建的工具和存储机制来处理、存储和处理每个数据集，可以构建出既易于维护又可随增长而扩展的优质系统。

以下是一些不同数据集和数据存储类型的示例

• 像电子商务商店的库存、客户订单和客户数据及其关系这样的关系型数据最好由一个旨在建模这种关系的 RDBMS 系统来处理。
• 与物联网设备相关的资产信息在 IBM 的 Maximo 和 SAP 的智能资产管理等专用数据存储中取得了巨大成功。资产元数据可以通过这些应用程序公开的集成 API 获取。
• 如业务流程、集成应用程序和底层网络拓扑这样的依赖数据最好由一个[图数据库](https://influxdb.org.cn/graph-database/)来处理。
• 如来自堆栈、传感器和系统的数百万个流数据点这样的时间序列数据最好由一个时间序列数据库来处理。

在遥远的过去，有人认为使用不同的数据存储会给程序员带来太大的学习压力，需要学习不同的数据工具。在今天的广泛和统一的API使用普及之前，这可能确实是如此。但如今，开发者们使用数据源驱动程序和API从不同地方拉取数据并将其整合在一起，已经变得非常轻松。

依赖开发者来处理和压缩数据对于专门定制解决特定问题的软件来说非常有效。但开发者永远不够用，通用型数据分析和可视化工具已经证明了对许多分析师和数据科学家具有价值。单数据源分析工具有时也能工作，但分析师们通常希望将来自不同数据存储的数据合并在一起，并经常发现将数据源整合在一起是一项艰巨的任务。

提供一种灵活地合并数据源并跨时间进行分析的工具是Flux的主要用途之一。开源的Flux旨在赋予每个查询和可视化工具以力量，以便它们可以使用通用、强大和统一的语言将相关数据集汇集在一起，从而生成洞察力。

使用来自SQL数据存储的时间序列数据

Flux首次实现从非InfluxDB数据存储中提取数据的函数是InfluxDB 2.0 Alpha 14的SQL.from()函数。让我们深入探讨一个具体的物联网用例来了解这个函数。

物联网资产数据存储

最近，我多次与使用InfluxDB在物联网用例中的客户进行了深入研究。这些物联网实践者有成千上万的设备和传感器，将它们捕获的所有数据都流式传输到InfluxDB，处理所有这些数据带来了独特的挑战。

没有连接不同数据存储的方法，许多物联网实践者将所有元数据作为时间序列指标的标签重复使用，以便他们可以进行基于元数据的时间序列分析，例如

“在过去一天中，每种设备类型的错误率是多少？”

当然，在SQL.from()之前，唯一让InfluxDB回答这个问题的方法是将设备类型作为标签添加到来自其传感器的每个时间序列指标上。

这可以工作，但存在随着时间的推移对元数据集变化反应不足的问题。例如，如果一个物联网设备制造商最初将时间序列指标标记为某种安全分类，但分类发生了变化，则只有在使用新分类进行查询时才会显示新数据，因为旧数据具有旧的标签值。

更好的解决方案是通过保持元数据在关系型存储中，并将时间序列数据存储在InfluxDB中，来对数据进行清晰的分离。使两个数据存储解决方案能够工作的是一种可以无缝查询时间序列数据和关系型数据的Flux语言。

import "sql"

phBalances = from(bucket: "PH_Balances")
  |> range(start: now-60)
  |> filter(fn: (r) => r._measurement == "water_sensor" )

deviceIds = sql.from(driverName: "mysql", dataSourceName: "username:password@tcp(host)/dbname", query: "SELECT customer_name, customer_location, device_id AS device FROM DeviceTable WHERE customer_name = 'InfluxData'")

join(tables: {t1: phBalances, t2: deviceIds}, on: ["device"])
	|> aggregateWindow(every: v.windowPeriod, fn: mean)
 	 |> yield(name: "mean")