如何将CSV数据写入InfluxDB v2和InfluxDB Cloud
作者：Anais Dotis-Georgiou / 产品，用例，开发者
2021年7月29日
导航到
将CSV写入InfluxDB v2和InfluxDB Cloud有多种方法，包括
- Telegraf文件插件
- CLI
- csv.from() Flux函数
- 任何客户端库
我将通过这些方法演示如何将CSV数据写入InfluxDB Cloud免费账户，因为注册免费账户是开始使用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。然而，我经常看到新用户在InfluxData社区网站上询问如何将CSV数据从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配置的输出部分
- Bucket。Bucket是数据存储的地方。您可以将其视为数据库。所有bucket都有保留策略。在创建bucket时分配保留策略。创建bucket最简单的方法是通过InfluxDB Cloud用户界面（UI），如此处所述。但是，您也可以使用CLI或API创建bucket。这个bucket将被称为我们的目标bucket，"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。在将数据写入 InfluxDB 之前，运行此命令以验证您的 Telegraf 配置和结果行协议通常很有价值。
telegraf --config writeCSV.conf --test
如果您是 Telegraf 新手，请注意配置文件中代理部分的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"提供标题标志以提供有关您的 CSV 的注释或元数据。按照您的 CSV 列出现的顺序注入注释标题。我们识别时间戳列及其格式、标签列、字段列（根据其相应的数据类型，如 double、float、int 等），最后是度量列。
为了避免每次使用 InfluxDB CLI 时都提供您的认证凭据，我建议您创建一个 CLI 配置。创建和使用 CLI 配置是与 InfluxDB CLI 交互的推荐方法。这种方法在TL;DR InfluxDB Tech Tips – How to Use the CLI Locally to Access the Flux REPL and Write a Regular CSV to InfluxDB Cloud中有详细描述。
最后，dryrun 命令以行协议将数据写入 stdout。在将数据写入 InfluxDB 之前，运行此命令以验证您的标题和结果行协议通常很有价值。在下面的示例中，我已经设置了一个 InfluxDB CLI 配置，因此我无需传入我的组织或令牌。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()函数在InfluxDB Cloud中将CSV转换为标注CSV，这是InfluxDB的输出格式。然后我们将应用一些额外的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的信息，请阅读TL;DR InfluxDB Tech Tips – How to Interpret an Annotated CSV。
上面的截图显示了从csv.from(csv: csvData, mode: "raw")行生成的标注CSV输出。数据通过在InfluxDB UI中切换到原始数据视图进行查看。csv.from()函数将数据转换为字符串类型，如数据类型标注行所示。
- 然后我们使用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()函数将你的dataframe的时间戳列设置为datetime对象。使用set_index()函数将datetime列设置为dataframe的索引。使用写入方法将数据写入InfluxDB。指定你希望将dataframe写入的目标bucket，并将其传递给record 参数。还要指定你的dataframe标签列和测量列。
将CSV写入InfluxDB v2 Cloud和OSS的结论
希望这篇InfluxDB博客文章能帮助你开始使用InfluxDB并将CSV写入InfluxDB。如果你使用的是新版本的InfluxDB并需要帮助，请在我们的社区网站或Slack频道中寻求帮助。如果你正在开发酷炫的IoT应用程序或在InfluxDB上监控应用程序，我们很愿意了解这些情况，所以请确保分享你的故事！此外，请在评论部分分享你的想法、担忧或问题。我们很乐意得到你的反馈并帮助你解决遇到的问题！
进一步阅读
虽然本文旨在提供如何将CSV写入InfluxDB的全面概述，但以下资源也可能对你有所帮助
- TL;DR InfluxDB技术技巧 – 如何在本地使用CLI访问Flux REPL并将常规CSV写入InfluxDB Cloud：这篇文章分享了如何使用InfluxDB CLI与InfluxDB Cloud将CSV写入InfluxDB。
- TL;DR InfluxDB技术技巧 – 如何解释带注释的CSV：这篇文章描述了如何解释带注释的CSV，这是InfluxDB的Flux查询结果格式。
- Python和InfluxDB v2.0入门：这篇文章描述了如何开始使用Python客户端库。
- InfluxDB和Pandas入门：这篇文章分享了如何使用Pandas将dataframe写入InfluxDB，并通过Flux查询返回dataframe。
- Flux初学者的五大挑战和资源，以及如何使用Flux：这篇文章描述了Flux初学者的常见挑战，并介绍了如何通过使用InfluxDB UI、理解带注释的CSV等方式克服这些挑战。
- 中级Flux用户的五大挑战和优化Flux的资源：这篇文章描述了中级和高级Flux用户的常见挑战，并提供了有关pushdown模式、Flux引擎的工作原理等更详细的说明。
