Базы данных

Файлов для хранения данных достаточно до тех пор, пока накопленное не потребовалось нескольким людям одновременно.

Ограничения хранения в файлах

Хранение данных в лаборатории почти всегда развивается по одному сценарию. Сначала появляется result.csv. Затем result_final.csv, result_final_v2.csv и папка 2026-08-20/ с двадцатью файлами, названными по энергии пучка, номеру захода и фамилии оператора. Через год накапливаются десятки гигабайт CSV-файлов, и вопрос «показать все измерения канала BPM01 при энергии выше 10 МэВ» превращается в вечер работы со скриптом, рекурсивно обходящим папки и разбирающим имена файлов.

Файлы и папки оказываются непригодными в трёх отношениях:

  • Конкурентный доступ. Скрипт сбора данных записывает файл, а исследователь в этот момент открывает его для анализа и читает недописанную строку. Два процесса записывают одновременно, и файл оказывается повреждённым.
  • Целостность. CSV не проверяет ничего. В столбец с температурой попадает строка "error", порядок столбцов от файла к файлу различается, а измерение ссылается на заход, удалённый кем-то вместе с папкой.
  • Поиск. Чтобы найти нужные записи, приходится читать все файлы целиком в память. На миллионах записей это занимает минуты; при правильно организованных данных тот же запрос выполняется за миллисекунды.

Все три задачи решает СУБД — система управления базами данных. Она берёт на себя хранение, разграничивает одновременный доступ, следит за целостностью и предоставляет язык, приспособленный для запросов, на котором описывается, что нужно найти, а не как перебирать файлы.

Реляционная модель

Реляционная модель, предложенная Эдгаром Коддом в 1970 году, до сих пор считается основной. Данные укладываются в таблицы, называемые в теории отношениями (relations). Строка таблицы содержит запись об одном объекте, а столбец задаёт атрибут, наделённый фиксированным типом.

Спроектируем схему для типичного эксперимента. Имеются измерительные кампании, внутри каждой находятся заходы (runs), различающиеся параметрами установки, а внутри каждого захода — точки измерений, снятые с датчиков.

experiments ──< runs ──< measurements
   (1)          (N)          (N)
  • experiments описывает кампанию, её название, установку и дату начала;
  • runs описывает заход, его кампанию, номер, энергию пучка и оператора;
  • measurements описывает отдельную точку, её заход, канал, время и значение.

Знак ──< читается как связь «один ко многим»: у одной кампании много заходов, а у одного захода много снятых точек.

Ключи и связи

Первичным ключом (primary key) называется столбец или набор столбцов, однозначно идентифицирующий строку. Обычно это суррогатный целочисленный id, выдаваемый базой самостоятельно.

Внешним ключом (foreign key) называется столбец, ссылающийся на первичный ключ другой таблицы. runs.experiment_id хранит id кампании, и таким образом строки связываются между собой. СУБД следит за валидностью ссылки: нельзя вставить заход с experiment_id = 42, если кампании с таким id нет, и нельзя удалить кампанию, на которую ссылаются заходы.

Связь «многие ко многим» (например, публикации и авторы) выражается через промежуточную таблицу, снабжённую двумя внешними ключами, и отдельного механизма для неё не требуется.

Нормализация

Соблазнительно поместить всё в одну широкую таблицу, повторяя в каждой строке измерения название кампании, энергию и фамилию оператора. Такая схема работает до первой правки. Если оператор оказался «Ивановым», а не «Ивановым А.», обновлять приходится миллион строк, а при сбое скрипта на середине данные остаются противоречивыми. Это явление называется аномалиями обновления.

Нормализацией называется устранение дублирования, при котором каждый факт хранится один раз, а всё остальное на него ссылается. Формально она описывается нормальными формами, а для практики достаточно первых трёх, сводящихся к мнемонике: неключевые атрибуты зависят «от ключа, от всего ключа и ни от чего, кроме ключа». Энергия является свойством захода, а не точки измерения, следовательно, ей место в runs. Название установки принадлежит кампании, поэтому оно хранится в experiments.

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

Основы SQL на примере SQLite

SQL (Structured Query Language) — декларативный язык для реляционных баз. Пользователь описывает нужный результат, а СУБД самостоятельно решает, как его получить.

Изучение ведётся на SQLite, встраиваемой СУБД без сервера: вся база лежит в одном файле, а драйвер входит в стандартную библиотеку Python (модуль sqlite3). При этом SQLite остаётся полноценной реляционной СУБД с транзакциями, работающей с базами в десятки гигабайт. Для лабораторных данных этого достаточно.

Открыть консоль SQLite можно командой sqlite3 lab.db, которая создаёт файл самостоятельно, а просмотреть базу в графическом интерфейсе позволяет программа DB Browser for SQLite.

