Как решали технические долги на проекте?

Как решали технические долги на проекте

Технический долг — это один из самых важных вызовов в разработке, особенно в data-driven проектах. Это ситуация, когда быстрые решения накапливаются и позже требуют переделки. Рассмотрю систематический подход к управлению и погашению технического долга.

Что такое технический долг

Технический долг — это код, архитектура или процессы, которые:

• Работают, но неоптимальны
• Создают проблемы при масштабировании
• Требуют дополнительных усилий при поддержке
• Замедляют разработку новых функций
• Вызывают багиопреемник в production

Примеры в Data Engineering:

• Неоптимизированные SQL запросы
• Дублирование кода в скриптах ETL
• Отсутствие логирования и мониторинга
• Хардкодированные пути и параметры
• Отсутствие тестов
• Архаичные версии библиотек

Фаза 1: Идентификация технического долга

1.1 Аудит кодовой базы

# Пример скрипта для поиска проблемных участков
import os
import re

def find_hardcoded_values():
    """Найти хардкодированные пути и секреты"""
    patterns = [
        r'path\s*=\s*["\']/',  # Абсолютные пути
        r'password\s*=',  # Пароли
        r'api_key\s*=',  # API ключи
        r'TODO|FIXME|HACK',  # Комментарии разработчиков
    ]
    
    issues = []
    for root, dirs, files in os.walk('.'):
        for file in files:
            if file.endswith('.py'):
                with open(os.path.join(root, file)) as f:
                    for i, line in enumerate(f, 1):
                        for pattern in patterns:
                            if re.search(pattern, line, re.IGNORECASE):
                                issues.append(f"{file}:{i} - {line.strip()}")
    
    return issues

def check_test_coverage():
    """Проверить покрытие тестами"""
    # pytest --cov=src --cov-report=html
    pass

1.2 Метрики технического долга

Метрики для отслеживания:
├── Code Coverage: % покрытия тестами (целевой: >90%)
├── Cyclomatic Complexity: средняя сложность функций
├── Code Duplication: % повторяющегося кода
├── Documentation: % функций с docstrings
├── Security Issues: найденные уязвимости
├── Performance Issues: медленные операции
└── Technical Debt Ratio: % времени для погашения долга

Фаза 2: Приоритизация технического долга

2.1 Матрица приоритизации

Высокий приоритет (делать сразу):
- Проблемы с производительностью, влияющие на SLA
- Проблемы безопасности (утечки данных, credentials)
- Критические баги в production
- Блокирующие новую разработку

Средний приоритет (планировать на спринты):
- Дублирование кода, требующее рефакторинга
- Отсутствие логирования
- Неутилизируемые зависимости
- Низкое качество кода

Низкий приоритет (делать по мере возможности):
- Улучшение документации
- Оптимизация некритичного кода
- Обновление версий библиотек
- Переименование переменных

Фаза 3: Планирование погашения долга

3.1 Выделение времени

Правило 20%: на каждый спринт выделяйте 20% времени на погашение технического долга:

Sprinт: 2 недели (10 дней)

Время по категориям:
├── Новые функции: 7 дней (70%)
├── Баги: 1 день (10%)
└── Технический долг: 2 дня (20%)

3.2 Пример плана погашения

Месяц 1 (Приоритет 1 - Критические):
- Неделя 1-2: Удаление hardcoded secrets и использование environment variables
- Неделя 3-4: Добавление логирования в критичные функции ETL

Месяц 2 (Приоритет 2 - Средний):
- Неделя 1-2: Извлечение общего кода в отдельный модуль
- Неделя 3-4: Написание unit тестов для core функций

Месяц 3 (Приоритет 3 - Низкий):
- Неделя 1-2: Обновление dependencies
- Неделя 3-4: Рефакторинг и улучшение документации

Фаза 4: Реализация решений

4.1 Пример: Удаление дублирования кода

ДО (дублирование):

# etl_users.py
def load_users():
    df = spark.read.csv("s3://bucket/users.csv")
    df = df.fillna(0)
    df = df.drop_duplicates(subset=['id'])
    return df

# etl_products.py
def load_products():
    df = spark.read.csv("s3://bucket/products.csv")
    df = df.fillna(0)
    df = df.drop_duplicates(subset=['id'])
    return df

# etl_orders.py
def load_orders():
    df = spark.read.csv("s3://bucket/orders.csv")
    df = df.fillna(0)
    df = df.drop_duplicates(subset=['id'])
    return df

ПОСЛЕ (извлечение общей логики):

# data_utils.py
def load_and_clean_csv(path: str, id_column: str = 'id') -> DataFrame:
    """Load CSV and apply standard cleaning."""
    df = spark.read.csv(path, header=True)
    df = df.fillna(0)  # Или использовать default_value parameter
    df = df.drop_duplicates(subset=[id_column])
    return df

# etl_users.py
from data_utils import load_and_clean_csv

def load_users():
    return load_and_clean_csv("s3://bucket/users.csv")

