Перейти к содержанию

Python для дата-инженера — основы

Python — основной язык автоматизации в дата-инженерии. Ты пишешь на нём ETL-скрипты, работаешь с API, управляешь пайплайнами в Airflow и трансформируешь данные в pandas. В этой статье — минимум, который нужен DE-инженеру, чтобы уверенно читать и писать код.

Это не курс Python с нуля

Статья предполагает, что ты знаешь базовый синтаксис: переменные, циклы, функции. Здесь — фокус на инструментах и практиках, которые нужны именно дата-инженеру.

Зачем Python дата-инженеру

У Data Scientist, бэкенд-разработчика и дата-инженера — разные задачи на Python:

Роль Типичные задачи на Python
Data Scientist Модели, эксперименты, Jupyter
Backend HTTP-серверы, ORM, бизнес-логика
Data Engineer ETL-скрипты, работа с API и БД, Airflow DAG, трансформации

Дата-инженер чаще всего:

  • забирает данные из REST API и складывает в хранилище;
  • пишет DAG для Airflow (Python — основной язык DAG);
  • читает/записывает файлы: CSV, JSON, Parquet;
  • делает ad-hoc трансформации в pandas;
  • автоматизирует инфраструктуру: миграции, бэкапы, мониторинг.

Виртуальные окружения

Каждый проект должен жить в изолированном окружении. Это защищает от конфликта зависимостей между проектами.

venv — встроенный модуль

Bash
# Создать окружение
python3 -m venv .venv

# Активировать
source .venv/bin/activate

# Установить зависимости
pip install pandas requests sqlalchemy

# Зафиксировать
pip freeze > requirements.txt

# Воспроизвести на другой машине
pip install -r requirements.txt

uv — быстрая замена pip

uv — менеджер пакетов от Astral (авторы ruff). Работает в 10–100 раз быстрее pip.

Bash
# Установить uv
curl -LsSf https://astral.sh/uv/install.sh | sh

# Создать проект
uv init my-etl-project
cd my-etl-project

# Добавить зависимости
uv add pandas requests sqlalchemy

# Запустить скрипт в окружении проекта
uv run python main.py

uv vs pip

uv автоматически создаёт .venv, управляет pyproject.toml и lock-файлом. Не нужно отдельно создавать окружение и фиксировать зависимости — всё из коробки.

poetry — для сложных проектов

Bash
poetry init
poetry add pandas requests
poetry run python main.py

Сравнение

Критерий venv + pip uv poetry
Скорость установки Базовая Очень быстрая Средняя
Lock-файл Нет (pip freeze) Да (uv.lock) Да (poetry.lock)
Управление Python Нет Да (uv python install) Нет
Кривая входа Минимальная Низкая Средняя
Рекомендация Обучение, скрипты Новые проекты Библиотеки с публикацией

Не ставь пакеты глобально

pip install pandas без активного окружения ставит пакет в системный Python. Это ломает зависимости между проектами и может сломать системные утилиты.

Работа с файлами

CSV

Python
import csv
from pathlib import Path

# Чтение
with Path("users.csv").open(encoding="utf-8") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(row["name"], row["email"])

# Запись
rows = [
    {"name": "Alice", "email": "alice@example.com"},
    {"name": "Bob", "email": "bob@example.com"},
]
with Path("output.csv").open("w", encoding="utf-8", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=["name", "email"])
    writer.writeheader()
    writer.writerows(rows)

Когда CSV не хватает

CSV не хранит типы данных и медленно читается на больших объёмах. Для аналитических задач используй Parquet.

JSON

Python
import json
from pathlib import Path

# Чтение
data = json.loads(Path("config.json").read_text(encoding="utf-8"))

# Запись
Path("output.json").write_text(
    json.dumps(data, ensure_ascii=False, indent=2),
    encoding="utf-8",
)

Типичный паттерн DE — разобрать JSON-ответ API:

Python
import json

raw = '{"users": [{"id": 1, "name": "Alice"}, {"id": 2, "name": "Bob"}]}'
payload = json.loads(raw)

for user in payload["users"]:
    print(f"{user['id']}: {user['name']}")
Text Only
1: Alice
2: Bob

Parquet

Parquet — колоночный формат. Занимает в 5–10 раз меньше места, чем CSV, и читается значительно быстрее.

Python
import pyarrow as pa
import pyarrow.parquet as pq