CREATE TABLE: создание таблиц

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

CREATE TABLE experiments (
    id         INTEGER PRIMARY KEY,   -- первичный ключ, база выдаёт сама
    name       TEXT NOT NULL UNIQUE,  -- название кампании, без повторов
    setup      TEXT,                  -- описание установки
    started_at TEXT NOT NULL          -- дата в ISO-формате: '2026-08-20'
);

CREATE TABLE runs (
    id            INTEGER PRIMARY KEY,
    experiment_id INTEGER NOT NULL REFERENCES experiments(id),  -- внешний ключ
    run_number    INTEGER NOT NULL,
    energy_mev    REAL NOT NULL,      -- энергия пучка в заходе, МэВ
    operator      TEXT,
    started_at    TEXT NOT NULL,
    UNIQUE (experiment_id, run_number)  -- номер захода уникален внутри кампании
);

CREATE TABLE measurements (
    id      INTEGER PRIMARY KEY,
    run_id  INTEGER NOT NULL REFERENCES runs(id),
    channel TEXT NOT NULL,            -- имя канала: 'BPM01', 'T_magnet', ...
    t       REAL NOT NULL,            -- время от начала захода, с
    value   REAL NOT NULL             -- измеренное значение
);

Ограничения NOT NULL, UNIQUE и внешние ключи обеспечивают целостность, отсутствовавшую у CSV: попытка записать некорректные данные завершится ошибкой, а не порчей, обнаруживаемой через полгода.

Внешние ключи SQLite проверяет только при явно включённой проверке, и включать её необходимо в каждом новом соединении:

PRAGMA foreign_keys = ON;

Без этой строки вставка ссылки на несуществующую кампанию пройдёт без ошибок, и база незаметно накопит нарушенные связи. PostgreSQL и MySQL проверяют внешние ключи по умолчанию, однако MySQL без предупреждения игнорирует объявленные ключи в таблицах MyISAM, а сессионная переменная foreign_key_checks=0, выставляемая всеми дампами mysqldump, снимает проверку и в InnoDB.

INSERT: вставляем данные

Заполняемые базой столбцы перечислять не требуется: id она выдаст очередной. experiment_id для захода необходимо указать, иначе неизвестно, к какой кампании он относится.

INSERT INTO experiments (name, setup, started_at)
VALUES ('Калибровка датчиков положения', 'стенд ЛИУ', '2026-08-20');

INSERT INTO runs (experiment_id, run_number, energy_mev, operator, started_at)
VALUES (1, 1, 12.5, 'Иванов', '2026-08-20T10:15:00');

-- несколько строк одним запросом
INSERT INTO measurements (run_id, channel, t, value) VALUES
    (1, 'BPM01',    0.0, 0.412),
    (1, 'BPM01',    0.1, 0.418),
    (1, 'T_magnet', 0.0, 42.3);

Последняя форма, со списком кортежей после VALUES, вставляет группу строк одним запросом; для тысячи точек с АЦП это на порядки быстрее тысячи отдельных INSERT; причина рассматривается в разделе про транзакции.

SELECT: выборка

Вопрос, требовавший вечера работы со скриптом-обходчиком папок, записывается в три строки:

-- все точки канала BPM01 из первого захода, по возрастанию времени
SELECT t, value
FROM measurements
WHERE run_id = 1 AND channel = 'BPM01'
ORDER BY t;

-- десять последних заходов с энергией выше 10 МэВ
SELECT run_number, energy_mev, started_at
FROM runs
WHERE energy_mev > 10
ORDER BY started_at DESC
LIMIT 10;

WHERE фильтрует строки, ORDER BY сортирует отобранное (DESC задаёт убывание), LIMIT ограничивает результат. Звёздочка SELECT * возвращает все столбцы, что удобно в консоли, однако в скриптах нужные столбцы перечисляются явно.

JOIN: соединение таблиц

Нормализация разложила данные по трём таблицам, а JOIN собирает разложенное обратно. Соединение сопоставляет строки по условию, обычно «внешний ключ равен первичному»:

-- каждая точка вместе с номером захода и названием кампании
SELECT e.name, r.run_number, m.t, m.value
FROM measurements AS m
JOIN runs        AS r ON r.id = m.run_id
JOIN experiments AS e ON e.id = r.experiment_id
WHERE m.channel = 'BPM01'
ORDER BY e.name, r.run_number, m.t;

AS m задаёт псевдоним, избавляющий от длинных имён. Кроме обычного (внутреннего) JOIN существует LEFT JOIN, сохраняющий строки левой таблицы даже при отсутствии пары справа. Таким образом находят заходы без измерений.

GROUP BY: агрегация

Агрегатные функции (COUNT, AVG, SUM, MIN, MAX) сворачивают отобранную группу строк в одно число, а GROUP BY задаёт признак группировки:

