如何将 CSV 中的数据点写入 InfluxDB v2 和 InfluxDB Cloud

作者：Anais Dotis-Georgiou / 产品, 用例, 开发者
2021 年 7 月 29 日

有几种方法可以将 CSV 写入 InfluxDB v2 和 InfluxDB Cloud，包括

• Telegraf 文件插件
• CLI
• csv.from() Flux 函数
• 任何客户端库

我将演示如何使用这些方法将 CSV 写入 InfluxDB Cloud Free Tier 帐户，因为注册 Free Tier 帐户是开始使用 InfluxDB Cloud 的最快方法。但是，该过程对于InfluxDB v2 OSS 是相同的。对于上面列表中的每一种写入方法，我们将把以下 CSV 写入 InfluxDB Cloud

"
_time,region,host,usage_user,_measurement
2018-05-08T20:50:00Z,east,A,15.43,cpu
2018-05-08T20:50:20Z,east,A,59.25,cpu
2018-05-08T20:50:40Z,east,A,52.62,cpu
"

使用 Telegraf 文件插件将 CSV 写入 InfluxDB Cloud

Telegraf 是 InfluxData 的插件驱动服务器代理，用于收集和报告指标。有超过 200 个输入插件，这意味着有很多方法可以将数据导入 InfluxDB。但是，我经常看到新的 Influx 用户在 InfluxData 社区站点上询问如何将 CSV 中的数据点写入 InfluxDB。从 CSV 文件写入数据点是将熟悉的数据插入 InfluxDB 的一种简单方法，这可以使您更容易熟悉该平台。

首先，我需要下载一个 Telegraf 配置文件，其中包含适当的输入和输出插件。根据 Telegraf 入门 文档，我将在终端中选择的目录中使用以下命令。

telegraf -sample-config -input-filter file -output-filter influxdb_v2 > writeCSV.conf

• -sample-config 标志将生成 telegraf 配置文件。
• -input-filter 和 -output-filter 标志分别指定数据的输入和输出源。
• 符号 > 后面的文本命名配置文件。

运行命令后，我打开 writeCSV.conf。我的 telegraf 配置已完成，包含文件输入插件和 InfluxDB v2 输出插件。

更改 telegraf 配置的输入部分，使其如下所示

[[inputs.file]]
  ## Files to parse each interval.  Accept standard unix glob matching rules,
  ## as well as ** to match recursive files and directories.
  files = ["/path/to/writeCSV.csv"]

  ## Data format to consume.
  ## Each data format has its own unique set of configuration options, read
  ## more about them here:
  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md

## Data format to consume.
data_format = "csv"

## Indicates how many rows to treat as a header. By default, the parser assumes
## there is no header and will parse the first row as data. If set to anything more
## than 1, column names will be concatenated with the name listed in the next header row.
## If `csv_column_names` is specified, the column names in header will be overridden.
csv_header_row_count = 1

## Indicates the number of rows to skip before looking for header information.
 csv_skip_rows = 1

## Indicates the number of columns to skip before looking for data to parse.
## These columns will be skipped in the header as well.
 csv_skip_columns = 1
 
## The character reserved for marking a row as a comment row
 ## Commented rows are skipped and not parsed
csv_comment = "#"

 ## The column to extract the name of the metric from
 csv_measurement_column = "_measurement"

 ## Columns listed here will be added as tags. Any other columns
 ## will be added as fields.
csv_tag_columns = ["host","region"]

 ## The column to extract time information for the metric
 ## `csv_timestamp_format` must be specified if this is used
csv_timestamp_column = "_time"

 ## The format of time data extracted from `csv_timestamp_column`
 ## this must be specified if `csv_timestamp_column` is specified
 csv_timestamp_format = "2006-01-02T15:04:05Z07:00"

您需要收集以下帐户信息以配置 telegraf 配置的输出部分

• 存储桶。存储桶是数据存储的位置。您可以将其视为数据库。所有存储桶都有保留策略。您在创建存储桶时为存储桶分配保留策略。创建存储桶的最简单方法是通过 InfluxDB Cloud 用户界面 (UI)，如 此处 所述。但是，您也可以使用 CLI 或 API 创建存储桶。此存储桶将被称为我们的目标存储桶“mybucket”。
• 令牌：InfluxDB 身份验证令牌确保用户及其数据的安全性。令牌属于组织并描述权限。您需要创建一个作用域限定为目标存储桶的读/写或全访问令牌。创建存储桶的最简单方法是通过 InfluxDB Cloud UI，如 此处 所述。但是，您也可以使用 CLI 或 API 创建令牌。
• 组织：您的组织将是您用于设置 InfluxDB Cloud 帐户的电子邮件。

更改 telegraf 配置的输出部分，使其如下所示

