Home » Парсинг CSV файлов в Python — пошаговое руководство

Парсинг CSV файлов в Python — пошаговое руководство

Если работаешь с серверами, то наверняка знаешь: данные приходят в разных форматах, и часто это банальный CSV. Логи, метрики, конфигурационные файлы — все это может быть в CSV-формате. И если умеешь правильно парсить CSV в Python, то автоматизация админских задач становится гораздо проще.

Парсинг CSV — это не просто “прочитать файл”. Это обработка данных, которая может серьезно облегчить жизнь сисадмина. Мониторинг серверов, анализ логов, генерация отчетов — все это можно автоматизировать с помощью Python и правильного подхода к CSV.

Сегодня разберем, как это работает на практике, с реальными примерами и кейсами. Никакой теории ради теории — только то, что пригодится в реальной работе.

Как работает парсинг CSV в Python?

CSV (Comma-Separated Values) — это текстовый формат, где данные разделены запятыми. Звучит просто, но на практике есть подводные камни: экранирование, разные разделители, кодировки.

Python предлагает несколько способов работы с CSV:

csv — встроенный модуль, быстрый и надежный
pandas — мощная библиотека для анализа данных
ручной парсинг — когда нужен полный контроль

Для большинства задач админа достаточно встроенного модуля csv. Он работает быстро, не требует дополнительных зависимостей и отлично справляется с типичными задачами.

Базовый парсинг с модулем csv

Начнем с простого примера. Допустим, у нас есть файл с данными мониторинга сервера:

import csv

# Читаем CSV файл
with open('server_stats.csv', 'r', encoding='utf-8') as file:
    reader = csv.reader(file)
    
    # Пропускаем заголовок
    next(reader)
    
    for row in reader:
        timestamp, cpu_usage, memory_usage, disk_usage = row
        print(f"Время: {timestamp}, CPU: {cpu_usage}%, RAM: {memory_usage}%, Диск: {disk_usage}%")

Если файл содержит заголовки, удобнее использовать DictReader:

import csv

with open('server_stats.csv', 'r', encoding='utf-8') as file:
    reader = csv.DictReader(file)
    
    for row in reader:
        if float(row['cpu_usage']) > 80:
            print(f"Предупреждение: высокая загрузка CPU на {row['timestamp']}")
            print(f"CPU: {row['cpu_usage']}%, RAM: {row['memory_usage']}%")

Обработка нестандартных CSV файлов

Реальные CSV файлы редко бывают идеальными. Часто встречаются:

Нестандартные разделители (точка с запятой, табуляция)
Кавычки в данных
Пустые строки
Разные кодировки

Пример обработки файла с нестандартными параметрами:

import csv

# Файл с разделителем ";" и кавычками
with open('complex_data.csv', 'r', encoding='cp1251') as file:
    reader = csv.reader(file, delimiter=';', quotechar='"')
    
    for row in reader:
        # Пропускаем пустые строки
        if not row or not any(row):
            continue
            
        # Обрабатываем данные
        server_name = row[0].strip()
        if server_name:
            print(f"Сервер: {server_name}")

Практические кейсы для админов

Рассмотрим несколько реальных сценариев использования:

Анализ логов веб-сервера

import csv
from collections import defaultdict
from datetime import datetime

def analyze_access_logs(filename):
    ip_counter = defaultdict(int)
    status_codes = defaultdict(int)
    
    with open(filename, 'r') as file:
        reader = csv.reader(file)
        
        for row in reader:
            if len(row) >= 3:
                ip = row[0]
                status = row[2]
                
                ip_counter[ip] += 1
                status_codes[status] += 1
    
    # Топ IP-адресов
    print("Топ 10 IP-адресов:")
    for ip, count in sorted(ip_counter.items(), key=lambda x: x[1], reverse=True)[:10]:
        print(f"{ip}: {count} запросов")
    
    # Статистика по кодам ответов
    print("\nКоды ответов:")
    for code, count in sorted(status_codes.items()):
        print(f"{code}: {count}")

# Использование
analyze_access_logs('access_log.csv')

Мониторинг использования ресурсов

import csv
import smtplib
from email.mime.text import MIMEText