# Создать таблицу
table = pa.table({
    "user_id": [1, 2, 3],
    "name": ["Alice", "Bob", "Carol"],
    "score": [95.5, 87.3, 91.0],
})

# Записать в Parquet
pq.write_table(table, "users.parquet")

# Прочитать обратно
table = pq.read_table("users.parquet")
print(table.to_pandas())
Text Only
   user_id   name  score
0        1  Alice   95.5
1        2    Bob   87.3
2        3  Carol   91.0

pyarrow vs fastparquet

pyarrow — стандарт де-факто. fastparquet — легче, но менее совместим. Для DE-задач бери pyarrow.

Библиотеки DE-стека

pandas — трансформации данных

Python
import pandas as pd

df = pd.read_csv("sales.csv")
result = (
    df.groupby("region")["revenue"]
    .sum()
    .reset_index()
    .sort_values("revenue", ascending=False)
)
print(result.head())

Подробнее — в статье pandas: трансформации данных.

requests — работа с API

Python
import requests

response = requests.get(
    "https://dummyjson.com/products",
    params={"limit": 5},
    timeout=10,
)
response.raise_for_status()
products = response.json()["products"]

for p in products:
    print(f"{p['title']}: ${p['price']}")

Подробнее — в статье Работа с API.

SQLAlchemy — работа с базами данных

Python
from sqlalchemy import create_engine, text

engine = create_engine("postgresql://user:pass@localhost:5432/mydb")

with engine.connect() as conn:
    result = conn.execute(text("SELECT id, name FROM users LIMIT 5"))
    for row in result:
        print(row.id, row.name)

SQLAlchemy vs psycopg2

psycopg2 — низкоуровневый драйвер PostgreSQL. SQLAlchemy — ORM и toolkit, который работает поверх драйвера. Для DE-задач SQLAlchemy удобнее: pd.read_sql() принимает engine напрямую.

boto3 — AWS (S3, Redshift, Glue)

Python
import boto3

s3 = boto3.client("s3")

# Загрузить файл в S3
s3.upload_file("data.parquet", "my-bucket", "raw/data.parquet")

# Скачать файл из S3
s3.download_file("my-bucket", "raw/data.parquet", "local_data.parquet")

Сводная таблица

Библиотека Задача Установка
pandas DataFrame, трансформации, read_csv/read_sql pip install pandas
requests HTTP-запросы к API pip install requests
SQLAlchemy Подключение к БД, ORM pip install sqlalchemy
psycopg2 Драйвер PostgreSQL pip install psycopg2-binary
pyarrow Parquet, Arrow-таблицы pip install pyarrow
boto3 AWS S3, Glue, Redshift pip install boto3
httpx Async HTTP (замена requests) pip install httpx

Типизация и линтеры

Type hints

Python — динамический язык, но type hints помогают ловить ошибки до запуска:

Python
def extract_users(api_url: str, limit: int = 100) -> list[dict]:
    """Забирает пользователей из API."""
    import requests

    response = requests.get(api_url, params={"limit": limit}, timeout=10)
    response.raise_for_status()
    return response.json()["users"]

Зачем типы в ETL

В ETL-скриптах данные проходят через много функций. Type hints — документация прямо в коде: сразу видно, что функция принимает и что возвращает.

ruff — линтер и форматтер

Bash
# Установить
pip install ruff

# Проверить код
ruff check .

# Исправить автоматически
ruff check --fix .

# Отформатировать
ruff format .

ruff заменяет flake8, isort, black и ещё десяток инструментов. Работает в 10–100 раз быстрее.

mypy — статическая проверка типов

Bash
pip install mypy

# Проверить
mypy my_etl_script.py

Минимальная конфигурация для DE-проекта

pyproject.toml
[tool.ruff]
line-length = 120
target-version = "py311"

[tool.ruff.lint]
select = ["E", "F", "I", "UP"]  # ошибки, import sort, upgrade

[tool.mypy]
python_version = "3.11"
warn_return_any = true
disallow_untyped_defs = true

Что запомнить

Тема Ключевая мысль
Окружения Всегда используй виртуальное окружение. Для новых проектов — uv
Файлы CSV для обмена, JSON для API, Parquet для аналитики
Библиотеки pandas + requests + SQLAlchemy — базовый стек DE
Типизация Type hints + ruff — минимум для читаемого кода

Проверь себя

Источники