# Configuration for sending metrics to InfluxDB 2.0
[[outputs.influxdb_v2]]
  ## The URLs of the InfluxDB cluster nodes.
  ##
  ## Multiple URLs can be specified for a single cluster, only ONE of the
  ## urls will be written to each interval.
  ##   ex:  urls = ["http://127.0.0.1:8086"] if your using OSS
  urls = ["https://us-west-2-1.aws.cloud2.influxdata.com"]
 
  ## Token for authentication.
  token = "mytoken"
 
  ## Organization is the name of the organization you wish to write to.
  organization = "[email protected]"
 
  ## Destination bucket to write into.
  bucket = "mybucket"

要测试您的 telegraf 配置是否按预期工作，请使用 --test 标志。此命令以 行协议（InfluxDB 的输入格式）输出指标到 stdout。通常值得先运行此命令，以验证您的 Telegraf 配置和生成的行协议是否正确，然后再将数据写入 InfluxDB。

telegraf --config writeCSV.conf --test

如果您是 Telegraf 的新手，则值得注意 telegraf 配置的 agent 部分下的 debug 选项。默认情况下，此选项设置为 false

debug = false

将此值设置为 true 以输出有关 telegraf 的详细日志。此选项对于调试 telegraf 非常有价值。

最后，运行 telegraf 并传入配置文件的路径，以将 CSV 数据写入您的 InfluxDB Cloud 帐户

telegraf --config $pwd/writeCSV.conf

使用 CLI 将 CSV 写入 InfluxDB Cloud

首先，您必须下载 CLI 才能使用 CLI 将 CSV 写入 InfluxDB Cloud。使用以下命令将 CSV 写入您的 InfluxDB 目标存储桶

influx write -b mybucket\
  -t mytoken
  -o [email protected]
  -f  path/to/writeCSV.csv \
  --header "#datatype dateTime:2006-01-02T15:04:05Z07:00,tag,tag,double,measurement"

提供 header 标志以提供有关 CSV 的注释或元数据。注入注释标头，顺序与 CSV 中的列出现顺序相同。我们按各自的数据类型（double、float、int 等）标识时间戳列和格式、标签列、字段列，最后是度量列。

为了避免每次使用 InfluxDB CLI 时都提供身份验证凭据，我建议您创建一个 CLI 配置。创建和使用 CLI 配置是与 InfluxDB CLI 交互的推荐方式。 TL;DR InfluxDB 技术技巧 – 如何在本地使用 CLI 访问 Flux REPL 并将常规 CSV 写入 InfluxDB Cloud 中详细介绍了这种方法。

最后，dryrun 命令以 行协议将您的数据写入 stdout。通常值得运行该命令以确保您的标头和生成的行协议在写入 InfluxDB 之前是正确的。在下面的示例中，我已经设置了 InfluxDB CLI 配置，因此我不需要传入我的 org 或令牌。 dryrun 命令输出我的行协议，并帮助我验证我的 influx 写入是否会成功

influx write dryrun  -b mybucket -f ~/path/to/writeCSV.csv --header "#datatype dateTime:2006-01-02T15:04:05Z07:00,tag,tag,double,measurement"
cpu,host=A,region=east usage_user=15.43 1525812600000000000
cpu,host=A,region=east usage_user=59.25 1525812620000000000
cpu,host=A,region=east usage_user=52.62 1525812640000000000

使用 csv.from() Flux 函数将 CSV 写入 InfluxDB Cloud

InfluxDB 的数据脚本语言称为 Flux。 Flux 使您能够...

• 从多个来源查询您的数据
• 分析和转换您的数据
• 编写自定义函数和脚本
• 按计划执行这些脚本或创建任务
• 监控数据并发送警报
• ...等等！

今天，我们将使用 csv.from() 函数将 CSV 转换为 带注释的 CSV（InfluxDB 输出格式），在 InfluxDB Cloud 中。然后，我们将应用一些额外的 Flux 函数来格式化输出，以便我们可以在 InfluxDB UI 中可视化它。最后，我们将使用 to() 函数将数据实际写入我们的目标存储桶。

import "csv"
csvData = "
_time,region,host,usage_user,_measurement
2018-05-08T20:50:00Z,east,A,15.43,cpu
2018-05-08T20:50:20Z,east,A,59.25,cpu
2018-05-08T20:50:40Z,east,A,52.62,cpu
"
csv.from(csv: csvData, mode: "raw")
|> rename(columns: {usage_user: "_value"})
|> map(fn: (r) => ({ r with _value: float(v: r._value) }))
|> map(fn: (r) => ({ r with _time: time(v: r._time) }))
|> range(start: v.timeRangeStart, stop: v.timeRangeStop)
|> to(bucket: "mybucket")

首先，您必须导入 Flux CSV 包。您可以通过两种方式将 CSV 数据传递到 csv.from() 函数

