多Python进程写入InfluxDB数据丢失问题排查求助

在RPI5服务器上通过cron运行多个Python进程，这些进程从Web API读取数据后写入同一InfluxDB Bucket，但出现部分数据丢失的情况。写入InfluxDB的代码如下：

influxdb_client = InfluxDBClient(url=url, token=token, org=org) 
...

def f(df): 
  write_api = influxdb_client.write_api()
  ...
  record = [] 
  for i in range(df.shape[0]):
    point = Point(measurement).tag("location", ...).time(...)  
    for col in list(df.columns): 
      value = df.loc[i, col]  
      point = point.field(col, value) 
      record += [point]   
  write_api.write(bucket=bucket, org=org, record=record) 

...
# df是包含20-500行、10-20列的DataFrame
f(df) 

数据丢失的原因可能是什么？是否和异步/同步有关？

1. 异步写入的未提交数据丢失 #

当前代码中write_api = influxdb_client.write_api()使用默认的异步批量写入模式。cron任务执行完成后进程会直接退出，此时异步write_api的后台批量线程可能还没来得及将缓存的数据发送到InfluxDB，导致这部分数据丢失。另外，多进程并发异步写入时，容易触发InfluxDB的限流机制，且默认配置下无重试逻辑，被拒绝的请求数据会直接丢失。

验证/修复方案：改用同步写入模式，确保数据提交成功后再结束进程：

from influxdb_client.client.write_api import WriteOptions

write_api = influxdb_client.write_api(write_options=WriteOptions(sync=True))

2. 数据覆盖问题 #

如果多个进程写入的Point存在相同的measurement、tag集合和时间戳，InfluxDB会自动覆盖旧数据。检查Point.time(...)的生成逻辑：

• 时间戳精度是否足够？若多进程在同一毫秒内写入同tag的Point，后写入的会覆盖先写入的。
• 是否存在不同进程生成完全相同的Point（measurement、tag、time均一致），导致数据被覆盖。

修复方案：优化时间戳生成逻辑，确保同一tag下的时间戳唯一（比如提高精度到微秒），或为不同进程添加唯一标识tag（如process_id）。

3. 未处理写入异常 #

当前代码未对write_api.write()做异常捕获。若写入时出现网络波动、InfluxDB服务不可用、权限不足等情况，会直接抛出异常但无重试或日志记录，导致对应数据丢失。

修复方案：添加异常捕获和重试逻辑：

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def safe_write(record):
    try:
        write_api.write(bucket=bucket, org=org, record=record)
    except Exception as e:
        print(f"写入失败: {str(e)}")
        raise  # 抛出异常触发重试

# 调用safe_write替代直接write
safe_write(record)

4. Cron任务并发冲突 #

若cron任务的执行间隔过短，可能出现上一次进程尚未完成，下一次任务已启动的情况，导致资源竞争或InfluxDB连接池耗尽，引发写入失败。

修复方案：为cron任务添加文件锁，确保同一时间仅一个实例运行：

import fcntl
import sys

def get_lock():
    lock_file = open('/var/lock/influx_writer.lock', 'w')
    try:
        fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
        return lock_file
    except IOError:
        print("已有任务在运行，退出")
        sys.exit(1)

# 任务启动前获取锁
lock = get_lock()

5. InfluxDB自身配置限制 #

检查InfluxDB的相关配置：

• 写入速率限制：若超过influxd.conf中rate-limit相关配置，请求会被拒绝。
• Bucket保留策略：若保留策略设置过短，数据会被自动清理。
• 资源状态：服务器内存或磁盘空间不足，会导致InfluxDB无法持久化数据。

修复方案：调整InfluxDB配置，确保写入速率在限制范围内，检查Bucket保留策略，保障服务器资源充足。

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