-- сводка по каждому заходу: число точек и статистика сигнала
SELECT r.run_number,
       r.energy_mev,
       COUNT(*)     AS n_points,
       AVG(m.value) AS mean_value,
       MIN(m.value) AS min_value,
       MAX(m.value) AS max_value
FROM measurements AS m
JOIN runs AS r ON r.id = m.run_id
WHERE m.channel = 'BPM01'
GROUP BY r.id
HAVING COUNT(*) >= 100   -- отбрасываем слишком короткие заходы
ORDER BY r.energy_mev;

HAVING фильтрует после группировки, по вычисленным агрегатам; WHERE отбирает строки до неё.

Индексы: ускорение поиска

Без индекса запрос с WHERE заставляет базу просматривать всю таблицу, строку за строкой. Индекс — это дополнительная структура (обычно B-дерево), позволяющая находить строки по значению столбца за логарифмическое время, наподобие предметного указателя в книге.

CREATE INDEX idx_measurements_run_channel
ON measurements (run_id, channel);

После этого выборка точек конкретного канала конкретного захода не зависит от общего размера таблицы. Первичный ключ и UNIQUE индексируются автоматически, а внешние ключи и столбцы, часто участвующие в фильтрации, приходится индексировать отдельно. Платой за индекс являются место на диске и несколько более медленная вставка, поэтому создавать индексы на все столбцы подряд не следует. Проверить, использует ли запрос построенный индекс, можно, приписав перед ним EXPLAIN QUERY PLAN.

Тот же SQL из Python

Модуль sqlite3, входящий в стандартную библиотеку, устанавливать не требуется.

import sqlite3

# открываем (или создаём) файл базы; ":memory:" — временная база в памяти
con = sqlite3.connect("lab.db")

# в SQLite проверка внешних ключей по умолчанию выключена — включаем
con.execute("PRAGMA foreign_keys = ON")

cur = con.cursor()  # курсор выполняет запросы и отдаёт результаты

# создаём таблицу, если её ещё нет (текст запроса — тот же SQL)
cur.execute("""
    CREATE TABLE IF NOT EXISTS measurements (
        id      INTEGER PRIMARY KEY,
        run_id  INTEGER NOT NULL,
        channel TEXT NOT NULL,
        t       REAL NOT NULL,
        value   REAL NOT NULL
    )
""")

# одиночная вставка: значения передаём кортежем, в запросе — знаки вопроса
cur.execute(
    "INSERT INTO measurements (run_id, channel, t, value) VALUES (?, ?, ?, ?)",
    (1, "BPM01", 0.0, 0.412),
)

# массовая вставка: executemany принимает любой итерируемый объект;
# read_adc() здесь — пользовательская функция опроса АЦП
points = [(1, "BPM01", 0.1 * i, read_adc()) for i in range(1000)]
cur.executemany(
    "INSERT INTO measurements (run_id, channel, t, value) VALUES (?, ?, ?, ?)",
    points,
)
con.commit()  # фиксируем изменения на диске

# выборка: по курсору можно итерироваться, каждая строка — кортеж
for t, value in cur.execute(
    "SELECT t, value FROM measurements WHERE channel = ? ORDER BY t",
    ("BPM01",),
):
    print(t, value)

con.close()

Для небольших выборок удобны cur.fetchone() и cur.fetchall(), однако итерация по курсору экономнее: строки читаются по мере надобности и не оседают в памяти все сразу.

Параметризованные запросы и SQL-инъекции

В примерах выше значения подставлялись через знаки вопроса, а не f-строками, и причиной этого является безопасность. Пусть имя канала приходит извне: из формы на веб-странице, из аргумента командной строки или из присланного файла.

channel = input("Канал: ")

# ТАК ДЕЛАТЬ НЕЛЬЗЯ: запрос склеивается из чужого текста
cur.execute(f"SELECT t, value FROM measurements WHERE channel = '{channel}'")

Если пользователь введёт ' OR '1'='1, итоговый запрос превратится в ... WHERE channel = '' OR '1'='1', где условие всегда истинно, фильтр исчезает, а наружу утекает вся таблица. Ввод вида '; DROP TABLE measurements; -- в СУБД, чей драйвер исполняет несколько команд подряд, уничтожит таблицу. Это явление называется SQL-инъекцией, одной из самых старых и до сих пор распространённых уязвимостей (см. xkcd про школьника Bobby Tables).

Параметризованный запрос неуязвим по построению: текст запроса и данные передаются драйверу раздельно, и введённая строка остаётся строкой при любом содержимом:

# правильно: запрос отдельно, данные отдельно
cur.execute("SELECT t, value FROM measurements WHERE channel = ?", (channel,))