def check_server_resources(filename, cpu_threshold=80, memory_threshold=85):
    alerts = []
    
    with open(filename, 'r') as file:
        reader = csv.DictReader(file)
        
        for row in reader:
            server = row['server_name']
            cpu = float(row['cpu_usage'])
            memory = float(row['memory_usage'])
            
            if cpu > cpu_threshold:
                alerts.append(f"Сервер {server}: CPU {cpu}%")
            
            if memory > memory_threshold:
                alerts.append(f"Сервер {server}: Memory {memory}%")
    
    if alerts:
        send_alert_email(alerts)
    
def send_alert_email(alerts):
    message = "\n".join(alerts)
    print(f"Отправка уведомления:\n{message}")
    # Здесь код отправки email

check_server_resources('monitoring.csv')

Работа с pandas для сложного анализа

Когда нужен более сложный анализ данных, pandas становится незаменимым:

import pandas as pd

# Читаем CSV
df = pd.read_csv('server_metrics.csv')

# Базовая статистика
print("Средние значения:")
print(df[['cpu_usage', 'memory_usage', 'disk_usage']].mean())

# Группировка по серверам
server_stats = df.groupby('server_name').agg({
    'cpu_usage': ['mean', 'max'],
    'memory_usage': ['mean', 'max'],
    'disk_usage': ['mean', 'max']
}).round(2)

print("\nСтатистика по серверам:")
print(server_stats)

# Поиск аномалий
cpu_anomalies = df[df['cpu_usage'] > df['cpu_usage'].mean() + 2 * df['cpu_usage'].std()]
print(f"\nАномалии CPU: {len(cpu_anomalies)} случаев")

Обработка ошибок и исключений

В реальной работе файлы могут быть повреждены или содержать некорректные данные:

import csv
import logging

def safe_csv_parse(filename):
    try:
        with open(filename, 'r', encoding='utf-8') as file:
            reader = csv.DictReader(file)
            
            for line_num, row in enumerate(reader, 1):
                try:
                    # Валидация данных
                    cpu = float(row.get('cpu_usage', 0))
                    memory = float(row.get('memory_usage', 0))
                    
                    if cpu < 0 or cpu > 100:
                        logging.warning(f"Строка {line_num}: некорректное значение CPU: {cpu}")
                        continue
                    
                    # Обработка данных
                    process_server_data(row)
                    
                except ValueError as e:
                    logging.error(f"Строка {line_num}: ошибка парсинга - {e}")
                    continue
                except Exception as e:
                    logging.error(f"Строка {line_num}: неожиданная ошибка - {e}")
                    continue
                    
    except FileNotFoundError:
        logging.error(f"Файл {filename} не найден")
    except PermissionError:
        logging.error(f"Нет прав доступа к файлу {filename}")
    except UnicodeDecodeError:
        logging.error(f"Ошибка кодировки файла {filename}")

def process_server_data(row):
    # Здесь логика обработки данных
    pass

Сравнение различных подходов

Подход	Скорость	Память	Функциональность	Лучше для
csv.reader	Высокая	Низкое потребление	Базовая	Простой парсинг, большие файлы
csv.DictReader	Средняя	Среднее потребление	Удобная	Файлы с заголовками
pandas.read_csv	Средняя	Высокое потребление	Очень высокая	Анализ данных, агрегации
Ручной парсинг	Зависит от реализации	Контролируемое	Максимальная гибкость	Нестандартные форматы

Генерация CSV файлов

Часто нужно не только читать, но и создавать CSV файлы:

import csv
import psutil
from datetime import datetime

def generate_system_report():
    filename = f"system_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
    
    with open(filename, 'w', newline='', encoding='utf-8') as file:
        writer = csv.writer(file)
        
        # Заголовок
        writer.writerow(['Timestamp', 'CPU %', 'Memory %', 'Disk %', 'Network Sent', 'Network Recv'])
        
        # Данные
        cpu_percent = psutil.cpu_percent(interval=1)
        memory = psutil.virtual_memory()
        disk = psutil.disk_usage('/')
        network = psutil.net_io_counters()
        
        writer.writerow([
            datetime.now().isoformat(),
            cpu_percent,
            memory.percent,
            disk.percent,
            network.bytes_sent,
            network.bytes_recv
        ])
    
    print(f"Отчет сохранен в {filename}")

