Глоссарий — System Design (DE)

# Типичный System Design процесс
1. Сбор требований (functional + non-functional)
2. Оценка нагрузки (QPS, data volume, latency)
3. High-level design (компоненты + потоки данных)
4. Детализация (storage, processing, serving)
5. Анализ bottleneck-ов и tradeoff-ов

# CP-системы (Consistency + Partition tolerance)
# HBase, MongoDB (с majority write concern)
# → Откажут в обслуживании при network partition

# AP-системы (Availability + Partition tolerance)
# Cassandra, DynamoDB
# → Вернут устаревшие данные при partition

# Пример: Kafka — AP для producers, CP для consumers с read_committed

# Пример tradeoff: выбор формата хранения
#
# Parquet:
#   + Columnar → быстрые аналитические запросы
#   + Compression → меньше storage cost
#   - Медленная запись (row → column конвертация)
#
# Avro:
#   + Быстрая запись (row-based)
#   + Schema evolution
#   - Медленнее для аналитики
#
# Решение: Avro для ingestion, Parquet для serving

# Fallacies of Distributed Computing
# 1. Сеть надёжна → нет, нужен retry + timeout
# 2. Латентность нулевая → нет, нужен batching
# 3. Пропускная способность бесконечна → нет, нужен backpressure
# 4. Сеть безопасна → нет, нужен TLS + authN

# Пример: Kafka — распределённый лог
# Partitions → параллелизм
# Replication factor 3 → отказоустойчивость
# ISR (In-Sync Replicas) → consistency guarantee

# SLI для data pipeline:
#   freshness: время от события до доступности в DWH
#   completeness: % записей без потерь
#   accuracy: % корректных трансформаций

# SLO:
#   freshness < 15 min (p99)
#   completeness > 99.9%
#   accuracy > 99.99%

# SLA:
#   Нарушение freshness > 1 час → алерт oncall
#   Нарушение completeness < 99% → incident P1

# ETL: трансформация до загрузки
raw_data = extract(source_db)
cleaned = transform(raw_data)  # в памяти ETL-сервера
load(cleaned, target_warehouse)

# ELT: загрузка → трансформация в warehouse
raw_data = extract(source_db)
load(raw_data, staging_table)  # as-is в warehouse
# dbt model трансформирует в warehouse
# SELECT cleaned_col FROM staging_table WHERE ...

# Lambda Architecture
#
# ┌─────────┐    ┌──────────────┐    ┌──────────────┐
# │  Source  │───→│  Batch Layer │───→│              │
# │  Events  │    │  (Spark)     │    │  Serving     │
# │          │───→│  Speed Layer │───→│  Layer       │
# └─────────┘    │  (Flink)     │    │  (Druid)     │
#                └──────────────┘    └──────────────┘
#
# Batch: ежечасный пересчёт агрегатов
# Speed: real-time дельта с последнего batch

# Kappa Architecture — единый поток
#
# ┌─────────┐    ┌──────────────┐    ┌──────────────┐
# │  Source  │───→│  Stream      │───→│  Serving     │
# │  Events  │    │  Processor   │    │  Layer       │
# └─────────┘    │  (Flink)     │    │  (ClickHouse)│
#                └──────────────┘    └──────────────┘
#
# Переобработка: replay topic с offset 0
# Нет batch layer → единая кодовая база

# Bronze → Silver → Gold
#
# Bronze: raw JSON events из Kafka
# spark.readStream.format('kafka')...writeStream.format('delta')
#
# Silver: parsed, deduplicated, typed
# spark.read.format('delta').load('/bronze/events')
#   .dropDuplicates(['event_id'])
#   .withColumn('ts', to_timestamp('raw_ts'))
#   .write.format('delta').save('/silver/events')
#
# Gold: business aggregates
# spark.read.format('delta').load('/silver/events')
#   .groupBy('date', 'product_id').agg(sum('revenue'))
#   .write.format('delta').save('/gold/daily_revenue')

# Reverse ETL: DWH → Salesforce
#
# 1. Модель в dbt (Gold layer):
#    SELECT user_id, ltv_score, churn_risk
#    FROM gold.user_segments
#
# 2. Sync job (Census / Hightouch):
#    Source: gold.user_segments
#    Destination: Salesforce.Contact
#    Mapping: ltv_score → Custom_LTV__c
#    Schedule: every 4 hours
#    Mode: upsert on user_id

