Какие знаешь базы данных?

Базы данных для Data Scientists

В работе Data Scientist используются разные типы БД в зависимости от задач: аналитика, ML pipelines, production сервисы.

1. Relational Databases (RDBMS)

PostgreSQL — мой основной выбор

-- Создание таблицы
CREATE TABLE users (
    user_id SERIAL PRIMARY KEY,
    name VARCHAR(255),
    email VARCHAR(255) UNIQUE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    is_active BOOLEAN DEFAULT TRUE
);

-- Индексы для быстрого поиска
CREATE INDEX idx_users_email ON users(email);
CREATE INDEX idx_users_created ON users(created_at);

-- Сложный запрос с joins и aggregation
SELECT 
    u.user_id,
    u.name,
    COUNT(o.order_id) as total_orders,
    SUM(o.amount) as total_spent,
    AVG(o.amount) as avg_order_value
FROM users u
LEFT JOIN orders o ON u.user_id = o.user_id
WHERE u.created_at >= NOW() - INTERVAL '1 year'
GROUP BY u.user_id, u.name
HAVING COUNT(o.order_id) > 5
ORDER BY total_spent DESC;

Преимущества:

• ACID гарантии
• Мощный SQL
• Быстро для OLTP (Online Transaction Processing)
• Открытый исходный код

Недостатки:

• Не масштабируется горизонтально
• Медленнее на очень больших объемах данных

MySQL

Популярен в веб-приложениях, но медленнее PostgreSQL для аналитики.

-- Базовый запрос
SELECT * FROM products WHERE price > 100 LIMIT 10;

Используется: Web apps, CMS, e-commerce

2. Data Warehouses (OLAP)

Оптимизированы для аналитики больших объемов данных.

BigQuery (Google Cloud)

from google.cloud import bigquery

client = bigquery.Client()

# SQL запрос
query = """
    SELECT 
        DATE_TRUNC(DATE(created_at), WEEK) as week,
        COUNT(*) as orders,
        SUM(amount) as revenue,
        AVG(amount) as avg_order
    FROM `project.dataset.orders`
    WHERE created_at >= '2024-01-01'
    GROUP BY week
    ORDER BY week DESC
"""

# Выполнение запроса
job = client.query(query)
results = job.result().to_dataframe()  # конвертирует в pandas

Особенности:

• Serverless (no infrastructure management)
• Масштабируется автоматически
• Pay-per-query
• Поддерживает стандартный SQL
• Интеграция с ML (BigQuery ML)

Цена: $6.25 за TB сканированных данных

Snowflake

-- Snowflake использует свой SQL диалект
SELECT 
    DATE_TRUNC('WEEK', created_at) as week,
    COUNT(*) as order_count,
    SUM(amount) as revenue
FROM ORDERS
WHERE YEAR(created_at) = 2024
GROUP BY DATE_TRUNC('WEEK', created_at)
ORDER BY week DESC;

Особенности:

• Separates compute и storage
• Легко scale-up и scale-down
• Zero copy cloning
• Time travel (посмотреть историю)

Amazon Redshift

Аналог Snowflake от AWS. Columnar database.

import psycopg2

# Redshift совместим с PostgreSQL
conn = psycopg2.connect(
    host="redshift-cluster-1.xxxxx.us-east-1.redshift.amazonaws.com",
    database="analytics",
    user="admin",
    password="password"
)

cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM fact_orders;")
print(cursor.fetchone())

3. NoSQL Databases

MongoDB (Document database)

from pymongo import MongoClient

client = MongoClient('mongodb://localhost:27017/')
db = client['analytics']
collection = db['user_events']

# Insert
collection.insert_one({
    'user_id': 123,
    'event_type': 'click',
    'timestamp': datetime.now(),
    'metadata': {
        'page': '/products',
        'session_id': 'abc123'
    }
})

# Query
events = collection.find({
    'user_id': 123,
    'timestamp': {'$gte': datetime(2024, 1, 1)}
}).limit(10)

for event in events:
    print(event)

# Aggregation pipeline
results = collection.aggregate([
    {'$match': {'event_type': 'click'}},
    {'$group': {'_id': '$user_id', 'count': {'$sum': 1}}},
    {'$sort': {'count': -1}},
    {'$limit': 10}
])

Используется: Неструктурированные данные, быстрое прототипирование

Redis (In-memory store)

import redis

r = redis.Redis(host='localhost', port=6379, db=0)

# Caching
r.set('user:123:profile', json.dumps({'name': 'John'}), ex=3600)
profile = json.loads(r.get('user:123:profile'))

# Leaderboard
r.zadd('high_scores', {'player1': 1000, 'player2': 950})
top_10 = r.zrevrange('high_scores', 0, 9, withscores=True)

# Real-time counters
r.incr('page_views')  # counter

Используется: Caching, real-time analytics, leaderboards