Значения передаются только через параметры, всегда, даже если данные собственные. Параметром может быть лишь значение, но не имя таблицы или столбца; имена, подставляемые динамически, необходимо проверять по белому списку.

Транзакции и ACID

Транзакция — группа запросов, выполняемая как единое целое: применяются либо все, либо ни один. Её гарантии описывает аббревиатура ACID:

  • Atomicity (атомарность) означает неделимость транзакции: сбой, случившийся на середине, откатывает всё сделанное;
  • Consistency (согласованность) требует, чтобы база переходила из одного корректного состояния в другое, не нарушая ограничений (ключи, NOT NULL, UNIQUE);
  • Isolation (изолированность) скрывает промежуточные состояния, возникающие в параллельных транзакциях, друг от друга;
  • Durability (долговечность) сохраняет подтверждённые (COMMIT) данные даже при внезапном отключении питания.

Заход и снятые в нём точки должны попасть в базу вместе: заход без точек и точки без захода одинаково бессмысленны.

def save_run(con, run_row, points):
    """Сохраняет заход и все его измерения одной транзакцией."""
    cur = con.cursor()
    try:
        # первый INSERT неявно открывает транзакцию
        cur.execute(
            "INSERT INTO runs "
            "(experiment_id, run_number, energy_mev, started_at) "
            "VALUES (?, ?, ?, ?)",
            run_row,
        )
        run_id = cur.lastrowid  # id только что вставленной строки
        cur.executemany(
            "INSERT INTO measurements (run_id, channel, t, value) "
            "VALUES (?, ?, ?, ?)",
            [(run_id, ch, t, v) for ch, t, v in points],
        )
        con.commit()    # всё получилось — фиксируем
    except sqlite3.Error:
        con.rollback()  # любая ошибка — откатываем целиком
        raise

Ту же пару commit/rollback обеспечивает конструкция with con:, подтверждающая транзакцию при выходе из блока и откатывающая её при исключении. Транзакции ускоряют и массовую вставку: миллион INSERT с commit после каждого означает миллион синхронизаций с диском, а те же вставки в одной транзакции выполняются на порядки быстрее.

SQLite, PostgreSQL или MySQL

Три реляционные СУБД, чаще всего встречающиеся в лаборатории, различаются не языком запросов (SQL везде почти один и тот же), а тем, кто и каким образом получает доступ к базе.

SQLitePostgreSQLMySQL / MariaDB
Архитектуравстраиваемая библиотека, вся база в одном файлеклиент-сервернаяклиент-серверная
Установкане нужна, есть в Pythonотдельный сервисотдельный сервис
Одновременная записьодин пишущий процессмного клиентовмного клиентов
Типизациядинамическая, мягкаястрогая, богатая (массивы, JSONB, диапазоны)строгая, скромнее
Права доступанет (права на файл)пользователи, роли, гранулярные правапользователи и права
Сильная сторонанулевая настройка, переносимостьфункциональность и расширения (PostGIS, TimescaleDB)распространённость в веб-хостинге
Типичный случайлокальные данные, прототипы, файлы до единиц ГБсерверное приложение, общая база группывеб-проекты, легаси-системы

Если данные лежат на локальном диске и работают с ними один исследователь и его скрипты, то выбирается SQLite. Если база нужна нескольким людям или сервисам по сети, если в неё пишут одновременно и требуются права доступа, то предпочтительнее PostgreSQL [17]: сегодня это выбор по умолчанию для серверной СУБД. MySQL чаще достаётся как данность вместе с унаследованным проектом или хостингом, чем выбирается осознанно для новой системы.

ORM: SQLAlchemy

SQL в виде строк внутри Python-кода не проверяется до запуска, результаты приходят кортежами, а логика, связывающая объект с его окружением, оказывается распределённой по запросам. ORM (Object-Relational Mapping) отображает таблицы на классы, строки на объекты, а внешние ключи на атрибуты-связи. Стандартом де-факто в Python является SQLAlchemy.

from sqlalchemy import ForeignKey, create_engine, select
from sqlalchemy.orm import (
    DeclarativeBase, Mapped, mapped_column, relationship, Session,
)

class Base(DeclarativeBase):
    pass

class Run(Base):
    __tablename__ = "runs"

    id: Mapped[int] = mapped_column(primary_key=True)
    energy_mev: Mapped[float]
    # связь "один ко многим": у захода — список измерений
    measurements: Mapped[list["Measurement"]] = relationship(
        back_populates="run"
    )

class Measurement(Base):
    __tablename__ = "measurements"

    id: Mapped[int] = mapped_column(primary_key=True)
    run_id: Mapped[int] = mapped_column(ForeignKey("runs.id"))
    channel: Mapped[str]
    value: Mapped[float]
    run: Mapped[Run] = relationship(back_populates="measurements")