# Spark batch job — ежедневная агрегация
df = spark.read.parquet(f'/data/events/dt={yesterday}')

aggregated = df \
    .filter(col('event_type') == 'purchase') \
    .groupBy('product_id') \
    .agg(
        count('*').alias('order_count'),
        sum('amount').alias('total_revenue')
    )

aggregated.write \
    .mode('overwrite') \
    .partitionBy('product_id') \
    .parquet(f'/output/daily_revenue/dt={yesterday}')

# Spark Structured Streaming — micro-batch
df = spark.readStream \
    .format('kafka') \
    .option('subscribe', 'events') \
    .load()

result = df \
    .selectExpr("CAST(value AS STRING)") \
    .select(from_json('value', schema).alias('data')) \
    .select('data.*')

result.writeStream \
    .trigger(processingTime='30 seconds')  # micro-batch каждые 30 сек\
    .format('delta') \
    .start('/silver/events')

# Flink — stream processing с event time
env = StreamExecutionEnvironment.get_execution_environment()

events = env \
    .add_source(FlinkKafkaConsumer('events', schema, props)) \
    .assign_timestamps_and_watermarks(
        WatermarkStrategy
            .for_bounded_out_of_orderness(Duration.of_seconds(10))
            .with_timestamp_assigner(EventTimestampAssigner())
    )

events \
    .key_by(lambda e: e.user_id) \
    .window(TumblingEventTimeWindows.of(Time.minutes(5))) \
    .aggregate(CountAggregateFunction()) \
    .add_sink(sink)

# Event time vs Processing time
#
# Event создан: 10:00:05 (event time)
# Доставлен в Kafka: 10:00:07
# Обработан Flink: 10:00:12 (processing time)
# Skew: 7 секунд
#
# Окно 10:00 - 10:05 по event time:
#   → Событие 10:00:05 попадёт в это окно
#
# Окно 10:00 - 10:05 по processing time:
#   → Событие НЕ попадёт (processing time = 10:00:12)
#
# Вывод: event time = корректные бизнес-метрики

# Watermark с допуском 10 секунд
#
# Поток событий:
# event(time=10:00:01) → watermark = 09:59:51
# event(time=10:00:08) → watermark = 09:59:58
# event(time=10:00:15) → watermark = 10:00:05
#
# Окно [10:00:00, 10:00:05) закрывается
# когда watermark >= 10:00:05
#
# Late event (time=10:00:03, arrived at 10:00:20)
# watermark уже 10:00:10 → событие отброшено

# Kafka exactly-once: transactional producer + consumer
producer = KafkaProducer(
    transactional_id='etl-job-01',
    enable_idempotence=True
)
producer.init_transactions()

try:
    producer.begin_transaction()
    producer.send('output-topic', value=result)
    producer.send_offsets_to_transaction(
        {tp: offset}, consumer_group_id
    )
    producer.commit_transaction()
except:
    producer.abort_transaction()

-- Типичный аналитический запрос в DWH
SELECT
    d.month_name,
    p.category,
    SUM(f.revenue) as total_revenue,
    COUNT(DISTINCT f.customer_id) as unique_customers
FROM fact_sales f
JOIN dim_date d ON f.date_key = d.date_key
JOIN dim_product p ON f.product_key = p.product_key
WHERE d.year = 2024
GROUP BY d.month_name, p.category
ORDER BY total_revenue DESC;

# Data Lake на S3 — структура директорий
# s3://company-datalake/
# ├── raw/                    # Bronze: as-is
# │   ├── salesforce/
# │   ├── clickstream/
# │   └── iot-sensors/
# ├── processed/              # Silver: cleaned
# │   ├── users/
# │   └── events/
# └── curated/                # Gold: business-ready
#     ├── daily_revenue/
#     └── user_segments/

# Delta Lake — Lakehouse операции
# ACID транзакции на Parquet
df.write.format('delta') \
    .mode('overwrite') \
    .save('/lakehouse/silver/orders')