# Создание отчета с использованием DictWriter
def generate_detailed_report(servers_data):
    with open('servers_report.csv', 'w', newline='', encoding='utf-8') as file:
        fieldnames = ['server_name', 'ip', 'cpu_usage', 'memory_usage', 'status', 'last_check']
        writer = csv.DictWriter(file, fieldnames=fieldnames)
        
        writer.writeheader()
        for server in servers_data:
            writer.writerow(server)

Интеграция с другими инструментами

CSV парсинг отлично работает в связке с другими инструментами:

Отправка в InfluxDB

import csv
from influxdb import InfluxDBClient

def csv_to_influxdb(filename):
    client = InfluxDBClient(host='localhost', port=8086, database='monitoring')
    
    with open(filename, 'r') as file:
        reader = csv.DictReader(file)
        
        points = []
        for row in reader:
            point = {
                "measurement": "server_stats",
                "tags": {"server": row['server_name']},
                "time": row['timestamp'],
                "fields": {
                    "cpu_usage": float(row['cpu_usage']),
                    "memory_usage": float(row['memory_usage'])
                }
            }
            points.append(point)
        
        client.write_points(points)

Создание графиков с matplotlib

import csv
import matplotlib.pyplot as plt
from datetime import datetime

def create_cpu_graph(filename):
    timestamps = []
    cpu_values = []
    
    with open(filename, 'r') as file:
        reader = csv.DictReader(file)
        
        for row in reader:
            timestamps.append(datetime.fromisoformat(row['timestamp']))
            cpu_values.append(float(row['cpu_usage']))
    
    plt.figure(figsize=(12, 6))
    plt.plot(timestamps, cpu_values, linewidth=2)
    plt.title('CPU Usage Over Time')
    plt.xlabel('Time')
    plt.ylabel('CPU %')
    plt.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.savefig('cpu_usage.png', dpi=300)
    plt.show()

Автоматизация и скрипты

Пример полноценного скрипта для автоматизации мониторинга:

#!/usr/bin/env python3
import csv
import os
import sys
import argparse
from datetime import datetime, timedelta

class ServerMonitor:
    def __init__(self, csv_file, config):
        self.csv_file = csv_file
        self.cpu_threshold = config.get('cpu_threshold', 80)
        self.memory_threshold = config.get('memory_threshold', 85)
        self.alert_cooldown = config.get('alert_cooldown', 300)  # 5 минут
        
    def check_alerts(self):
        alerts = []
        
        with open(self.csv_file, 'r') as file:
            reader = csv.DictReader(file)
            
            for row in reader:
                # Проверяем только последние записи
                timestamp = datetime.fromisoformat(row['timestamp'])
                if datetime.now() - timestamp > timedelta(minutes=10):
                    continue
                
                server = row['server_name']
                cpu = float(row['cpu_usage'])
                memory = float(row['memory_usage'])
                
                if cpu > self.cpu_threshold:
                    alerts.append({
                        'server': server,
                        'type': 'cpu',
                        'value': cpu,
                        'timestamp': timestamp
                    })
                
                if memory > self.memory_threshold:
                    alerts.append({
                        'server': server,
                        'type': 'memory',
                        'value': memory,
                        'timestamp': timestamp
                    })
        
        return alerts
    
    def generate_report(self, output_file):
        # Генерация отчета
        pass

def main():
    parser = argparse.ArgumentParser(description='Server Monitor')
    parser.add_argument('--csv', required=True, help='CSV file with metrics')
    parser.add_argument('--config', help='Configuration file')
    parser.add_argument('--output', help='Output report file')
    
    args = parser.parse_args()
    
    config = {'cpu_threshold': 80, 'memory_threshold': 85}
    
    monitor = ServerMonitor(args.csv, config)
    alerts = monitor.check_alerts()
    
    if alerts:
        print(f"Найдено {len(alerts)} предупреждений:")
        for alert in alerts:
            print(f"  {alert['server']}: {alert['type']} = {alert['value']}%")
    else:
        print("Все серверы в норме")

if __name__ == "__main__":
    main()

Интересные факты и нестандартные применения

Вот несколько интересных способов использования CSV парсинга:

Конфигурационные файлы — CSV можно использовать для хранения конфигураций серверов, что удобнее XML или JSON для табличных данных
Миграция данных — CSV идеален для переноса данных между разными системами мониторинга
A/B тестирование — анализ логов разных версий приложений
Биллинг — подсчет использования ресурсов по логам

Нестандартный пример — парсинг вывода команд системы:

import csv
import subprocess
import io

def parse_ps_output():
    # Получаем вывод ps в CSV формате
    result = subprocess.run(['ps', 'aux'], capture_output=True, text=True)
    
    # Превращаем в CSV
    lines = result.stdout.strip().split('\n')
    
    # Первая строка - заголовки
    headers = lines[0].split()
    
    processes = []
    for line in lines[1:]:
        parts = line.split(None, len(headers)-1)  # Разделяем на нужное количество частей
        if len(parts) == len(headers):
            processes.append(dict(zip(headers, parts)))
    
    # Находим процессы с высоким CPU
    cpu_hogs = [p for p in processes if float(p['%CPU']) > 50]
    
    for proc in cpu_hogs:
        print(f"Процесс {proc['PID']} ({proc['COMMAND']}) использует {proc['%CPU']}% CPU")

Оптимизация производительности

Для работы с большими CSV файлами:

import csv
from itertools import islice

def process_large_csv(filename, chunk_size=1000):
    """Обработка больших CSV файлов по частям"""
    with open(filename, 'r') as file:
        reader = csv.DictReader(file)
        
        while True:
            chunk = list(islice(reader, chunk_size))
            if not chunk:
                break
            
            # Обрабатываем chunk
            process_chunk(chunk)
            
def process_chunk(chunk):
    """Обработка части данных"""
    for row in chunk:
        # Логика обработки
        pass

# Для еще больших файлов - используем pandas с chunksize
import pandas as pd

def process_huge_csv(filename):
    chunk_size = 10000
    
    for chunk in pd.read_csv(filename, chunksize=chunk_size):
        # Обработка chunk
        high_cpu = chunk[chunk['cpu_usage'] > 80]
        if not high_cpu.empty:
            print(f"Найдено {len(high_cpu)} записей с высоким CPU")

Деплой и использование на серверах

Для работы на продакшн серверах нужно учесть несколько моментов:

# requirements.txt
pandas>=1.3.0
psutil>=5.8.0
matplotlib>=3.5.0

# Dockerfile для контейнеризации
FROM python:3.9-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY monitor.py .
CMD ["python", "monitor.py", "--csv", "/data/metrics.csv"]

Если нужен надежный хостинг для ваших скриптов мониторинга, рекомендую посмотреть на VPS или выделенные серверы — там можно развернуть полноценную систему мониторинга.

Заключение и рекомендации

Парсинг CSV в Python — это мощный инструмент для автоматизации админских задач. Основные рекомендации:

Используйте csv.DictReader для файлов с заголовками — это делает код более читаемым
Всегда обрабатывайте ошибки — реальные файлы могут содержать некорректные данные
Для анализа данных выбирайте pandas — он предоставляет гораздо больше возможностей
Оптимизируйте для больших файлов — используйте chunking для файлов размером больше оперативной памяти
Автоматизируйте процесс — создавайте скрипты, которые можно запускать по расписанию

CSV парсинг открывает огромные возможности для автоматизации: от простого анализа логов до создания полноценных систем мониторинга. Главное — правильно выбрать инструмент под задачу и не забывать про обработку ошибок.

Начните с простых скриптов и постепенно усложняйте их. Очень скоро вы обнаружите, что многие рутинные задачи можно автоматизировать с помощью Python и CSV.

В этой статье собрана информация и материалы из различных интернет-источников. Мы признаем и ценим работу всех оригинальных авторов, издателей и веб-сайтов. Несмотря на то, что были приложены все усилия для надлежащего указания исходного материала, любая непреднамеренная оплошность или упущение не являются нарушением авторских прав. Все упомянутые товарные знаки, логотипы и изображения являются собственностью соответствующих владельцев. Если вы считаете, что какой-либо контент, использованный в этой статье, нарушает ваши авторские права, немедленно свяжитесь с нами для рассмотрения и принятия оперативных мер.

Данная статья предназначена исключительно для ознакомительных и образовательных целей и не ущемляет права правообладателей. Если какой-либо материал, защищенный авторским правом, был использован без должного упоминания или с нарушением законов об авторском праве, это непреднамеренно, и мы исправим это незамедлительно после уведомления. Обратите внимание, что переиздание, распространение или воспроизведение части или всего содержимого в любой форме запрещено без письменного разрешения автора и владельца веб-сайта. Для получения разрешений или дополнительных запросов, пожалуйста, свяжитесь с нами.