使用 InfluxDB 客户端库写入行协议数据

使用 InfluxDB 客户端库将数据构造为时间序列点，然后将它们作为行协议写入 InfluxDB 集群数据库。

构造行协议
- 示例家模式
设置您的项目
构造点和写入行协议

构造行协议

在了解行协议的基本知识之后，您可以构造行协议数据并将其写入 InfluxDB。

所有 InfluxDB 客户端库都将数据以行协议格式写入 InfluxDB。客户端库的 write 方法允许您以原始行协议或客户端库转换为行协议的 Point 对象的形式提供数据。如果您的程序创建了写入 InfluxDB 的数据，请使用客户端库的 Point 接口，以利用程序中的类型安全。

示例家模式

考虑一个用例，您从家庭中的传感器收集数据。每个传感器收集温度、湿度和一氧化碳读数。

要收集这些数据，请使用以下模式

测量: home
- 标签
  - room: 客厅或厨房
- 字段
  - temp: 温度（°C）（浮点数）
  - hum: 百分比湿度（浮点数）
  - co: 一氧化碳（百万分之一）（整数）
- 时间戳：以秒精度表示的Unix时间戳

以下示例展示了如何构建并写入遵循home模式的点。

设置您的项目

本指南中的示例假设您已按照设置InfluxDB和数据写入设置说明开始使用。

设置好InfluxDB和您的项目后，您应该拥有以下内容

InfluxDB集群凭证
- 数据库
- 数据库令牌
- 集群主机名
用于您项目的目录。
以环境变量或在项目配置文件中存储的凭证——例如，一个.env（“dotenv”）文件。
已安装用于将数据写入InfluxDB的客户端库。

以下示例展示了如何构建遵循示例home模式的Point对象，然后将数据以行协议写入InfluxDB集群数据库。

示例使用InfluxDB v3客户端库。有关使用InfluxDB v2客户端库将数据写入InfluxDB v3的示例，请参阅InfluxDB v2客户端。

Go Node.js Python

以下步骤将使用InfluxDB v3 Go客户端设置Go项目。

安装Go 1.13或更高版本。
为您的Go模块创建一个目录并切换到该目录——例如
```
mkdir iot-starter-go && cd $_
```
初始化Go模块——例如
```
go mod init iot-starter
```
安装influxdb3-go，它提供了InfluxDB的influxdb3 Go客户端库模块。
```
go get github.com/InfluxCommunity/influxdb3-go
```

以下步骤将使用InfluxDB v3 JavaScript客户端设置JavaScript项目。

安装Node.js。
为您的JavaScript项目创建一个目录并切换到该目录——例如
```
mkdir -p iot-starter-js && cd $_
```
初始化项目——例如，使用npm
```
npm init
```
安装InfluxDB v3 JavaScript客户端库@influxdata/influxdb3-client。
```
npm install @influxdata/influxdb3-client
```

以下步骤将使用InfluxDB v3 Python客户端设置Python项目。

安装Python。
在您的项目目录内，为您的Python模块创建一个目录并切换到该模块目录——例如
```
mkdir -p iot-starter-py && cd $_
```
可选，但推荐：使用venv或conda激活虚拟环境以安装和执行代码——例如，使用venv输入以下命令以创建和激活项目虚拟环境
```
python3 -m venv envs/iot-starter && source ./envs/iot-starter/bin/activate
```
安装influxdb3-python，它提供了InfluxDB的influxdb_client_3 Python客户端库模块，并安装了pyarrow包以处理Arrow数据。
```
pip install influxdb3-python
```

构造点和写入行协议

客户端库提供了一个或多个Point构造方法。一些库支持语言原生数据结构，例如Go的struct，用于创建点。

Go Node.js Python

为您的模块创建一个文件——例如：main.go。

在main.go中，输入以下示例代码

package main

import (
 "context"
 "os"
 "fmt"
 "time"
 "github.com/InfluxCommunity/influxdb3-go/influxdb3"
 "github.com/influxdata/line-protocol/v2/lineprotocol"
)

func Write() error {
  url := os.Getenv("INFLUX_HOST")
  token := os.Getenv("INFLUX_TOKEN")
  database := os.Getenv("INFLUX_DATABASE")

  // To instantiate a client, call New() with InfluxDB credentials.
  client, err := influxdb3.New(influxdb3.ClientConfig{
   Host: url,
   Token: token,
   Database: database,
  })

  /** Use a deferred function to ensure the client is closed when the
    * function returns.
   **/
  defer func (client *influxdb3.Client)  {
   err = client.Close()
   if err != nil {
     panic(err)
   }
  }(client)

  /** Use the NewPoint method to construct a point.
    * NewPoint(measurement, tags map, fields map, time)
   **/
  point := influxdb3.NewPoint("home",
     map[string]string{
       "room": "Living Room",
     },
     map[string]any{
       "temp": 24.5,
       "hum":  40.5,
       "co":   15i},
     time.Now(),
   )

  /** Use the NewPointWithMeasurement method to construct a point with
    * method chaining.
   **/
  point2 := influxdb3.NewPointWithMeasurement("home").
   SetTag("room", "Living Room").
   SetField("temp", 23.5).
   SetField("hum", 38.0).
   SetField("co",  16i).
   SetTimestamp(time.Now())

  fmt.Println("Writing points")
  points := []*influxdb3.Point{point, point2}

  /** Write points to InfluxDB.
    * You can specify WriteOptions, such as Gzip threshold,
    * default tags, and timestamp precision. Default precision is lineprotocol.Nanosecond
   **/
  err = client.WritePoints(context.Background(), points,
    influxdb3.WithPrecision(lineprotocol.Second))
  return nil
}

