InfluxDB技术咨询：新增测量值、多测量值关联及Python操作方案疑问

Hey there! Let's break down your questions step by step—since you prefer Python, I'll prioritize that but also cover the terminal approach you're currently using.

1. 新增Measurement的方法 #

Quick note: InfluxDB doesn't require you to explicitly create a measurement upfront—it gets automatically created when you first write data to it. That's one of the things that trips up new users!

终端（Influx CLI）方式 #

1. 先进入InfluxDB的终端shell：

   influx
   

2. 切换到你的bucket和org（适用于InfluxDB 2.x）：

   use bucket/your-bucket-name
   

3. 用INSERT语句写入数据，同时创建measurement：

   INSERT measurement_name,tag_key=tag_value field_key=field_value
   

   举个实际例子，创建weathermeasurement并写入北京的天气数据：

   INSERT weather,city=Beijing temperature=25.5,humidity=60
   

   执行完这条命令，weather就已经存在了，你可以用SHOW MEASUREMENTS验证。

Python方式 #

推荐使用官方维护的influxdb-client库（比旧的influxdb库更稳定、功能更全），步骤如下：

1. 安装库：

   pip install influxdb-client
   

2. 编写代码写入数据，自动创建measurement：

   from influxdb_client import InfluxDBClient, Point
   from influxdb_client.client.write_api import SYNCHRONOUS
   
   # 替换成你的实际配置
   INFLUX_TOKEN = "your-influxdb-token"
   INFLUX_ORG = "your-org-name"
   INFLUX_BUCKET = "your-bucket-name"
   INFLUX_URL = "http://localhost:8086"  # 默认地址
   
   # 连接客户端并写入数据
   with InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG) as client:
       write_api = client.write_api(write_options=SYNCHRONOUS)
       # 构造Point对象，指定measurement、标签和字段
       weather_point = Point("weather") \
           .tag("city", "Shanghai") \
           .field("temperature", 27.0) \
           .field("humidity", 55)
       # 写入数据，自动创建weather measurement
       write_api.write(bucket=INFLUX_BUCKET, org=INFLUX_ORG, record=weather_point)
   

2. 类似SQL JOIN整合不同Measurement的数据 #

InfluxDB的InfluxQL对JOIN支持非常有限，所以推荐使用Flux查询语言（InfluxDB 2.x的默认查询语言）来实现类似SQL JOIN的效果。Flux的join()函数可以让你基于共同的键（比如时间戳、标签）关联不同measurement的数据。

终端执行Flux查询 #

1. 进入Influx shell，直接执行Flux查询：

   import "join"
   
   # 第一个数据集：从weather measurement取温度数据
   weather_data = from(bucket: "your-bucket-name")
     |> range(start: -1h)  # 取最近1小时的数据
     |> filter(fn: (r) => r._measurement == "weather" and r._field == "temperature")
     |> keep(columns: ["_time", "city", "_value"])  # 保留需要的字段
     |> rename(columns: {_value: "temperature"})  # 重命名字段方便识别
   
   # 第二个数据集：从air_quality measurement取PM2.5数据
   air_quality_data = from(bucket: "your-bucket-name")
     |> range(start: -1h)
     |> filter(fn: (r) => r._measurement == "air_quality" and r._field == "pm25")
     |> keep(columns: ["_time", "city", "_value"])
     |> rename(columns: {_value: "pm25"})
   
   # 基于时间戳和城市标签关联两个数据集
   join(
     tables: {weather: weather_data, air: air_quality_data},
     on: ["_time", "city"]
   )
   

Python中执行Flux查询 #

用influxdb-client的查询API来执行Flux并处理结果：

from influxdb_client import InfluxDBClient

# 替换成你的实际配置
INFLUX_TOKEN = "your-influxdb-token"
INFLUX_ORG = "your-org-name"
INFLUX_BUCKET = "your-bucket-name"
INFLUX_URL = "http://localhost:8086"

with InfluxDBClient(url=INFLUX_URL, token=INFLUX_TOKEN, org=INFLUX_ORG) as client:
    query_api = client.query_api()
    # 编写Flux查询语句
    flux_query = f"""
    import "join"

    weather_data = from(bucket: "{INFLUX_BUCKET}")
      |> range(start: -1h)
      |> filter(fn: (r) => r._measurement == "weather" and r._field == "temperature")
      |> keep(columns: ["_time", "city", "_value"])
      |> rename(columns: {{_value: "temperature"}})

    air_quality_data = from(bucket: "{INFLUX_BUCKET}")
      |> range(start: -1h)
      |> filter(fn: (r) => r._measurement == "air_quality" and r._field == "pm25")
      |> keep(columns: ["_time", "city", "_value"])
      |> rename(columns: {{_value: "pm25"}})

    join(
      tables: {{weather: weather_data, air: air_quality_data}},
      on: ["_time", "city"]
    )
    """
    # 执行查询
    result = query_api.query(org=INFLUX_ORG, query=flux_query)
    # 遍历结果并打印
    for table in result:
        for record in table.records:
            print(f"时间: {record.get_time()}, 城市: {record['city']}, 温度: {record['temperature']}, PM2.5: {record['pm25']}")

补充说明 #

• Flux的join()支持多种关联逻辑（内连接、左连接等），可以通过how参数指定，比如how: "inner"（默认）、how: "left"。
• 如果必须用InfluxQL，只能做有限的JOIN（比如同一个measurement内的标签关联），但Flux是更灵活的选择，官方也推荐用Flux做复杂查询。

内容的提问来源于stack exchange，提问作者Abhay Nayak