engine = create_engine("sqlite:///lab.db")  # та же база, поменяется только URL
Base.metadata.create_all(engine)            # создаёт таблицы по классам

with Session(engine) as session:
    # объекты вместо INSERT: связи расставляются сами
    run = Run(energy_mev=12.5)
    run.measurements.append(Measurement(channel="BPM01", value=0.412))
    session.add(run)
    session.commit()

    # запрос вместо SELECT ... JOIN: строится из Python-выражений
    stmt = select(Measurement).join(Run).where(Run.energy_mev > 10)
    for m in session.execute(stmt).scalars():
        print(m.channel, m.value, m.run.energy_mev)

Смена SQLite на PostgreSQL сводится к правке одной строки в create_engine; остальной код не изменяется.

ORM оправдан в приложении с десятком связанных таблиц и развивающейся схемой (накопленные миграции удобно вести инструментом Alembic), при командной разработке и множестве типовых операций «создать-прочитать-обновить-удалить». Чистый SQL предпочтителен для аналитических запросов с многоуровневыми агрегатами и оконными функциями (SQL здесь короче ORM-конструкций), для массовых загрузок и разовых скриптов. ORM не освобождает от знания SQL: он генерирует тот же SQL, и когда запрос выполняется медленно, разбираться приходится со сгенерированным текстом (для этого достаточно включить create_engine(..., echo=True) и просмотреть запросы, уходящие в базу).

Нереляционные базы данных

Под конкретные профили нагрузки существуют специализированные базы, объединяемые общим названием «NoSQL». Их устройство и цена каждого компромисса подробно рассмотрены у Клеппмана [16].

Проще всех устроено хранилище пар «ключ → значение», Redis. Оно располагается в оперативной памяти, и операции занимают микросекунды. Типичные роли: кеш (результат тяжёлого запроса или расчёта помещается под ключ с заданным временем жизни), очереди задач между процессами (на Redis работают брокеры для Celery), счётчики и pub/sub-уведомления. Запись на диск включается по желанию. Redis отвечает за скорость, а не за главное хранилище истины; типичные приёмы собраны у Карлсона [19].

Там, где у записей нет общей структуры, применяют документные базы наподобие MongoDB. Единицей хранения служит документ, вложенная структура наподобие JSON; документы собираются в коллекции, и жёсткой схемы нет, поэтому соседние документы могут иметь разные поля. Это удобно для разнородных метаданных, меняющих структуру от записи к записи. «Отсутствие схемы» означает, что схема находится не в базе, а в коде, и проверять её также приходится разработчику. Практические следствия рассмотрены в руководстве по MongoDB [18].

Телеметрию установки (давление в вакуумной камере, токи магнитов, температуры), приходящую с сотен датчиков каждую секунду годами, в ускорительной технике называют slow control и хранят в базах временных рядов. InfluxDB предназначена для потока «метка времени → значение» с тегами. Такие базы позволяют автоматически прореживать и удалять устаревшие данные (retention policies), быстро агрегировать по окнам времени («среднее за каждую минуту последних суток») и стыкуются с Grafana для дашбордов мониторинга. В мире PostgreSQL то же обеспечивает расширение TimescaleDB.

Наконец, для аналитики на миллиардах записей созданы колоночные базы, самая известная из которых называется ClickHouse. Она хранит данные не по строкам, а по столбцам. Значения одного столбца располагаются рядом, хорошо сжимаются, и запрос «среднее value по миллиарду строк» читает с диска только нужные столбцы, укладываясь в секунды на обычном сервере вместо часов; диалект SQL привычный. Платой является слабость в точечных обновлениях и удалениях: колоночные базы рассчитаны на дозапись и анализ, а не на правку отдельных строк.

Выбор хранилища: пример реальной системы

Система хранения данных ускорительного комплекса использует сразу три базы. Такой подход называется полиглотным хранением (polyglot persistence): под каждый профиль данных выбирается своё хранилище.

В PostgreSQL помещается всё, что должно быть строгим: пользователи и их роли, конфигурация ускорителя и его элементов, метаданные экспериментов (что за эксперимент, когда, кто ответственный), расписание работы установки и журнал событий. Структура этих данных известна заранее, меняется редко, а целостность критична: если запись эксперимента ссылается на несуществующего оператора, это ошибка, обнаруживать которую должна база, а не код. Здесь необходимы схема, внешние ключи и ACID-транзакции.

В MongoDB помещается всё разнородное: результаты измерений, данные пучка (траектории, размеры, интенсивность), временные ряды параметров, результаты моделирования и оптимизации. У этих данных нет единой схемы: каждый новый тип эксперимента приносит свой набор полей. Требование миграции базы под каждую новую методику приводит к тому, что данные складывают мимо базы, в файлы на рабочем столе.

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

Выбор сводится к одному признаку на каждую базу.

