查询 SQL 数据源

这个 Flux sql 包提供了用于处理 SQL 数据源的函数。 sql.from() 让你可以查询 SQL 数据源，如 PostgreSQL、MySQL、Snowflake、SQLite、Microsoft SQL Server、Amazon Athena 和 Google BigQuery，并将结果用于 InfluxDB 仪表板、任务和其他操作。

• 查询一个SQL数据源
• 将SQL数据与InfluxDB中的数据连接
• 使用SQL结果填充仪表板变量
• 使用秘密存储SQL数据库凭证
• 示例传感器数据

如果您刚刚开始使用Flux查询，请查看以下内容：

• 开始使用Flux 以获取Flux及Flux查询部分的概念概述。
• 执行查询 以发现多种运行查询的方法。

查询一个 SQL 数据源

要查询SQL数据源：

1. 在您的Flux查询中导入sql包
2. 使用 sql.from() 函数来指定驱动程序、数据源名称 (DSN) 和用于从您的 SQL 数据源查询数据的查询：

import "sql"

sql.from(
    driverName: "postgres",
    dataSourceName: "postgresql://user:password@localhost",
    query: "SELECT * FROM example_table",
)

import "sql"

sql.from(
    driverName: "mysql",
    dataSourceName: "user:password@tcp(localhost:3306)/db",
    query: "SELECT * FROM example_table",
)

import "sql"

sql.from(
    driverName: "snowflake",
    dataSourceName: "user:password@account/db/exampleschema?warehouse=wh",
    query: "SELECT * FROM example_table",
)

// NOTE: InfluxDB OSS and InfluxDB Cloud do not have access to
// the local filesystem and cannot query SQLite data sources.
// Use the Flux REPL to query an SQLite data source.

import "sql"

sql.from(
    driverName: "sqlite3",
    dataSourceName: "file:/path/to/test.db?cache=shared&mode=ro",
    query: "SELECT * FROM example_table",
)

import "sql"

sql.from(
    driverName: "sqlserver",
    dataSourceName: "sqlserver://user:password@localhost:1234?database=examplebdb",
    query: "GO SELECT * FROM Example.Table",
)

import "sql"
sql.from(
    driverName: "awsathena",
    dataSourceName: "s3://myorgqueryresults/?accessID=12ab34cd56ef&region=region-name&secretAccessKey=y0urSup3rs3crEtT0k3n",
    query: "GO SELECT * FROM Example.Table",
)

import "sql"
sql.from(
    driverName: "bigquery",
    dataSourceName: "bigquery://projectid/?apiKey=mySuP3r5ecR3tAP1K3y",
    query: "SELECT * FROM exampleTable",
)

请查看sql.from() 文档以获取有关必需函数参数的信息。

将SQL数据与InfluxDB中的数据进行连接

从InfluxDB查询SQL数据源的主要好处之一是能够用存储在InfluxDB之外的数据来丰富查询结果。

使用下面的 空气传感器样本数据，以下查询将存储在 InfluxDB 中的空气传感器指标与存储在 PostgreSQL 中的传感器信息进行连接。连接的数据使您能够根据未存储在 InfluxDB 中的传感器信息查询和过滤结果。

// Import the "sql" package
import "sql"

// Query data from PostgreSQL
sensorInfo = sql.from(
    driverName: "postgres",
    dataSourceName: "postgresql://localhost?sslmode=disable",
    query: "SELECT * FROM sensors",
)

// Query data from InfluxDB
sensorMetrics = from(bucket: "example-bucket")
    |> range(start: -1h)
    |> filter(fn: (r) => r._measurement == "airSensors")

// Join InfluxDB query results with PostgreSQL query results
join(tables: {metric: sensorMetrics, info: sensorInfo}, on: ["sensor_id"])

使用SQL结果填充仪表板变量

使用 sql.from() 从 SQL 查询结果中 创建仪表板变量。 以下示例使用下面的 空气传感器示例数据 创建一个变量，让您选择传感器的位置。

import "sql"

