如何高效将自托管InfluxDB v2数据导入Python Polars？

根据你测试的各方法耗时数据，query_csv() 或 query_raw() 是最优选择——这两个方法直接返回原始格式数据，跳过了pandas DataFrame的中间转换步骤，能大幅提升数据导入效率。

方案1：使用query_csv()直接导入Polars #

query_csv()返回CSV迭代器，Polars可直接读取CSV格式数据，无需额外中间转换，代码示例：

from dotenv import load_dotenv
from influxdb_client import InfluxDBClient
import polars
import os
from io import StringIO

load_dotenv()

client = InfluxDBClient(
    url=os.getenv("url"), token=os.getenv("token"), org=os.getenv("org"))

query_api = client.query_api()

flux_string = 'from(bucket: "test_bucket") |> range(start:-2y) |> drop(columns: ["_start","_stop"])'

# 将CSV迭代器转为字符串流
csv_iterator = query_api.query_csv(flux_string)
csv_content = "\n".join(csv_iterator)

# Polars直接读取CSV字符串生成DataFrame
polars_data = polars.read_csv(StringIO(csv_content))

方案2：使用query_raw()解析JSON格式 #

如果调整Flux查询指定返回JSON格式，query_raw()可快速获取原始JSON数据，Polars读取JSON的效率同样出色：

from dotenv import load_dotenv
from influxdb_client import InfluxDBClient
import polars
import os
from io import StringIO

load_dotenv()

client = InfluxDBClient(
    url=os.getenv("url"), token=os.getenv("token"), org=os.getenv("org"))

query_api = client.query_api()

# 修改Flux查询，指定返回JSON格式
flux_string = 'from(bucket: "test_bucket") |> range(start:-2y) |> drop(columns: ["_start","_stop"]) |> format(dataFormat: "json")'

# 获取原始JSON字符串
raw_response = query_api.query_raw(flux_string)
json_content = raw_response.read().decode("utf-8")

# Polars读取JSON生成DataFrame
polars_data = polars.read_json(StringIO(json_content))

额外优化建议 #

• 优化Flux查询：用keep()明确指定需要的字段（代替drop()），缩小查询时间范围，从源头减少返回的数据量，这是最有效的性能提升手段。
• 流式懒加载处理：如果数据量极大，可结合query_csv()的迭代器特性，用Polars的scan_csv进行懒加载，避免一次性加载全部数据到内存：

  from io import BytesIO
  
  csv_iterator = query_api.query_csv(flux_string)
  csv_bytes = "\n".join(csv_iterator).encode("utf-8")
  # 创建懒加载DataFrame，按需执行计算
  polars_lazy = polars.scan_csv(BytesIO(csv_bytes))
  result = polars_lazy.head(100).collect()
  

• 彻底跳过pandas中间层：你当前代码的主要性能瓶颈就是pandas DataFrame -> Polars的转换，直接读取原始数据格式可完全规避这一损耗。

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