ETL на Python: надёжный конвейер из CSV в SQLite

Повторный импорт CSV легко задваивает записи, а загрузка по одной строке без транзакции может оставить базу частично обновлённой. Покажем компактный ETL на Python 3.12: он проверяет входные данные, нормализует их и загружает в SQLite так, чтобы повторный запуск не создавал копии.

Сначала договоритесь о форме данных

Пример рассчитан на CSV с заказами и файл SQLite. В источнике должны быть поля order_id, customer_id, order_date и amount. Дата записана как YYYY-MM-DD, сумма — десятичное число с точкой. Идентификатор заказа уникален: по нему конвейер поймёт, какую запись обновлять при повторной загрузке.

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

Версии и входные файлы

Код использует стандартную библиотеку Python 3.12 и SQLite версии 3.24.0 или новее: в этой версии SQLite появилась конструкция ON CONFLICT ... DO UPDATE. Версию SQLite, с которой собран ваш Python, проверяет сам скрипт. Нужен файл CSV в кодировке UTF-8; отдельные пакеты устанавливать не требуется.

Сохраните CSV, например, как orders.csv. Скрипт получит путь к нему и к файлу базы через аргументы командной строки. Если таблицы ещё нет, она будет создана.

Полный пример конвейера

Сохраните код в etl.py. Конвейер читает файл потоком, отправляет записи в базу партиями по тысяче и держит импорт в одной транзакции: если проверка или запись завершится ошибкой, изменения текущего запуска откатятся.

import argparse
import csv
import sqlite3
from datetime import date
from decimal import Decimal, InvalidOperation

BATCH_SIZE = 1000
REQUIRED_COLUMNS = {"order_id", "customer_id", "order_date", "amount"}


def required_value(row, column, line_number):
    value = row.get(column)
    if value is None or not value.strip():
        raise ValueError(f"строка {line_number}: пустое поле {column}")
    return value.strip()


def transform(row, line_number):
    order_id = required_value(row, "order_id", line_number)
    customer_id = required_value(row, "customer_id", line_number)
    raw_date = required_value(row, "order_date", line_number)
    raw_amount = required_value(row, "amount", line_number)

    try:
        order_date = date.fromisoformat(raw_date).isoformat()
    except ValueError as error:
        raise ValueError(
            f"строка {line_number}: неверная дата {raw_date!r}; нужен формат YYYY-MM-DD"
        ) from error

    try:
        amount = Decimal(raw_amount)
        if not amount.is_finite():
            raise ValueError("сумма должна быть конечным числом")
        rounded = amount.quantize(Decimal("0.01"))
    except (InvalidOperation, ValueError) as error:
        raise ValueError(
            f"строка {line_number}: неверная сумма {raw_amount!r}"
        ) from error

    if amount != rounded:
        raise ValueError(
            f"строка {line_number}: у суммы больше двух знаков после точки"
        )

    return (order_id, customer_id, order_date, int(rounded * 100))


def run(csv_path, database_path):
    if sqlite3.sqlite_version_info < (3, 24, 0):
        raise ValueError(
            f"нужен SQLite 3.24.0 или новее; используется {sqlite3.sqlite_version}"
        )

    connection = sqlite3.connect(database_path)
    loaded = 0
    try:
        with connection:
            connection.execute(
                """CREATE TABLE IF NOT EXISTS orders (
                    order_id TEXT PRIMARY KEY,
                    customer_id TEXT NOT NULL,
                    order_date TEXT NOT NULL,
                    amount_cents INTEGER NOT NULL
                )"""
            )
            with open(csv_path, "r", encoding="utf-8-sig", newline="") as source:
                reader = csv.DictReader(source)
                if reader.fieldnames is None:
                    raise ValueError("CSV пуст или в нём нет строки заголовков")
                missing = REQUIRED_COLUMNS - set(reader.fieldnames)
                if missing:
                    raise ValueError(
                        "в CSV не хватает колонок: " + ", ".join(sorted(missing))
                    )

                batch = []
                for row in reader:
                    batch.append(transform(row, reader.line_num))
                    if len(batch) == BATCH_SIZE:
                        connection.executemany(
                            """INSERT INTO orders
                               (order_id, customer_id, order_date, amount_cents)
                               VALUES (?, ?, ?, ?)
                               ON CONFLICT(order_id) DO UPDATE SET
                                   customer_id = excluded.customer_id,
                                   order_date = excluded.order_date,
                                   amount_cents = excluded.amount_cents""",
                            batch,
                        )
                        loaded += len(batch)
                        batch.clear()

                if batch:
                    connection.executemany(
                        """INSERT INTO orders
                           (order_id, customer_id, order_date, amount_cents)
                           VALUES (?, ?, ?, ?)
                           ON CONFLICT(order_id) DO UPDATE SET
                               customer_id = excluded.customer_id,
                               order_date = excluded.order_date,
                               amount_cents = excluded.amount_cents""",
                        batch,
                    )
                    loaded += len(batch)
    finally:
        connection.close()
    return loaded


def main():
    parser = argparse.ArgumentParser(description="Загрузка заказов из CSV в SQLite")
    parser.add_argument("csv_path", help="путь к CSV-файлу")
    parser.add_argument("database_path", help="путь к файлу SQLite")
    args = parser.parse_args()

    try:
        loaded = run(args.csv_path, args.database_path)
    except (OSError, UnicodeError, ValueError, sqlite3.Error) as error:
        parser.exit(1, f"Ошибка ETL: {error}\n")
    print(f"Обработано строк: {loaded}")


if __name__ == "__main__":
    main()

Запуск из терминала: python etl.py orders.csv orders.sqlite. При успешной загрузке скрипт напечатает число обработанных строк. Проверьте результат запросом в любом клиенте SQLite: таблица orders должна содержать записи с целочисленной суммой в центах.

Что здесь защищает данные

Извлечение — чтение CSV через DictReader. Преобразование — проверка обязательных полей, формата даты и суммы. Загрузка — пакетная запись через executemany. SQL-параметры передаются отдельно от запроса, поэтому значения из файла не склеиваются с текстом SQL.

Записи с одним order_id обновляются, а не дублируются. Это пример идемпотентности — свойства, при котором повтор того же действия не меняет итоговый результат по сравнению с одним запуском. Здесь гарантия действует для повторной загрузки тех же заказов: если запись исчезла из нового CSV, скрипт не удалит её из базы.

Транзакция решает не вопрос «как импортировать быстрее», а вопрос «что останется в базе, если импорт оборвётся на середине».

Почему загрузка идёт партиями

Можно вызывать execute для каждой строки, а можно накопить несколько записей и передать их в executemany. В примере выбран второй вариант: он сокращает число обращений из Python к SQLite, но не требует держать весь CSV в памяти. Транзакция при этом одна на весь файл, а не отдельная на каждую партию.

Когда менять подход

Для небольшого CSV и одной локальной базы вариант со стандартной библиотекой проще сопровождать. Если преобразований много и данные уже обрабатываются в pandas, нет смысла переносить каждую операцию в циклы вручную. Если файл не помещается в доступное время импорта, проверьте объём, индексы и скорость диска; увеличение размера партии само по себе не гарантирует ускорения.

Что может сломаться

• Неподходящая версия SQLite. Скрипт завершится сообщением с фактической версией. Обновите Python-сборку или SQLite, а не убирайте проверку и не рассчитывайте на неподдерживаемый синтаксис.
• Заголовки отличаются от ожидаемых. Сообщение перечислит недостающие колонки. Сопоставьте имена в исходном CSV с контрактом или явно измените список REQUIRED_COLUMNS и места чтения полей.
• В сумме запятая вместо точки. Значение вроде 12,50 не соответствует принятому формату и будет отклонено. Исправьте экспорт либо добавьте отдельное, явно протестированное преобразование локального формата.
• В файле повторяется один ID. Последняя по порядку строка обновит предыдущую. Если такое поведение недопустимо, проверяйте уникальность идентификаторов в источнике и отклоняйте дубли до загрузки.
• Одна строка испорчена. Транзакция откатит изменения этого запуска, а сообщение укажет номер строки и поле. Исправьте источник и повторите запуск; частичного результата в таблице не останется.

Где этот вариант перестаёт подходить

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

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

Что делать дальше

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