Elasticsearch

from elasticsearch import Elasticsearch

es = Elasticsearch(['localhost:9200'])

# Index документ
es.index(index='logs', id=1, body={
    'timestamp': '2024-01-01T12:00:00',
    'message': 'User login successful',
    'user_id': 123,
    'level': 'info'
})

# Search
results = es.search(index='logs', body={
    'query': {
        'bool': {
            'must': [
                {'match': {'message': 'login'}},
                {'range': {'timestamp': {'gte': '2024-01-01'}}}
            ]
        }
    },
    'aggs': {
        'by_level': {'terms': {'field': 'level'}}
    }
})

Используется: Full-text search, log analysis, real-time search

4. Time Series Databases

Для временных рядов (metrics, logs, sensor data).

InfluxDB

from influxdb import InfluxDBClient

client = InfluxDBClient(host='localhost', port=8086, database='metrics')

# Write metrics
json_body = [
    {
        "measurement": "cpu_usage",
        "tags": {"host": "server01"},
        "time": "2024-01-01T12:00:00Z",
        "fields": {"value": 75.5}
    }
]
client.write_points(json_body)

# Query
result = client.query('SELECT mean(value) FROM cpu_usage WHERE time > now() - 1h')

Используется: System monitoring, IoT sensors, financial data

Prometheus + TimescaleDB

# TimescaleDB = PostgreSQL + time series extensions
# Подходит когда нужна мощь SQL + оптимизация для TS

CREATE TABLE metrics (
    time TIMESTAMPTZ NOT NULL,
    metric_name TEXT NOT NULL,
    host_id INT,
    value FLOAT
);

SELECT hypertable_name FROM timescaledb_information.hypertables;

5. Graph Databases

Для связанных данных.

Neo4j

from neo4j import GraphDatabase

driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))

def create_user_friendship(tx, user1_id, user2_id):
    tx.run(
        "MERGE (u1:User {id: $id1}) "
        "MERGE (u2:User {id: $id2}) "
        "MERGE (u1)-[:FRIEND]->(u2)",
        id1=user1_id,
        id2=user2_id
    )

# Recommendations: найди друзей моих друзей
with driver.session() as session:
    result = session.run(
        "MATCH (user:User {id: $id})-[:FRIEND]->(friend)-[:FRIEND]->(friend_of_friend) "
        "RETURN friend_of_friend",
        id=123
    )
    for record in result:
        print(record)

Используется: Social networks, recommendations, knowledge graphs

6. Data Lake / Object Storage

Для хранения raw данных.

AWS S3

import boto3

s3_client = boto3.client('s3')

# Upload file
s3_client.upload_file(
    'local_file.csv',
    'my-bucket',
    'data/local_file.csv'
)

# Read with pandas
import pandas as pd
df = pd.read_csv('s3://my-bucket/data/local_file.csv')

# Query S3 with SELECT
import json

response = s3_client.select_object_content(
    Bucket='my-bucket',
    Key='data/file.csv',
    Expression='SELECT * FROM s3object WHERE age > 18',
    ExpressionType='SQL',
    InputSerialization={'CSV': {"FileHeaderInfo": "Use"}},
    OutputSerialization={'CSV': {}}
)

7. Distributed Processing Engines

Для big data обработки.

Apache Spark + Hadoop

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("DataAnalysis").getOrCreate()

# Read data
df = spark.read.csv('s3://bucket/data.csv', header=True)

# Transformations
result = (df
    .filter(df.age > 18)
    .groupBy('country')
    .agg({'salary': 'avg'})
    .sort('avg(salary)', ascending=False)
)

result.show(10)
result.write.mode('overwrite').parquet('s3://bucket/results/')

Используется: Batch processing, data preprocessing, ML pipeline

Apache Kafka (Streaming)

from kafka import KafkaProducer, KafkaConsumer
import json

# Producer
producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
producer.send('events', json.dumps({'user_id': 123, 'action': 'login'}).encode())

# Consumer
consumer = KafkaConsumer('events', bootstrap_servers=['localhost:9092'])
for message in consumer:
    event = json.loads(message.value)
    process_event(event)  # Real-time processing

Выбор БД в зависимости от задачи

Мой типичный стек

Data Collection: PostgreSQL / Kafka
         ↓
Data Lake: AWS S3
         ↓
Data Warehouse: BigQuery или Snowflake
         ↓
Analytics: pandas + Spark для processing
         ↓
ML: PostgreSQL для features, Redis для cache
         ↓
Production: FastAPI + PostgreSQL + Redis

Best Practices

1. Нормализация: Organize данные правильно
2. Индексы: Создавай на часто используемых полях
3. Партиционирование: Для больших таблиц
4. Backup & Recovery: Автоматические бэкапы
5. Мониторинг: Query performance, disk space
6. Security: Encryption, access control
7. Retention policies: Удаляй старые данные