# Time travel — чтение версии 3
spark.read.format('delta') \
    .option('versionAsOf', 3) \
    .load('/lakehouse/silver/orders')

# MERGE (upsert) — CDC применение
DeltaTable.forPath(spark, '/lakehouse/silver/orders') \
    .merge(updates, 'target.id = source.id') \
    .whenMatchedUpdateAll() \
    .whenNotMatchedInsertAll() \
    .execute()

# Row-oriented vs Columnar — чтение SELECT name, age
#
# Row: читаем ВСЕ колонки каждой строки
# [id=1, name='Alice', age=30, city='NY', ...]
# [id=2, name='Bob', age=25, city='LA', ...]
# → Читаем 100% данных, используем 20%
#
# Columnar: читаем только нужные столбцы
# name: ['Alice', 'Bob', ...]
# age:  [30, 25, ...]
# → Читаем 20% данных, используем 100%
#
# Parquet файл: ~10x compression vs CSV
# 1 TB CSV ≈ 100 GB Parquet

# Hive-style partitioning в Spark
df.write \
    .partitionBy('year', 'month', 'day') \
    .format('parquet') \
    .save('/data/events')

# Структура на диске:
# /data/events/year=2024/month=01/day=15/part-00000.parquet
# /data/events/year=2024/month=01/day=16/part-00000.parquet

# Запрос с partition pruning:
# SELECT * FROM events WHERE year=2024 AND month=01
# → Читает только 31 директорию вместо 365

-- Dimensional model: продажи
-- Fact table: зернистость = одна транзакция
CREATE TABLE fact_sales (
    sale_id BIGINT,
    date_key INT,       -- FK → dim_date
    product_key INT,    -- FK → dim_product
    customer_key INT,   -- FK → dim_customer
    store_key INT,      -- FK → dim_store
    quantity INT,       -- measure
    amount DECIMAL,     -- measure
    discount DECIMAL    -- measure
);

-- Star Schema: fact в центре, dimensions вокруг
--
--              dim_date
--                 |
-- dim_product — fact_sales — dim_customer
--                 |
--              dim_store
--
SELECT
    d.quarter,
    p.category,
    SUM(f.amount) as revenue
FROM fact_sales f
JOIN dim_date d ON f.date_key = d.date_key        -- 1 hop
JOIN dim_product p ON f.product_key = p.product_key -- 1 hop
GROUP BY d.quarter, p.category;

-- Snowflake: dimension разбита на под-таблицы
--
-- dim_product → dim_subcategory → dim_category
--
-- Star: dim_product.category = 'Electronics'
-- Snowflake: JOIN dim_subcategory → JOIN dim_category

SELECT c.category_name, SUM(f.amount)
FROM fact_sales f
JOIN dim_product p ON f.product_key = p.product_key
JOIN dim_subcategory sc ON p.subcategory_key = sc.subcategory_key
JOIN dim_category c ON sc.category_key = c.category_key
GROUP BY c.category_name;

-- SCD Type 2: история изменений адреса клиента
CREATE TABLE dim_customer (
    customer_key INT,          -- surrogate key
    customer_id VARCHAR(20),   -- natural key
    name VARCHAR(100),
    city VARCHAR(50),
    valid_from DATE,
    valid_to DATE,             -- '9999-12-31' для текущей
    is_current BOOLEAN
);

-- Клиент переехал из Москвы в Питер:
-- Row 1: city='Москва', valid_from='2023-01-01', valid_to='2024-06-15', is_current=false
-- Row 2: city='Питер',  valid_from='2024-06-15', valid_to='9999-12-31', is_current=true

-- Data Vault: Hub + Satellite + Link

-- Hub: бизнес-ключ
CREATE TABLE hub_customer (
    hub_customer_hk BINARY(32),  -- hash key
    customer_bk VARCHAR(20),     -- business key
    load_dts TIMESTAMP,
    record_source VARCHAR(50)
);

-- Satellite: атрибуты с историей
CREATE TABLE sat_customer_details (
    hub_customer_hk BINARY(32),
    load_dts TIMESTAMP,
    name VARCHAR(100),
    city VARCHAR(50),
    hash_diff BINARY(32)  -- detect changes
);

-- Link: связь customer ↔ order
CREATE TABLE link_customer_order (
    link_hk BINARY(32),
    hub_customer_hk BINARY(32),
    hub_order_hk BINARY(32),
    load_dts TIMESTAMP
);

# Debezium CDC — Postgres → Kafka
# connector config (JSON):
{
  "name": "pg-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.dbname": "orders_db",
    "table.include.list": "public.orders",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_slot",
    "topic.prefix": "cdc"
  }
}
# Результат: топик cdc.public.orders с before/after payload

# PostgreSQL WAL → Debezium → Kafka
#
# 1. PostgreSQL пишет в WAL (Write-Ahead Log)
# 2. Debezium читает WAL через logical replication slot
# 3. Каждое изменение → Kafka message:
#    {
#      "op": "u",  // c=create, u=update, d=delete
#      "before": {"id": 1, "status": "pending"},
#      "after":  {"id": 1, "status": "shipped"},
#      "source": {"lsn": 12345, "txId": 678}
#    }
# 4. Consumer применяет изменения к target

# Great Expectations — data quality checks
import great_expectations as gx

context = gx.get_context()
batch = context.get_batch('silver_orders')

# Completeness
batch.expect_column_values_to_not_be_null('order_id')

# Uniqueness
batch.expect_column_values_to_be_unique('order_id')

# Validity
batch.expect_column_values_to_be_between(
    'amount', min_value=0, max_value=1_000_000
)

# Freshness
batch.expect_column_max_to_be_between(
    'updated_at',
    min_value=datetime.now() - timedelta(hours=1)
)

# 5 столпов Data Observability:
#
# 1. Freshness: таблица обновлялась 3ч назад → ALERT
#    monitor: MAX(updated_at) > NOW() - INTERVAL 1 HOUR
#
# 2. Volume: сегодня 50% меньше строк → ALERT
#    monitor: COUNT(*) within 2σ of 7-day average
#
# 3. Schema: колонка удалена → ALERT
#    monitor: schema diff vs previous version
#
# 4. Distribution: NULL% скакнул с 1% до 40% → ALERT
#    monitor: column_null_pct < threshold
#
# 5. Lineage: upstream table broken → trace impact
#    monitor: dependency graph traversal

# data_contract.yaml
version: 1.0
owner: team-payments

schema:
  type: object
  properties:
    order_id:
      type: string
      format: uuid
      description: Unique order identifier
    amount:
      type: number
      minimum: 0
    status:
      type: string
      enum: [pending, paid, refunded]

sla:
  freshness: 15 minutes
  completeness: 99.9%

quality:
  - column: order_id
    check: unique
  - column: amount
    check: not_null

# Airflow DAG — ежедневный ETL
from airflow import DAG
from airflow.operators.python import PythonOperator

with DAG('daily_etl', schedule='@daily') as dag:
    extract = PythonOperator(
        task_id='extract_from_postgres',
        python_callable=extract_data
    )
    transform = PythonOperator(
        task_id='transform_with_spark',
        python_callable=run_spark_job
    )
    load = PythonOperator(
        task_id='load_to_warehouse',
        python_callable=load_data
    )
    quality = PythonOperator(
        task_id='run_quality_checks',
        python_callable=check_quality
    )
    extract >> transform >> load >> quality

# Airflow vs Dagster — подходы к orchestration
#
# Airflow: task-centric
#   DAG = набор задач + зависимости
#   Scheduler проверяет расписание
#   Task Instance выполняется Worker-ом
#
# Dagster: asset-centric
#   @asset
#   def daily_revenue(orders, products):
#       return orders.join(products).groupBy('date').sum('amount')
#   
#   # Dagster автоматически строит DAG из зависимостей
#   # между assets (не задачами)

# Data Lineage граф
#
# postgres.orders ─┐
#                   ├─→ bronze.orders ─→ silver.orders ─┐
# kafka.payments ──┘                                     │
#                                                         ├─→ gold.daily_revenue
# postgres.products ─→ bronze.products ─→ silver.products┘
#
# Impact analysis: изменение postgres.orders
# → затронуты: bronze.orders, silver.orders, gold.daily_revenue
# → 3 downstream dashboards, 2 ML models