• 将您的 CSV 数据存储在变量中。
• 指定 CSV 的路径。 请注意，CSV 必须存在于运行 fluxd 进程的同一文件系统中。 换句话说，此方法在 InfluxDB Cloud 的 InfluxDB UI 中不起作用。

还值得注意的是来自 Flux 实验包的第二个 csv.from() 函数，它允许您从远程 url 写入带注释的 CSV。

我们将原始 CSV 数据存储在名为 csvData 的变量中，并使用第一个 csv.from() 函数将我们的原始 CSV 转换为带注释的 CSV。要了解有关带注释的 CSV（Flux 查询的输出格式）的更多信息，请阅读 TL;DR InfluxDB 技术技巧 – 如何解释带注释的 CSV。

上面的屏幕截图显示了来自 csv.from(csv: csvData, mode: "raw") 行的带注释的 CSV 输出。通过在 InfluxDB UI 中切换原始数据视图来查看数据。 csv.from() 函数将数据转换为字符串类型，如 datatype 注释行所示。

• 然后我们使用 rename() 函数将 usage_user 列重命名为 "_value"。之所以重命名此列，是因为 InfluxDB UI 默认可视化名称为 "_value" 的列。
• csv.from() 函数默认会将我们的数据转换为字符串。
• map() 函数将 time() 函数应用于 "_time" 列中的每个记录。
• time() 函数将字符串值转换为时间。

使用 Python 客户端将 CSV 写入 InfluxDB Cloud

InfluxDB Python 客户端支持 Pandas 和 pandas 数据帧，我认为这是将 CSV 写入 InfluxDB 的最简单方法。您的代码如下所示

import pandas as pd
from influxdb_client import InfluxDBClient
 
client = InfluxDBClient(url="https://us-west-2-1.aws.cloud2.influxdata.com", token="mytoken", org="myorg")
write_client = client.write_api()
 
df = pd.read_csv("~/Desktop/writeCSV.csv") 
df['_time'] = pd.to_datetime(df['_time'], format="%Y-%m-%dT%H:%M:%SZ")
df.set_index(['_time'])
 
write_client.write("mybucket", "[email protected]", record=df, data_frame_measurement_name="_measurement",
                           data_frame_tag_columns=['region', 'host'])

在安装 pandas 和 安装 InfluxDB v2 Python 客户端之后，导入库。然后通过提供您的 InfluxDB 实例的 URL、您的令牌和您的 org 来实例化 InfluxDB 客户端。使用 to_datetime() 函数将您的数据帧的时间戳列设置为 datetime 对象。使用 set_index() 函数将 datetime 列设置为数据帧的索引。使用 write 方法将数据写入 InfluxDB。指定您希望将数据帧写入的目标存储桶，并传递给 record 参数。还要指定您的数据帧标签列和度量列。

关于将 CSV 写入 InfluxDB v2 Cloud 和 OSS 的结论

我希望这篇 InfluxDB 博客文章可以帮助您开始使用 InfluxDB 并将 CSV 写入 InfluxDB。如果您正在使用新的 InfluxDB 并且需要帮助，请在我们的社区站点或 Slack 频道中提出一些问题。如果您正在开发一个很酷的 IoT 应用程序或在 InfluxDB 之上监控您的应用程序，我们很乐意听到您的消息，因此请务必分享您的故事！此外，请在评论部分分享您的想法、疑虑或问题。我们很乐意获得您的反馈并帮助您解决遇到的任何问题！

延伸阅读

虽然这篇文章旨在全面概述如何将 CSV 写入 InfluxDB，但以下资源也可能让您感兴趣

1. TL;DR InfluxDB 技术技巧 – 如何在本地使用 CLI 访问 Flux REPL 并将常规 CSV 写入 InfluxDB Cloud：这篇文章分享了如何将 InfluxDB CLI 与 InfluxDB Cloud 一起使用以将 CSV 写入 InfluxDB。
2. TL;DR InfluxDB 技术技巧 – 如何解释带注释的 CSV：这篇文章描述了如何解释带注释的 CSV，即 InfluxDB 的 Flux 查询结果格式。
3. Python 和 InfluxDB v2.0 入门：这篇文章描述了如何开始使用 Python 客户端库。
4. InfluxDB 和 Pandas 入门：这篇文章分享了如何开始使用 Pandas 将数据帧写入 InfluxDB 并使用 Flux 查询返回数据帧。
5. Flux 初学者面临的 5 大障碍以及学习使用 Flux 的资源：这篇文章描述了 Flux 初学者常遇到的障碍以及如何通过使用 InfluxDB UI、理解带注释的 CSV 等来克服这些障碍。
6. Flux 中级用户面临的 5 大障碍以及优化 Flux 的资源：这篇文章描述了 Flux 中级和高级用户常遇到的障碍，同时提供了有关下推模式、Flux 引擎如何工作等方面的更多详细信息。