ETL из S3: потоковый разбор и защита от повторной обработки

Условие

Описание ведется от бизнес аналитика, который написал задачу "как смог" и может давать бизнесов описания требований. Имеется жуткий легаси код написанный на PHP, который

получает файлы логов из S3 разбирает их, записывает результат в нужное место. Известно:

файлы приходят раз в час, сейчас 5 шт файлы лежат все в одном бакете Данные должны в зависимости от формата направляться в разные места – postgres/локальный файл Необходимо написать ETL, реализующий такой функционал. Написанием конкретной реализаций методов (считывания данных/ отправки данных и тд) можно пренебрегать

Ответ

Разделите ETL на обнаружение объектов S3, определение формата, потоковый разбор и запись через соответствующий sink. Повторный запуск не должен дублировать уже обработанные данные.

Сначала уточните размеры, форматы, схему, SLA, часовой пояс и способ определить завершённую загрузку. Ключ обработки — бакет, ключ объекта и версия, если versioning доступен; один ETag нельзя всегда считать хешем содержимого. Храните состояния обработки и ошибки, направляйте некорректные файлы в карантин. Запись должна быть идемпотентной, а статус complete выставляться после успешного sink.

for obj in source.discover():
    if registry.is_complete(obj.identity):
        continue
    parser = parsers.for_object(obj)
    sink = sinks.for_format(parser.format)
    sink.write_idempotently(obj.identity, parser.parse(source.open(obj)))
    registry.mark_complete(obj.identity)

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