ETL: выбор parser и sink, журнал версий файлов

Условие

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

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

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

Ответ

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

Каркас с внедрёнными адаптерами:

def run_etl(source, registry, ledger, logger):
    for obj in source.list_ready():
        key = (obj.bucket, obj.key, obj.version)
        if ledger.is_done(key):
            continue
        try:
            parser, sink = registry.for_format(obj.format)
            with source.open(obj) as stream:
                sink.write_idempotently(key, parser.iter_rows(stream))
            ledger.mark_done(key)
        except Exception:
            logger.exception('ETL item failed', extra={'object_key': obj.key})

Имена адаптеров условны. Postgres-sink использует транзакцию и ключ уникальности, файловый — временный файл и атомарную публикацию. mark_done после записи может не успеть выполниться, поэтому sink обязан выдерживать повтор. Для нескольких исполнителей добавьте lease; повреждённые форматы отправляйте в карантин, не угадывайте молча.