Признак данныхПодходящая база
Структура известна заранее, а нарушение связей недопустимоPostgreSQL
Структура меняется от записи к записиMongoDB
Данные нужны очень быстро, а их потеря не смертельнаRedis

Цена полиглотного хранения: как только данные распределяются по трём базам, исчезает транзакция, охватывающая их все, и одним атомарным действием уже нельзя записать эксперимент в PostgreSQL вместе с его результатами в MongoDB. Согласованность приходится обеспечивать вручную, распределёнными транзакциями или компенсирующими действиями, откатывающими то, что успело записаться. Эта сложность окупается, только когда профили данных действительно различны.

В студенческом проекте целесообразно начинать с одной базы. Одна PostgreSQL закрывает потребности лабораторного сервиса на годы вперёд, а для документов в ней есть тип JSONB, индексируемый по полям внутри документа. Вторую базу заводят не потому, что так делают в больших системах, а когда первая измеримо перестала справляться с конкретным профилем нагрузки.

Одна задача в четырёх хранилищах

Рассмотрим учебный сервис, список записей с операциями «прочитать страницу», «добавить», «изменить», «удалить», и переложим его последовательно в четыре хранилища: текстовый файл со строками произвольной длины, файл с записями фиксированной длины, реляционную базу и кеш поверх неё.

Код почти не меняется от варианта к варианту; меняется то, где лежат данные и во что обходится каждая операция. Сложность операций рассматривалась в главе «Основные структуры данных».

Сервис написан на микрофреймворке Flask: декоратор @app.route привязывает функцию к URL и методу, объект request даёт доступ к параметрам и телу запроса. Заголовочная часть общая для всех вариантов и далее не повторяется.

import json
from contextlib import closing

import psycopg2
import redis
from flask import Flask, request

app = Flask(__name__)

FILE_PATH = 'data.txt'
PAGE_SIZE = 20
ELEMENT_SIZE = 64
FILL_CHAR = b' '
TTL = 60
postgres_creds = {'dbname': 'lab', 'host': 'localhost'}
redis_creds = {'host': 'localhost', 'port': 6379}

Наивное решение. GET

Самый прямолинейный вариант — обычный текстовый файл, по строке на запись. Чтобы отдать одну страницу, приходится прочитать весь файл, а при изменении и удалении ещё и переписать его заново: \(O(n)\) на каждую операцию.

@app.route('/', methods=['GET'])
def paginated_get():
    page = int(request.args.get('page', '0'))
    first = page * PAGE_SIZE
    last = first + PAGE_SIZE
    result = []
    with open(FILE_PATH, 'r') as f:
        for i, line in enumerate(f.readlines()[first:last]):
            result.append({'id': first + i, 'data': line.strip()})
    return {"result": result}

Наивное решение. POST

@app.route('/', methods=['POST'])
def post():
    data = request.json['data']
    with open(FILE_PATH, 'a') as f:
        f.write(data + '\n')
    return {}, 201

Наивное решение. PUT

@app.route('/<int:data_id>', methods=['PUT'])
def put(data_id):
    data = request.json['data']
    with open(FILE_PATH, 'r') as f:
        new_data = f.readlines()
        new_data[data_id] = data + '\n'
    with open(FILE_PATH, 'w') as f:
        f.writelines(new_data)
    return {}, 204

Наивное решение. DELETE

@app.route('/<int:data_id>', methods=['DELETE'])
def delete(data_id):
    with open(FILE_PATH, 'r') as f:
        new_data = f.readlines()
    del new_data[data_id]
    with open(FILE_PATH, 'w') as f:
        f.writelines(new_data)
    return {}, 204

Фиксированные записи. GET

Если отвести каждой записи одинаковое число байт, файл превращается в массив, и нужная запись читается сразу, переходом к её смещению через seek(). Зато удаление дорожает: хвост файла за ней приходится сдвигать вручную.

@app.route('/', methods=['GET'])
def paginated_get():
    page = int(request.args.get('page', '0'))
    first = page * PAGE_SIZE
    result = []
    with open(FILE_PATH, 'rb') as f:
        f.seek(first * ELEMENT_SIZE)
        data = f.read(PAGE_SIZE * ELEMENT_SIZE)
        for i in range(PAGE_SIZE):
            result.append(
                {
                    "id": first + i,
                    "data": (
                        data[i * ELEMENT_SIZE: (i + 1) * ELEMENT_SIZE].
                        strip(FILL_CHAR).decode('utf-8')
                    )
                }
            )
    return {"result": result}

Фиксированные записи. POST

@app.route('/', methods=['POST'])
def post():
    data = str(request.json['data'])
    with open(FILE_PATH, 'a+b') as f:
        f.write(data.encode('utf-8').ljust(ELEMENT_SIZE, FILL_CHAR))
    return {}, 201

