使用Python脚本将InfluxDB导出的带注释CSV转换为标准CSV

问题背景 #

从InfluxDB UI导出的CSV包含元数据行，格式示例如下：

#group  FALSE   FALSE   TRUE    TRUE    FALSE   FALSE   TRUE    TRUE    TRUE    TRUE    TRUE
#datatype   string  long    dateTime:RFC3339    dateTime:RFC3339    dateTime:RFC3339    double  string  string  string  string  string
#default    mean                                        
    result  table   _start  _stop   _time   _value  _field  _measurement    smart_module    serial  type
        0   2023-03-31T08:12:40.697076925Z  2023-03-31T09:12:40.697076925Z  2023-03-31T08:20:00Z    0   sm_alarm    system_test 8   2.14301E+11 sm_extended
        0   2023-03-31T08:12:40.697076925Z  2023-03-31T09:12:40.697076925Z  2023-03-31T08:40:00Z    0   sm_alarm    system_test 8   2.14301E+11 sm_extended
        0   2023-03-31T08:12:40.697076925Z  2023-03-31T09:12:40.697076925Z  2023-03-31T09:00:00Z    0   sm_alarm    system_test 8   2.14301E+11 sm_extended
        0   2023-03-31T08:12:40.697076925Z  2023-03-31T09:12:40.697076925Z  2023-03-31T09:12:40.697076925Z  0   sm_alarm    system_test 8   2.14301E+11 sm_extended

需要将其转换为按时间聚合的格式：把_field作为列名，对应_value为列值，示例如下：

_time                   sm_alarm
2023-03-31T08:20:00Z    0
2023-03-31T08:40:00Z    0
2023-03-31T09:00:00Z    0
2023-03-31T09:12:40.697076925Z    0

原脚本存在以下问题：

• 仅跳过1行元数据，实际需要跳过前3行（#group、#datatype、#default）
• 错误使用逗号分隔字段，InfluxDB导出的CSV是制表符分隔
• 未实现字段转置逻辑，无法将_field转为列名

修改后的Python脚本 #

import csv
from collections import defaultdict

input_file = 'influx.csv'
output_file = 'output.csv'

try:
    with open(input_file, 'r') as csv_file:
        # 指定制表符为分隔符，匹配InfluxDB导出格式
        csv_reader = csv.reader(csv_file, delimiter='\t')
        
        # 跳过前3行元数据行
        for _ in range(3):
            next(csv_reader)
        
        # 读取表头，清理空字符串（处理开头的制表符）
        header = [col.strip() for col in next(csv_reader) if col.strip()]
        
        # 获取关键列的索引
        time_idx = header.index('_time')
        field_idx = header.index('_field')
        value_idx = header.index('_value')
        
        # 按时间分组存储数据：key为时间，value为{字段名: 对应值}
        time_data = defaultdict(dict)
        # 收集所有唯一的字段名
        all_fields = set()
        
        for row in csv_reader:
            # 清理每行数据，过滤空行
            cleaned_row = [col.strip() for col in row if col.strip()]
            if not cleaned_row:
                continue
            
            current_time = cleaned_row[time_idx]
            current_field = cleaned_row[field_idx]
            current_value = cleaned_row[value_idx]
            
            time_data[current_time][current_field] = current_value
            all_fields.add(current_field)
        
        # 构建输出表头：_time + 排序后的所有字段名
        output_header = ['_time'] + sorted(all_fields)
        
        # 写入输出CSV
        with open(output_file, 'w', newline='') as output_csv:
            csv_writer = csv.writer(output_csv, delimiter='\t')
            csv_writer.writerow(output_header)
            
            # 按时间顺序写入数据
            for time_point in sorted(time_data.keys()):
                row = [time_point]
                for field in sorted(all_fields):
                    # 若该时间点无对应字段值，留空
                    row.append(time_data[time_point].get(field, ''))
                csv_writer.writerow(row)
    
    print(f'转换完成，结果已保存到{output_file}')

except FileNotFoundError:
    print(f'错误：未找到文件{input_file}')
except ValueError as e:
    print(f'错误：CSV中缺少必要列 - {e}')
except Exception as e:
    print(f'错误：{e}')

脚本关键点说明 #

1. 分隔符匹配：明确指定制表符\t作为分隔符，适配InfluxDB导出的CSV格式
2. 元数据处理：循环3次跳过前3行元数据，确保读取到有效表头
3. 数据分组：用defaultdict按时间聚合数据，自动处理同一时间点的多字段数据
4. 动态表头：自动收集所有唯一的_field值，生成适配所有字段的输出表头
5. 空值兼容：若某个时间点无对应字段数据，自动填充空字符串保证格式统一

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