Проект конвейера данных

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

Особенности

• Поддержка нескольких источников данных: JSON, CSV, REST API
• Поддержка нескольких приемников данных: SQLite, CSV
• Сопоставление полей и проверка данных
• Обработка ошибок и ведение журнала
• Конфигурация на основе окружения
• Включены модульные тесты

Требования

• Python 3.8+
• Зависимости перечислены в
  requirements.txt

Установка

1. Клонируйте репозиторий
2. Установите зависимости:

Использование

Запуск конвейера

Примеры:

Доступные параметры

• --source
  : Тип источника (json, csv, api)
• --sink
  : Тип приемника (sqlite, csv)
• --source-path
  : Пользовательский путь к исходному файлу
• --sink-path
  : Пользовательский путь к файлу/базе данных приемника
• --api-url
  : Пользовательский URL API (для источника API)
• --truncate
  : Очистить существующие данные перед загрузкой

Тестирование источника API

Для тестирования с фиктивным сервером API:

Затем запустите конвейер:

Конфигурация

Переменные окружения можно задать в файле

:

• DB_HOST
  ,
  DB_PORT
  ,
  DB_NAME
  ,
  DB_USER
  ,
  DB_PASSWORD
  - Настройки базы данных
• API_URL
  ,
  API_TIMEOUT
  - Настройки API
• JSON_SOURCE_PATH
  ,
  CSV_SOURCE_PATH
  - Пути к исходным файлам
• OUTPUT_CSV_PATH
  ,
  OUTPUT_DB_PATH
  - Выходные пути
• LOG_LEVEL
  ,
  LOG_FILE
  - Настройки ведения журнала

Архитектура

Проект следует модульной схеме:

• extractors/
  : Реализации источников данных
• transformers/
  : Логика преобразования данных
• loaders/
  : Реализации приемников данных
• utils/
  : Вспомогательные функции, такие как ведение журнала
• tests/
  : Модульные тесты
• scripts/
  : Вспомогательные скрипты, такие как фиктивный сервер API

Схема данных

Структура целевой таблицы (

):