Фиксированные записи. PUT

@app.route('/<int:data_id>', methods=['PUT'])
def put(data_id):
    data = request.json['data']
    with open(FILE_PATH, 'r+b') as f:
        f.seek(data_id * ELEMENT_SIZE)
        f.write(data.encode('utf-8').ljust(ELEMENT_SIZE, FILL_CHAR))
    return {}, 204

Фиксированные записи. DELETE

@app.route('/<int:data_id>', methods=['DELETE'])
def delete(data_id):
    with open(FILE_PATH, 'r+b') as f:
        point = data_id * ELEMENT_SIZE
        f.seek(point)
        while True:
            f.seek(point + ELEMENT_SIZE)
            complex_data = f.read(ELEMENT_SIZE)
            f.seek(point)
            if len(complex_data):
                f.write(complex_data)
                point += ELEMENT_SIZE
            else:
                f.truncate()
                break
    return {}, 204

Реляционная база. GET

База данных скрывает эту механику за индексами: поиск по ключу, вставку и удаление она выполняет не хуже структур данных из главы «Основные структуры данных».

@app.route('/', methods=['GET'])
def paginated_get():
    page = int(request.args.get('page', '0'))
    with closing(psycopg2.connect(**postgres_creds)) as conn:
        with conn.cursor() as cursor:
            cursor.execute(
                'SELECT "id", "todo" FROM "todos" '
                'ORDER BY "id" OFFSET %s LIMIT %s;',
                (page * PAGE_SIZE, PAGE_SIZE)
            )
            return {
                "result": [{"id": row[0], "data": row[1]} for row in cursor]
            }

Реляционная база. POST

@app.route('/', methods=['POST'])
def post():
    data = str(request.json['data'])
    with closing(psycopg2.connect(**postgres_creds)) as conn:
        with conn.cursor() as cursor:
            cursor.execute(
                'INSERT INTO "todos" ("todo") VALUES (%s) RETURNING "id";',
                (data,)
            )
            conn.commit()
            return {"id": cursor.fetchone()[0]}, 201

Реляционная база. PUT

@app.route('/<int:data_id>', methods=['PUT'])
def put(data_id):
    data = request.json['data']
    with closing(psycopg2.connect(**postgres_creds)) as conn:
        with conn.cursor() as cursor:
            cursor.execute(
                'UPDATE "todos" SET "todo" = %s WHERE "id" = %s;',
                (data, data_id)
            )
            conn.commit()
            return {}, 204

Реляционная база. DELETE

@app.route('/<int:data_id>', methods=['DELETE'])
def delete(data_id):
    with closing(psycopg2.connect(**postgres_creds)) as conn:
        with conn.cursor() as cursor:
            cursor.execute(
                'DELETE FROM "todos" WHERE "id" = %s;',
                (data_id,)
            )
            conn.commit()
            return {}, 204

Кеш. GET

Наконец, самые частые запросы можно не доводить до базы: готовый ответ помещается в Redis и в следующий раз отдаётся из оперативной памяти.

@app.route('/', methods=['GET'])
def paginated_get():
    page = int(request.args.get('page', '0'))

    redis_client = redis.Redis(**redis_creds)
    cached_page = redis_client.get(page)

    if cached_page:
        return cached_page

    with closing(psycopg2.connect(**postgres_creds)) as conn:
        with conn.cursor() as cursor:
            cursor.execute(
                'SELECT "id", "todo" FROM "todos" '
                'ORDER BY "id" OFFSET %s LIMIT %s;',
                (page * PAGE_SIZE, PAGE_SIZE)
            )
            result = json.dumps({
                "result": [{"id": row[0], "data": row[1]} for row in cursor]
            })
            redis_client.set(page, result, ex=TTL)
            return result

Обратная сторона кеша: клиент, выполнивший POST или PUT, не увидит собственной правки, пока не истечёт TTL, и чем длиннее время жизни ключа, тем дольше сервис отдаёт устаревшее. Поэтому каждая изменяющая операция обязана сбрасывать затронутые ключи, и трудность заключается в том, чтобы определить, какие именно: правка обесценивает ту страницу, на которой находится запись, а вставка и удаление сдвигают нумерацию и обесценивают всё последующее.

Кеш. POST

Вставка попадает в конец списка, однако выяснять, где он заканчивается, дороже, чем сбросить все страницы.

@app.route('/', methods=['POST'])
def post():
    data = str(request.json['data'])
    with closing(psycopg2.connect(**postgres_creds)) as conn:
        with conn.cursor() as cursor:
            cursor.execute(
                'INSERT INTO "todos" ("todo") VALUES (%s) RETURNING "id";',
                (data,)
            )
            conn.commit()
            new_id = cursor.fetchone()[0]

    redis_client = redis.Redis(**redis_creds)
    cached_pages = redis_client.keys()  # в кеше лежат только страницы
    if cached_pages:
        redis_client.delete(*cached_pages)
    return {"id": new_id}, 201

