chore: update README
This commit is contained in:
parent
bde0bb0e6f
commit
74243fd258
149
README.md
149
README.md
|
|
@ -12,6 +12,7 @@
|
||||||
- Взаимодействие с БД
|
- Взаимодействие с БД
|
||||||
- HTTP API
|
- HTTP API
|
||||||
- Воркеры, пайплайны и добавление новых ETL‑задач
|
- Воркеры, пайплайны и добавление новых ETL‑задач
|
||||||
|
- Существующие пайплайны
|
||||||
- Логирование, метрики, аудит
|
- Логирование, метрики, аудит
|
||||||
- Тестирование
|
- Тестирование
|
||||||
- Эксплуатация и масштабирование
|
- Эксплуатация и масштабирование
|
||||||
|
|
@ -455,6 +456,124 @@ curl http://localhost:8081/api/v1/jobs/{job_id}/status
|
||||||
- `partition_key` используется для параллелизации независимых задач
|
- `partition_key` используется для параллелизации независимых задач
|
||||||
|
|
||||||
|
|
||||||
|
## Существующие пайплайны
|
||||||
|
|
||||||
|
### `load.tenera` - Загрузка котировок
|
||||||
|
|
||||||
|
**Описание:**
|
||||||
|
Загружает финансовые котировки (биржевые индексы, валюты, сырьевые товары) из SuperTenera API.
|
||||||
|
|
||||||
|
**Источник данных:**
|
||||||
|
- SuperTenera API - агрегатор котировок из множества источников (CBR, Investing.com, SGX, Bloomberg, TradingView, TradingEconomics)
|
||||||
|
|
||||||
|
**Назначение:**
|
||||||
|
- Таблицы в схеме `quotes`:
|
||||||
|
- `quote_section` - разделы котировок (CBR, SGX, и т.д.)
|
||||||
|
- `quote` - инструменты (тикеры, названия)
|
||||||
|
- `quote_value` - временные ряды значений (OHLCV + дополнительные метрики)
|
||||||
|
|
||||||
|
**Процесс:**
|
||||||
|
1. Запрос данных из SuperTenera API
|
||||||
|
2. Парсинг различных форматов (каждый источник имеет свою структуру)
|
||||||
|
3. Трансформация в единый формат с преобразованием временных зон
|
||||||
|
4. UPSERT в БД (идемпотентная вставка/обновление)
|
||||||
|
|
||||||
|
**Пример запуска:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Настроить воркер в .env
|
||||||
|
WORKERS_JSON='[{"queue":"load.tenera","concurrency":1}]'
|
||||||
|
|
||||||
|
# Запустить задачу через API
|
||||||
|
curl -X POST http://localhost:8081/api/v1/jobs/trigger \
|
||||||
|
-H "Content-Type: application/json" \
|
||||||
|
-d '{
|
||||||
|
"queue": "load.tenera",
|
||||||
|
"task": "load.tenera",
|
||||||
|
"args": {},
|
||||||
|
"lock_key": "tenera_quotes",
|
||||||
|
"priority": 100,
|
||||||
|
"max_attempts": 3,
|
||||||
|
"lease_ttl_sec": 300
|
||||||
|
}'
|
||||||
|
```
|
||||||
|
|
||||||
|
**Особенности:**
|
||||||
|
- Поддержка множества форматов временных меток (ISO, Unix timestamp, кастомные форматы)
|
||||||
|
- Автоматическое преобразование временных зон в Europe/Moscow
|
||||||
|
- Обработка разных структур данных от различных провайдеров
|
||||||
|
- UPSERT по композитному ключу (quote_id + date/timestamp)
|
||||||
|
|
||||||
|
|
||||||
|
### `load.opu` - Загрузка данных OPU
|
||||||
|
|
||||||
|
**Описание:**
|
||||||
|
Загружает данные OPU из Gmap2Brief API. Данные выгружаются в виде zstandard-сжатого архива с JSON Lines.
|
||||||
|
|
||||||
|
**Источник данных:**
|
||||||
|
- Gmap2Brief API - экспорт данных о структуре OPU
|
||||||
|
|
||||||
|
**Назначение:**
|
||||||
|
- Таблица `brief_digital_certificate_opu` в схеме `opu` (23 поля, композитный PK из 8 полей)
|
||||||
|
|
||||||
|
**Процесс:**
|
||||||
|
1. Запуск экспорта через API (`/export/opu/start`)
|
||||||
|
2. Polling статуса экспорта до завершения
|
||||||
|
3. Скачивание zstandard-архива с JSON Lines
|
||||||
|
4. TRUNCATE целевой таблицы (полная перезагрузка)
|
||||||
|
5. Потоковая распаковка архива (64KB чанки)
|
||||||
|
6. Парсинг JSON Lines и батчевая вставка (по 5000 записей)
|
||||||
|
7. Преобразование ISO-дат в date/datetime объекты
|
||||||
|
|
||||||
|
**Пример запуска:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Настроить воркер в .env
|
||||||
|
WORKERS_JSON='[{"queue":"load.opu","concurrency":1}]'
|
||||||
|
|
||||||
|
# Запустить задачу через API
|
||||||
|
curl -X POST http://localhost:8081/api/v1/jobs/trigger \
|
||||||
|
-H "Content-Type: application/json" \
|
||||||
|
-d '{
|
||||||
|
"queue": "load.opu",
|
||||||
|
"task": "load.opu",
|
||||||
|
"args": {},
|
||||||
|
"lock_key": "opu_export",
|
||||||
|
"priority": 100,
|
||||||
|
"max_attempts": 2,
|
||||||
|
"lease_ttl_sec": 600
|
||||||
|
}'
|
||||||
|
```
|
||||||
|
|
||||||
|
**Особенности:**
|
||||||
|
- Потоковая обработка больших архивов без полной загрузки в память
|
||||||
|
- Zstandard декомпрессия с буферизацией неполных строк
|
||||||
|
- TRUNCATE перед загрузкой (стратегия полной замены данных)
|
||||||
|
- Батчевая вставка для оптимизации производительности
|
||||||
|
- Heartbeat после каждого батча для отслеживания прогресса
|
||||||
|
- Композитный первичный ключ из 8 полей обеспечивает уникальность
|
||||||
|
|
||||||
|
|
||||||
|
### `noop` - Тестовый пайплайн
|
||||||
|
|
||||||
|
**Описание:**
|
||||||
|
Демонстрационный пайплайн для тестирования инфраструктуры очереди. Выполняет серию sleep-операций с heartbeat.
|
||||||
|
|
||||||
|
**Пример запуска:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
curl -X POST http://localhost:8081/api/v1/jobs/trigger \
|
||||||
|
-H "Content-Type: application/json" \
|
||||||
|
-d '{
|
||||||
|
"queue": "etl.default",
|
||||||
|
"task": "noop",
|
||||||
|
"args": {"sleep1": 2, "sleep2": 3, "sleep3": 1},
|
||||||
|
"lock_key": "test_noop",
|
||||||
|
"priority": 100
|
||||||
|
}'
|
||||||
|
```
|
||||||
|
|
||||||
|
|
||||||
## Логирование, метрики, аудит
|
## Логирование, метрики, аудит
|
||||||
- Логи: структурированные, через `logger/*`. Middleware (`api/middleware.py`) логирует входящие запросы/исходящие ответы, время обработки, пишет метрики и аудит‑события.
|
- Логи: структурированные, через `logger/*`. Middleware (`api/middleware.py`) логирует входящие запросы/исходящие ответы, время обработки, пишет метрики и аудит‑события.
|
||||||
- Метрики: простые счётчики (пример: likes/dislikes, requests_total, responses_total, duration_ms).
|
- Метрики: простые счётчики (пример: likes/dislikes, requests_total, responses_total, duration_ms).
|
||||||
|
|
@ -612,11 +731,6 @@ TOTAL 778 35 95.50%
|
||||||
|
|
||||||
**Решение:**
|
**Решение:**
|
||||||
```bash
|
```bash
|
||||||
# Проверить логи воркеров
|
|
||||||
docker logs dataloader | grep worker
|
|
||||||
|
|
||||||
# Проверить конфигурацию
|
|
||||||
echo $WORKERS_JSON
|
|
||||||
|
|
||||||
# Проверить задачи в БД
|
# Проверить задачи в БД
|
||||||
SELECT job_id, queue, status, available_at, created_at
|
SELECT job_id, queue, status, available_at, created_at
|
||||||
|
|
@ -694,37 +808,12 @@ ORDER BY avg_duration_sec DESC;
|
||||||
- Оптимизировать запросы к БД (индексы, batching)
|
- Оптимизировать запросы к БД (индексы, batching)
|
||||||
- Масштабировать горизонтально (больше реплик)
|
- Масштабировать горизонтально (больше реплик)
|
||||||
|
|
||||||
### Утечка памяти
|
|
||||||
|
|
||||||
**Симптомы:** Постепенный рост потребления памяти, OOM kills.
|
|
||||||
|
|
||||||
**Диагностика:**
|
|
||||||
```bash
|
|
||||||
# Мониторинг памяти
|
|
||||||
kubectl top pods -l app=dataloader --containers
|
|
||||||
|
|
||||||
# Проверить логи перед падением
|
|
||||||
kubectl logs dataloader-xxx --previous
|
|
||||||
```
|
|
||||||
|
|
||||||
**Возможные причины:**
|
|
||||||
- Накопление объектов в памяти в пайплайне
|
|
||||||
- Незакрытые соединения/файлы
|
|
||||||
- Утечки в зависимостях
|
|
||||||
|
|
||||||
**Решение:**
|
|
||||||
- Использовать context managers (`async with`) для ресурсов
|
|
||||||
- Обрабатывать данные чанками, не загружать всё в память
|
|
||||||
- Периодически перезапускать воркеры (restart policy)
|
|
||||||
|
|
||||||
### Проблемы с LISTEN/NOTIFY
|
### Проблемы с LISTEN/NOTIFY
|
||||||
|
|
||||||
**Симптомы:** Воркеры не просыпаются сразу после постановки задачи.
|
**Симптомы:** Воркеры не просыпаются сразу после постановки задачи.
|
||||||
|
|
||||||
**Диагностика:**
|
**Диагностика:**
|
||||||
```bash
|
```bash
|
||||||
# Проверить логи listener
|
|
||||||
docker logs dataloader | grep "notify_listener\|LISTEN"
|
|
||||||
|
|
||||||
# Проверить триггеры в БД
|
# Проверить триггеры в БД
|
||||||
SELECT * FROM pg_trigger WHERE tgname LIKE 'dl_jobs%';
|
SELECT * FROM pg_trigger WHERE tgname LIKE 'dl_jobs%';
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue