Documentation

InfluxDB 云端无服务器

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

使用InfluxDB客户端库构造时间序列点的数据，然后将它们作为行协议写入InfluxDB Cloud Serverless存储桶。

构建行协议

通过对 行协议的基本理解， 您可以构建行协议数据并将其写入InfluxDB。

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

示例家庭架构

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

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

  • 测量: home
    • 标签
      • room: 客厅或厨房
    • 字段
      • temp: 摄氏度中的温度 (浮点数)
      • hum: 湿度百分比 (浮点数)
      • co: 一百万分之一的空气中一氧化碳含量 (整数)
    • 时间戳: Unix 时间戳，单位为秒精度

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

设置您的项目

本指南中的示例假设您遵循了 设置 InfluxDB 和 写入数据集设置 在开始使用中的说明。

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

  • InfluxDB Cloud 无服务器凭据：

  • 您的项目的目录。

  • 凭据存储为环境变量或在项目配置文件中–例如，一个 .env (“dotenv”) 文件。

  • 已安装的客户端库用于将数据写入InfluxDB。

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

这些示例使用了 InfluxDB 3 客户端库。 有关使用 InfluxDB v2 客户端库将数据写入 InfluxDB 3 的示例，请参见 InfluxDB v2 clients。

去 节点.js Python

以下步骤使用InfluxDB 3 Go 客户端设置一个 Go 项目：

  1. 安装 Go 1.13 或更高版本。

  2. 为您的 Go 模块创建一个目录并切换到该目录—例如：

    mkdir iot-starter-go && cd $_

  3. 初始化一个 Go 模块——例如：

    go mod init iot-starter

  4. 安装 influxdb3-go, 提供 InfluxDB influxdb3 Go 客户端库模块。

    go get github.com/InfluxCommunity/influxdb3-go/v2

以下步骤使用InfluxDB 3 JavaScript client设置一个JavaScript项目。

  1. 安装 Node.js。

  2. 为您的JavaScript项目创建一个目录并切换到该目录–例如：

    mkdir -p iot-starter-js && cd $_

  3. 初始化一个项目——例如，使用 npm:

    npm init

  4. 安装 @influxdata/influxdb3-client InfluxDB 3 JavaScript 客户端库。

    npm install @influxdata/influxdb3-client

以下步骤使用InfluxDB 3 Python client设置Python项目：

  1. 安装 Python

  2. 在你的项目目录内部，创建一个用于你的Python模块的目录并切换到该模块目录——举例来说：

    mkdir -p iot-starter-py && cd $_

  3. 可选，但推荐: 使用 venv 或 conda 来激活一个虚拟 环境以便于安装和执行代码——例如，使用 venv 输入以下命令以创建并激活项目的虚拟环境：

    python3 -m venv envs/iot-starter && source ./envs/iot-starter/bin/activate

  4. 安装 influxdb3-python， 它提供了 InfluxDB influxdb_client_3 Python 客户端库模块 并且还安装了 pyarrow 包用于 处理 Arrow 数据。

    pip install influxdb3-python

构建点并编写行协议

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

去 节点.js Python

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

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

    package main

import (
 "context"
 "os"
 "fmt"
 "time"
 "github.com/InfluxCommunity/influxdb3-go/v2/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_BUCKET")

  // 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()
}

  3. 要运行模块并将数据写入您的 InfluxDB Cloud Serverless 存储桶，请在终端中输入以下命令：

    go run main.go

  1. 为你的模块创建一个文件，例如： write-points.js。

  2. 在 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_BUCKET;
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 (bucket),
   * 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();

  3. 要运行该模块并将数据写入您的 {{< product-name >}} 桶，请在终端中输入以下命令：

    node writePoints.js

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

  2. 在 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_BUCKET')

# 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')

  3. 要运行模块并将数据写入您的 InfluxDB Cloud Serverless 存储桶，请在终端中输入以下命令：

    python write-points.py

示例代码完成以下操作：

  1. 实例化一个配置了 InfluxDB URL 和 API 令牌的客户端。
  2. 构造 home measurement Point 对象。
  3. 以行协议格式将数据发送到InfluxDB并等待响应。
  4. 如果写入成功，将成功消息记录到标准输出；否则，记录失败消息和错误详细信息。
  5. 关闭客户端以释放资源。
© 2025 InfluxData, Inc.

选择您的 InfluxDB 云无服务器区域

选择您的 InfluxDB Cloud Serverless 区域，我们将为您定制代码示例。

AWS

有关更多信息，请参见 InfluxDB Cloud Serverless regions。

感谢您的反馈!

让我们知道我们可以做得更好的地方：

谢谢！

No thanks

Flux的未来

Flux 正在进入维护模式。您可以像现在一样继续使用它，而无需对您的代码进行任何更改。

阅读更多

InfluxDB 3 开源版本现已公开Alpha测试

InfluxDB 3 Open Source is now available for alpha testing, licensed under MIT or Apache 2 licensing.

我们将发布两个产品作为测试版的一部分。

InfluxDB 3 核心，是我们新的开源产品。 它是一个用于时间序列和事件数据的实时数据引擎。 InfluxDB 3 企业版是建立在核心基础之上的商业版本，增加了历史查询能力、读取副本、高可用性、可扩展性和细粒度安全性。

有关如何开始的更多信息，请查看：

InfluxDB 云端无服务器

您目前正在查看特定于 InfluxDB Cloud Serverless 的文档，该文档由 v3 存储引擎 提供，功能与由 TSM 存储引擎提供的 InfluxDB Cloud 有所不同。

您是否在使用 InfluxDB Cloud Serverless？ 如何查找？

访问 您组织的主页，并在右侧栏底部找到以下内容：

InfluxDB Cloud Serverless Storage Engine Version 3
Yes – I'm using InfluxDB Cloud Serverless No – I'm using InfluxDB Cloud (TSM)