sql.from(
    driverName: "postgres",
    dataSourceName: "postgresql://localhost?sslmode=disable",
    query: "SELECT * FROM sensors",
)
    |> rename(columns: {location: "_value"})
    |> keep(columns: ["_value"])

使用变量在您的仪表板中操作查询。

使用秘密存储SQL数据库凭据

如果您的 SQL 数据库需要身份验证，请使用 InfluxDB 秘密 存储和填充连接凭据。默认情况下，InfluxDB 以 base64 编码并将秘密存储在其内部键值存储中，BoltDB。为了提高安全性，将秘密存储在 Vault。

将您的数据库凭据存储为机密

使用InfluxDB API或influx CLI将您的数据库凭据存储为秘密。

curl --request PATCH http://localhost:8086/api/v2/orgs/<org-id>/secrets \
  --header 'Authorization: Token YOURAUTHTOKEN' \
  --header 'Content-type: application/json' \
  --data '{
  "POSTGRES_HOST": "http://example.com",
  "POSTGRES_USER": "example-username",
  "POSTGRES_PASS": "example-password"
}'

# Syntax
influx secret update -k <secret-key>

# Example
influx secret update -k POSTGRES_PASS

influx secret update -k <secret-key> -v <secret-value>

在查询中使用秘密

导入 influxdata/influxdb/secrets 包，并使用 字符串插值 在您的 Flux 查询中填充存储的秘密作为连接凭据。

import "sql"
import "influxdata/influxdb/secrets"

POSTGRES_HOST = secrets.get(key: "POSTGRES_HOST")
POSTGRES_USER = secrets.get(key: "POSTGRES_USER")
POSTGRES_PASS = secrets.get(key: "POSTGRES_PASS")

sql.from(
    driverName: "postgres",
    dataSourceName: "postgresql://${POSTGRES_USER}:${POSTGRES_PASS}@${POSTGRES_HOST}",
    query: "SELECT * FROM sensors",
)

示例传感器数据

该空气传感器样本数据和样本传感器信息模拟了一组在建筑物各个房间内测量温度、湿度和一氧化碳的传感器。每一个收集到的数据点都存储在InfluxDB中，并带有一个sensor_id标签，标识数据来自于哪个特定的传感器。样本传感器信息存储在PostgreSQL中。

示例数据包括：

• 从每个传感器收集的模拟数据存储在InfluxDB中的airSensors测量中：

  • temperature
  • humidity
  • co

• 关于存储在PostgreSQL中的sensors表的每个传感器的信息：

  • sensor_id
  • location
  • model_number
  • last_inspected

下载样本空气传感器数据

1. 创建一个存储数据的桶。

2. 创建一个 InfluxDB 任务 并使用sample.data() 函数 每 15 分钟下载一次示例空气传感器数据。 将下载的示例数据写入您的新桶：

   import "influxdata/influxdb/sample"
   
   option task = {name: "Collect sample air sensor data", every: 15m}
   
   sample.data(set: "airSensor")
       |> to(org: "example-org", bucket: "example-bucket")
   

3. 查询您的目标桶 在第一次任务运行后，以确保示例数据成功写入。

   from(bucket: "example-bucket")
       |> range(start: -1m)
       |> filter(fn: (r) => r._measurement == "airSensors")
   

导入示例传感器信息

1. 下载并安装 PostgreSQL。

2. 下载示例传感器信息CSV。

   下载示例传感器信息

3. 使用 PostgreSQL 客户端 (psql 或 GUI) 创建 sensors 表：

   CREATE TABLE sensors (
     sensor_id character varying(50),
     location character varying(50),
     model_number character varying(50),
     last_inspected date
   );
   

4. 导入下载的 CSV 示例数据。将 FROM 文件路径更新为下载的 CSV 示例数据的路径。

   COPY sensors(sensor_id,location,model_number,last_inspected)
   FROM '/path/to/sample-sensor-info.csv' DELIMITER ',' CSV HEADER;
   

5. 查询表以确保数据正确导入：

   SELECT * FROM sensors;
   

导入示例数据仪表板

下载并导入空气传感器仪表板以可视化生成的数据：

查看空气传感器仪表盘 JSON

有关导入仪表板的信息，请参见 Create a dashboard。