# etl_products.py
def load_products():
    return load_and_clean_csv("s3://bucket/products.csv")

4.2 Пример: Добавление логирования

ДО (без логирования):

def extract_data(query: str):
    connection = get_db_connection()
    result = connection.execute(query)
    return result

ПОСЛЕ (с логированием и мониторингом):

import logging
import time
from functools import wraps
from prometheus_client import Counter, Histogram

logger = logging.getLogger(__name__)
execution_time = Histogram('extraction_duration_seconds', 'Time spent extracting')
extraction_counter = Counter('extractions_total', 'Total extractions')
error_counter = Counter('extraction_errors_total', 'Failed extractions')

def extract_data(query: str) -> pd.DataFrame:
    """Extract data from database with logging and monitoring."""
    start_time = time.time()
    logger.info(f"Starting data extraction with query: {query[:100]}...")
    
    try:
        connection = get_db_connection()
        logger.debug(f"Connected to database")
        
        result = connection.execute(query)
        row_count = len(result)
        
        logger.info(f"Successfully extracted {row_count} rows")
        extraction_counter.inc()
        
        return result
        
    except Exception as e:
        logger.error(f"Extraction failed: {str(e)}", exc_info=True)
        error_counter.inc()
        raise
        
    finally:
        duration = time.time() - start_time
        execution_time.observe(duration)
        logger.debug(f"Extraction completed in {duration:.2f} seconds")

4.3 Пример: Использование environment variables вместо hardcoding

ДО (хардкод):

DATABASE_URL = "postgresql://user:password@localhost:5432/mydb"
S3_BUCKET = "my-private-bucket"
API_KEY = "sk_live_abc123xyz"

ПОСЛЕ (environment variables):

import os
from dotenv import load_dotenv

load_dotenv()  # Загрузить из .env файла

DATABASE_URL = os.getenv(
    "DATABASE_URL",
    "postgresql://localhost:5432/mydb"  # Default для development
)
S3_BUCKET = os.getenv("S3_BUCKET")
API_KEY = os.getenv("API_KEY")

if not API_KEY:
    raise ValueError("API_KEY environment variable not set")

Фаза 5: Тестирование и валидация

5.1 Убедиться, что изменения не сломали функциональность

# Запустить все тесты
pytest -v

# Проверить покрытие
pytest --cov=src --cov-report=html

# Запустить linting
ruff check .
pylint src/

# Проверить типы
mypy src/

5.2 Регрессионное тестирование

def test_load_and_clean_csv_backward_compatibility():
    """Убедиться что новая функция работает как старая"""
    # Тестовые данные
    test_file = "test_data.csv"
    
    # Результаты обеих функций должны быть одинаковы
    result_old = old_load_users()  # Старая реализация (если есть)
    result_new = load_users()  # Новая реализация
    
    pd.testing.assert_frame_equal(result_old, result_new)

Фаза 6: Мониторинг и поддержка

6.1 Метрики успеха

# Отслеживание уменьшения технического долга
metrics = {
    "code_coverage": 92,  # Увеличилось с 70%
    "duplication_ratio": 5,  # Уменьшилось с 25%
    "security_issues": 0,  # Уменьшилось с 8
    "time_to_deploy": 15,  # Уменьшилось с 45 мин
    "production_incidents": 1,  # Уменьшилось с 5
}

Лучшие практики управления техдолгом

1. Регулярный код-ревью: выявляйте проблемы на ранних этапах

2. Автоматизация: используйте linting, testing, security scanning

# GitHub Actions example
name: Code Quality
on: [push, pull_request]
jobs:
  quality:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
      - name: Run tests
        run: pytest --cov
      - name: Run linting
        run: ruff check .
      - name: Check security
        run: bandit -r src/

1. Документация: ведите реестр технического долга

# Technical Debt Register

## High Priority
- [ ] Remove hardcoded secrets (Estimated: 2 days, Status: In Progress)
- [ ] Add monitoring to ETL pipeline (Estimated: 3 days, Status: Open)

## Medium Priority
- [ ] Refactor duplicate code in transformations (Estimated: 4 days, Status: Backlog)
- [ ] Improve test coverage to 90% (Estimated: 5 days, Status: Open)

## Low Priority
- [ ] Update dependencies (Estimated: 1 day, Status: Open)

1. Культура: подчеркивайте важность качества кода в команде

2. Баланс: не жертвуйте сроками ради идеального кода, но не игнорируйте долг

Метрики для отслеживания

metrics_to_track = {
    "technical_debt_ratio": "% времени на погашение долга",
    "code_coverage": "% кода покрытого тестами",
    "cyclomatic_complexity": "средняя сложность функций",
    "documentation_ratio": "% функций с документацией",
    "security_issues": "количество найденных уязвимостей",
    "mean_time_to_deploy": "время развертывания",
    "production_incidents": "количество инцидентов",
}

Управление техническим долгом — это не разовая задача, а постоянный процесс, который требует дисциплины и системного подхода. В Data Engineering проектах это критично для надежности и масштабируемости систем.