ETL пропускает плохие строки: как сделать карантин и сверку количества

Ночная загрузка стала зелёной после добавления try/except: теперь ошибочные строки просто пропускаются. Через неделю выясняется, что из отчёта исчезли десятки заказов. Обработка продолжалась, но никто не сохранил, какие записи не дошли до витрины.

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

Проверим содержимое, а не только синтаксис

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

rows = [
    {"id": "A", "qty": "2"},
    {"id": "B", "qty": "bad"},
    {"id": "A", "qty": "3"},
    {"id": "C", "qty": "0"},
    {"id": "D", "qty": "4"},
]
accepted, rejected, seen = [], [], set()
for number, row in enumerate(rows, start=1):
    reason = None
    if not row["id"] or row["id"] in seen:
        reason = "missing or duplicate id"
    elif not row["qty"].isascii() or not row["qty"].isdigit():
        reason = "qty must be ASCII integer"
    elif len(row["qty"]) > 6 or not 1 <= int(row["qty"]) <= 999999:
        reason = "qty outside range"
    if reason:
        rejected.append({"record": number, "row": row, "reason": reason})
    else:
        seen.add(row["id"])
        accepted.append({"id": row["id"], "qty": int(row["qty"])})
assert len(rows) == len(accepted) + len(rejected)
assert sum(x["qty"] for x in accepted) == 6
assert [x["record"] for x in rejected] == [2, 3, 4]
print("input", len(rows), "accepted", len(accepted), "rejected", len(rejected))
print(rejected)

Пять входов разделились на два принятых и три отклонённых. Итоговое количество равно шести. Номер record — номер бизнес-записи после парсинга; он не обязан совпадать с физической строкой CSV, поскольку поле может содержать перевод строки.

Здесь seen запоминает только принятые идентификаторы. Если первая запись A некорректна, следующая исправная A сможет пройти. Для другой политики — например, запрета любых повторов независимо от валидности — потребуется отдельный предварительный подсчёт. Правило дублей должно быть выбрано по смыслу поставки.

Отклонение строки и отказ всей партии

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

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

Карантин должен иметь выход

Сохраняйте идентификатор поставки, версию проверки и причину. После исправления источника можно переобработать исходный файл или конкретные записи по тому же устойчивому ключу. Не добавляйте исправления поверх витрины вслепую: повторная обработка не должна создавать второй заказ.

В рабочем конвейере принятые строки, карантин и отчёт о количестве публикуют согласованно. Иначе сбой после записи accepted, но до сохранения rejected разрушит проверяемость партии. Начните с файла-манифеста, который объявляет готовность всего набора только после успешного сохранения каждой части.

В курсе Data Engineering Fast-Track вы отработаете схемы и контракты, обработку в Spark и управление загрузками через Airflow. Откройте программу, чтобы собирать пайплайны, где отклонённые записи имеют понятный путь разбора и повторного запуска.