func main() {
  Write()
}

要运行模块并将数据写入您的InfluxDB集群数据库，请在您的终端中输入以下命令
```
go run main.go
```

为您的模块创建一个文件——例如：write-points.js。

在write-points.js中，输入以下示例代码

// write-points.js
import { InfluxDBClient, Point } from '@influxdata/influxdb3-client';

/**
 * Set InfluxDB credentials.
 */
const host = process.env.INFLUX_HOST ?? '';
const database = process.env.INFLUX_DATABASE;
const token = process.env.INFLUX_TOKEN;

/**
 * Write line protocol to InfluxDB using the JavaScript client library.
 */
export async function writePoints() {
  /**
   * Instantiate an InfluxDBClient.
   * Provide the host URL and the database token.
   */
  const client = new InfluxDBClient({ host, token });

  /** Use the fluent interface with chained methods to construct Points. */
  const point = Point.measurement('home')
    .setTag('room', 'Living Room')
    .setFloatField('temp', 22.2)
    .setFloatField('hum', 35.5)
    .setIntegerField('co', 7)
    .setTimestamp(new Date().getTime() / 1000);

  const point2 = Point.measurement('home')
    .setTag('room', 'Kitchen')
    .setFloatField('temp', 21.0)
    .setFloatField('hum', 35.9)
    .setIntegerField('co', 0)
    .setTimestamp(new Date().getTime() / 1000);

  /** Write points to InfluxDB.
   * The write method accepts an array of points, the target database, and
   * an optional configuration object.
   * You can specify WriteOptions, such as Gzip threshold, default tags,
   * and timestamp precision. Default precision is lineprotocol.Nanosecond
   **/

  try {
    await client.write([point, point2], database, '', { precision: 's' });
    console.log('Data has been written successfully!');
  } catch (error) {
    console.error(`Error writing data to InfluxDB: ${error.body}`);
  }

  client.close();
}

writePoints();

要运行模块并将数据写入您的{{< 产品名称 >}}数据库，请在您的终端中输入以下命令
```
node writePoints.js
```

为您的模块创建一个文件——例如：write-points.py。

在write-points.py中，输入以下示例代码以批量模式写入数据

import os
from influxdb_client_3 import (
  InfluxDBClient3, InfluxDBError, Point, WritePrecision,
  WriteOptions, write_client_options)

host = os.getenv('INFLUX_HOST')
token = os.getenv('INFLUX_TOKEN')
database = os.getenv('INFLUX_DATABASE')

# Create an array of points with tags and fields.
points = [Point("home")
            .tag("room", "Kitchen")
            .field("temp", 25.3)
            .field('hum', 20.2)
            .field('co', 9)]

# With batching mode, define callbacks to execute after a successful or
# failed write request.
# Callback methods receive the configuration and data sent in the request.
def success(self, data: str):
    print(f"Successfully wrote batch: data: {data}")

def error(self, data: str, exception: InfluxDBError):
    print(f"Failed writing batch: config: {self}, data: {data} due: {exception}")

def retry(self, data: str, exception: InfluxDBError):
    print(f"Failed retry writing batch: config: {self}, data: {data} retry: {exception}")

# Configure options for batch writing.
write_options = WriteOptions(batch_size=500,
                                    flush_interval=10_000,
                                    jitter_interval=2_000,
                                    retry_interval=5_000,
                                    max_retries=5,
                                    max_retry_delay=30_000,
                                    exponential_base=2)

# Create an options dict that sets callbacks and WriteOptions.
wco = write_client_options(success_callback=success,
                          error_callback=error,
                          retry_callback=retry,
                          write_options=write_options)

# Instantiate a synchronous instance of the client with your
# InfluxDB credentials and write options, such as Gzip threshold, default tags,
# and timestamp precision. Default precision is nanosecond ('ns').
with InfluxDBClient3(host=host,
                        token=token,
                        database=database,
                        write_client_options=wco) as client:

      client.write(points, write_precision='s')

要运行模块并将数据写入您的InfluxDB集群数据库，请在您的终端中输入以下命令
```
python write-points.py
```

示例代码执行以下操作

创建一个配置了InfluxDB URL和API令牌的客户端。
构建home 测量 点对象。
以行协议格式将数据发送到InfluxDB并等待响应。
如果写入成功，将成功消息记录到stdout；否则，记录错误消息和错误详情。
关闭客户端以释放资源。

这个页面有帮助吗？

感谢您的反馈！

支持和反馈

感谢您成为我们社区的一部分！我们欢迎并鼓励您对InfluxDB和本文档的反馈和错误报告。要获取支持，请使用以下资源

有年度或支持合同的客户可以联系InfluxData支持。

编辑此页面提交文档问题提交InfluxDB问题

使用 InfluxDB 客户端库写入行协议数据

构造行协议

示例家模式

设置您的项目

构造点和写入行协议

支持和反馈

Flux的未来

InfluxDB v3增强功能和InfluxDB Clustered现已上市

InfluxDB v3性能和功能

InfluxDB Clustered上市

使用 InfluxDB 客户端库写入行协议数据

构造行协议

示例家模式

设置您的项目

构造点和写入行协议

相关

支持和反馈

您的InfluxDB集群URL是什么？

输入集群URL

感谢您的反馈！

Flux的未来

InfluxDB v3增强功能和InfluxDB Clustered现已上市

InfluxDB v3性能和功能

InfluxDB Clustered上市