Кеш. PUT

Правка не сдвигает нумерацию, поэтому номер страницы вычисляется по числу записей, лежащих перед изменённой, и сбрасывается один ключ.

@app.route('/<int:data_id>', methods=['PUT'])
def put(data_id):
    data = request.json['data']
    with closing(psycopg2.connect(**postgres_creds)) as conn:
        with conn.cursor() as cursor:
            cursor.execute(
                'UPDATE "todos" SET "todo" = %s WHERE "id" = %s;',
                (data, data_id)
            )
            cursor.execute(
                'SELECT count(*) FROM "todos" WHERE "id" < %s;',
                (data_id,)
            )
            page = cursor.fetchone()[0] // PAGE_SIZE
            conn.commit()

    redis_client = redis.Redis(**redis_creds)
    redis_client.delete(page)
    return {}, 204

Кеш. DELETE

После удаления всё последующее сдвигается на одну позицию вперёд, поэтому устаревают и страница удалённой записи, и каждая следующая за ней.

@app.route('/<int:data_id>', methods=['DELETE'])
def delete(data_id):
    with closing(psycopg2.connect(**postgres_creds)) as conn:
        with conn.cursor() as cursor:
            cursor.execute(
                'SELECT count(*) FROM "todos" WHERE "id" < %s;',
                (data_id,)
            )
            page = cursor.fetchone()[0] // PAGE_SIZE
            cursor.execute(
                'DELETE FROM "todos" WHERE "id" = %s;',
                (data_id,)
            )
            conn.commit()

    redis_client = redis.Redis(**redis_creds)
    stale = [key for key in redis_client.keys() if int(key) >= page]
    if stale:
        redis_client.delete(*stale)
    return {}, 204

Pandas и базы данных

Функция read_sql в pandas выполняет запрос и возвращает DataFrame, а to_sql записывает DataFrame в таблицу.

import pandas as pd
import sqlite3

con = sqlite3.connect("lab.db")

# фильтрация и агрегация выполняются в базе,
# в память попадает только готовая сводка
df = pd.read_sql(
    """
    SELECT r.energy_mev, AVG(m.value) AS mean_signal
    FROM measurements AS m
    JOIN runs AS r ON r.id = m.run_id
    WHERE m.channel = 'BPM01'
    GROUP BY r.id
    """,
    con,
)

# дальше — обычный pandas: график зависимости сигнала от энергии
df.plot.scatter(x="energy_mev", y="mean_signal")

# и обратно: результат анализа — в новую таблицу
df.to_sql("run_summary", con, if_exists="replace", index=False)

Тяжёлую фильтрацию и агрегацию следует передавать базе (SQL с WHERE и GROUP BY), а в pandas загружать свёрнутый результат: память и время расходуются на порядки экономнее, чем при чтении всех данных и фильтрации в DataFrame. Связка «SQL-запрос → DataFrame → график» даёт простейшее приложение для визуализации данных; построение графиков рассматривается в главах про обработку и визуализацию данных.

С чего начать в своей лаборатории

Рецепт перехода от тысячи CSV к одной базе включает следующие шаги:

  1. Завести один файл lab.db и описать схему. Минимально необходимы таблицы вида experiments, runs, measurements с первичными и внешними ключами и NOT NULL на важных полях. Схема служит документацией, проверяющей сама себя.

  2. Импортировать накопленные CSV одним скриптом:

    from pathlib import Path
import pandas as pd
import sqlite3

con = sqlite3.connect("lab.db")

for csv_path in Path("data").glob("**/*.csv"):
    df = pd.read_csv(csv_path)
    df["source_file"] = str(csv_path)  # не теряем происхождение данных
    df.to_sql("measurements_raw", con, if_exists="append", index=False)

    Далее «сырую» таблицу можно разложить по нормализованной схеме SQL-запросами.

  3. Новые данные записывать сразу в базу, применяя в скрипте сбора executemany группами точек и commit раз в несколько секунд, а не файл на каждый заход.

  4. Включить PRAGMA journal_mode=WAL: в этом режиме SQLite позволяет читать базу (строить графики) параллельно с записью.

  5. Просматривать данные удобно в DB Browser for SQLite, а резервной копией служит обычная копия одного файла (Connection.backup() из Python выполняет её корректно даже во время работы).

Когда данными начнёт пользоваться вся группа с нескольких машин, схема переносится в PostgreSQL; SQL и почти весь код при этом остаются прежними.

Полезные ссылки

Задание. Собрать и развернуть всё вместе, приложение, контейнеры, сеть и базу: «Деплой стартапа».