# DataHub — каталогизация таблицы
# Автоматически собирает метаданные:
#
# Table: gold.daily_revenue
# Owner: team-analytics
# Description: Ежедневная выручка по продуктам
# Schema:
#   - date: DATE (partition key)
#   - product_id: STRING
#   - revenue: DECIMAL
#   - order_count: INT
# Lineage: silver.orders + silver.products → gold.daily_revenue
# Quality: 99.8% freshness SLO met (30-day)
# Usage: 47 queries/day, 12 unique users
# Tags: #revenue #finance #daily

# Cost optimization checklist
#
# Storage:
#   ✓ Parquet вместо CSV (10x compression)
#   ✓ Storage tiering: S3 Standard → Glacier после 90 дней
#   ✓ Partition pruning: queries read 5% of data
#
# Compute:
#   ✓ Spot instances для batch (70% savings)
#   ✓ Auto-scaling: 0 instances ночью
#   ✓ Right-sizing: m5.2xlarge → m5.xlarge
#
# Query:
#   ✓ Materialized views для hot queries
#   ✓ Predicate pushdown (filter at source)
#   ✓ CLUSTER BY для DWH tables

# Feast Feature Store
from feast import FeatureStore

store = FeatureStore(repo_path='feature_repo/')

# Offline: получение фичей для training
training_df = store.get_historical_features(
    entity_df=entities,  # user_id + event_timestamp
    features=[
        'user_features:total_purchases_30d',
        'user_features:avg_order_value',
        'user_features:days_since_last_order'
    ]
).to_df()

# Online: получение фичей для serving (p99 < 10ms)
online_features = store.get_online_features(
    features=['user_features:total_purchases_30d'],
    entity_rows=[{'user_id': 'u123'}]
).to_dict()

# Feature engineering для prediction модели
import pyspark.sql.functions as F

# Temporal features
df = df.withColumn('hour', F.hour('event_ts'))
df = df.withColumn('day_of_week', F.dayofweek('event_ts'))
df = df.withColumn('is_weekend', F.col('day_of_week').isin(1, 7))

# Aggregation features (window)
window_30d = Window.partitionBy('user_id') \
    .orderBy('event_ts') \
    .rangeBetween(-30*86400, 0)

df = df.withColumn(
    'purchases_30d',
    F.count('order_id').over(window_30d)
)
df = df.withColumn(
    'avg_amount_30d',
    F.avg('amount').over(window_30d)
)

# ClickHouse — real-time analytics
# Ingestion: 500K events/sec
# Query latency: p99 < 100ms

CREATE TABLE events (
    event_id UUID,
    user_id UInt64,
    event_type LowCardinality(String),
    amount Decimal(10,2),
    event_time DateTime
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(event_time)
ORDER BY (event_type, user_id, event_time);

-- Dashboard query: revenue last 5 minutes
SELECT
    toStartOfMinute(event_time) as minute,
    sum(amount) as revenue
FROM events
WHERE event_time > now() - INTERVAL 5 MINUTE
GROUP BY minute
ORDER BY minute;

-- ClickHouse: Materialized View для real-time агрегатов
CREATE MATERIALIZED VIEW hourly_revenue
ENGINE = SummingMergeTree()
ORDER BY (hour, product_id)
AS SELECT
    toStartOfHour(event_time) as hour,
    product_id,
    sum(amount) as total_revenue,
    count() as order_count
FROM events
WHERE event_type = 'purchase'
GROUP BY hour, product_id;

-- Query: мгновенный ответ вместо scan 1B rows
SELECT hour, sum(total_revenue)
FROM hourly_revenue
WHERE hour >= today()
GROUP BY hour;

# OLAP операции на примере revenue cube
# Dimensions: date, product, region
# Measure: revenue
#
# Slice: revenue WHERE region = 'Europe'
# Dice: revenue WHERE region IN ('EU','US') AND year = 2024
# Roll-up: daily → monthly → yearly
# Drill-down: yearly → quarterly → monthly
#
# Apache Druid — modern OLAP
# Pre-aggregation at ingestion time
# Sub-second queries on billions of rows
# Native support for time-series roll-up