Чем занимался?

Мой опыт как Data Engineer

В своей карьере я занимался проектированием и разработкой полного цикла data pipeline, от сбора данных до аналитики и машинного обучения.

ETL и Data Pipelines

# Разработка критичных ETL процессов
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    "owner": "data-eng",
    "retries": 3,
    "retry_delay": timedelta(minutes=5),
}

dag = DAG(
    "daily_sales_etl",
    default_args=default_args,
    schedule_interval="0 2 * * *",  # 2 AM ежедневно
    start_date=datetime(2023, 1, 1),
)

# Extract: данные с разных источников
def extract_sales_data():
    """Извлекаем продажи из production БД"""
    connection = psycopg2.connect(DB_PROD)
    cursor = connection.cursor()
    cursor.execute("""
        SELECT 
            order_id, customer_id, product_id,
            amount, order_date, status
        FROM orders
        WHERE created_date >= %s
    """, (datetime.now() - timedelta(days=1),))
    
    rows = cursor.fetchall()
    connection.close()
    
    # Сохраняем в staging
    with open("/tmp/sales_raw.csv", "w") as f:
        writer = csv.writer(f)
        writer.writerows(rows)

# Transform: очистка и обогащение
def transform_sales_data():
    """Трансформируем и обогащаем данные"""
    df = pd.read_csv("/tmp/sales_raw.csv")
    
    # Очистка
    df = df.dropna()
    df = df[df["amount"] > 0]
    
    # Обогащение
    df["year"] = pd.to_datetime(df["order_date"]).dt.year
    df["month"] = pd.to_datetime(df["order_date"]).dt.month
    df["day_of_week"] = pd.to_datetime(df["order_date"]).dt.day_name()
    
    # Фильтрация
    df = df[df["status"].isin(["completed", "shipped"])]
    
    df.to_csv("/tmp/sales_transformed.csv", index=False)

# Load: загружаем в data warehouse
def load_to_warehouse():
    """Загружаем в Redshift/BigQuery"""
    df = pd.read_csv("/tmp/sales_transformed.csv")
    
    # Используем COPY команду для быстрой загрузки
    connection = psycopg2.connect(DB_WAREHOUSE)
    cursor = connection.cursor()
    
    with open("/tmp/sales_transformed.csv", "r") as f:
        cursor.copy_from(
            f,
            "stg_sales",
            sep=",",
            columns=["order_id", "customer_id", "product_id", "amount", "order_date", "status", "year", "month", "day_of_week"]
        )
    
    connection.commit()
    connection.close()

# DAG задачи
extract_task = PythonOperator(
    task_id="extract",
    python_callable=extract_sales_data,
    dag=dag,
)

transform_task = PythonOperator(
    task_id="transform",
    python_callable=transform_sales_data,
    dag=dag,
)

load_task = PythonOperator(
    task_id="load",
    python_callable=load_to_warehouse,
    dag=dag,
)

extract_task >> transform_task >> load_task

Работа с Big Data

Apache Spark: Обработка петабайтов данных

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum, avg, count, window

spark = SparkSession.builder \
    .appName("AnalyticsProcessing") \
    .config("spark.sql.shuffle.partitions", "200") \
    .getOrCreate()

# Загружаем данные
events = spark.read.parquet("/warehouse/events")  # Петабайты данных

# Агрегация в реальном времени
user_analytics = (
    events
    .filter(col("event_type").isin(["purchase", "view", "click"]))
    .groupBy(
        col("user_id"),
        window(col("timestamp"), "1 hour")
    )
    .agg(
        count("*").alias("event_count"),
        sum("amount").alias("total_revenue"),
        avg("session_duration").alias("avg_session"),
        collect_set("product_id").alias("products_viewed")
    )
)

# Оптимизированная загрузка в хранилище
user_analytics.repartition(50).write.mode("overwrite") \
    .partitionBy("window") \
    .parquet("/warehouse/user_analytics")

Проектирование Data Warehouse

Star Schema для аналитики

-- Fact таблица: события
CREATE TABLE sales_fact (
    sale_id BIGINT PRIMARY KEY,
    customer_key INT FOREIGN KEY REFERENCES customers_dim(customer_key),
    product_key INT FOREIGN KEY REFERENCES products_dim(product_key),
    time_key INT FOREIGN KEY REFERENCES time_dim(time_key),
    location_key INT FOREIGN KEY REFERENCES location_dim(location_key),
    amount DECIMAL(10, 2),
    quantity INT,
    discount_amount DECIMAL(10, 2)
);

-- Dimension таблицы: справочники
CREATE TABLE customers_dim (
    customer_key INT PRIMARY KEY,
    customer_id INT,
    name VARCHAR(255),
    email VARCHAR(255),
    country VARCHAR(100),
    segment VARCHAR(50),
    created_date DATE
);

CREATE TABLE time_dim (
    time_key INT PRIMARY KEY,
    date DATE,
    year INT,
    quarter INT,
    month INT,
    day INT,
    day_of_week VARCHAR(10),
    is_weekend BOOLEAN
);

-- Индексы для быстрых запросов
CREATE INDEX idx_sales_customer ON sales_fact(customer_key);
CREATE INDEX idx_sales_time ON sales_fact(time_key);
CREATE INDEX idx_sales_product ON sales_fact(product_key);

Оптимизация запросов

# Использование материализованных представлений
spark.sql("""
CREATE MATERIALIZED VIEW mv_monthly_sales AS
SELECT 
    DATE_TRUNC(month, order_date) as month,
    customer_id,
    SUM(amount) as total_sales,
    COUNT(*) as order_count,
    AVG(amount) as avg_order
FROM sales_fact
GROUP BY 1, 2
WITH DATA;
""")

# Партиционирование для быстрого pruning
spark.sql("""
CREATE TABLE events_partitioned (
    event_id STRING,
    user_id STRING,
    event_type STRING,
    amount DECIMAL(10, 2),
    timestamp TIMESTAMP
)
PARTITIONED BY (year INT, month INT, day INT)
""")

Мониторинг и качество данных

import pandas as pd
from great_expectations.dataset import PandasDataset

# Data Quality checks
def validate_sales_data(df):
    """
    Проверяем качество данных перед загрузкой в production
    """
    # Проверка nulls
    assert df["customer_id"].isnull().sum() == 0, "customer_id has nulls"
    assert df["amount"].isnull().sum() == 0, "amount has nulls"
    
    # Проверка диапазона
    assert (df["amount"] > 0).all(), "Negative amounts found"
    assert (df["order_date"] <= pd.Timestamp.now()).all(), "Future dates found"
    
    # Проверка уникальности
    assert df["order_id"].is_unique, "Duplicate order_ids"
    
    # Проверка бизнес-правил
    assert df[df["status"] == "completed"]["amount"].sum() > 0, "No completed sales"
    
    return True

# Мониторинг pipeline
def log_pipeline_metrics(job_name, rows_processed, duration_seconds):
    """
    Логируем метрики для мониторинга
    """
    throughput = rows_processed / duration_seconds
    
    # Отправляем в Prometheus/DataDog
    metrics_client.gauge(
        f"pipeline.{job_name}.throughput",
        throughput,
        tags=[f"env:{ENV}"]
    )

Работа с различными источниками данных

# API
import requests

def fetch_from_api(url, headers):
    response = requests.get(url, headers=headers)
    return response.json()

# Databases
import psycopg2
import mysql.connector

conn = psycopg2.connect("postgresql://user:pass@host/db")

# S3 / Cloud Storage
import boto3

s3 = boto3.client("s3")
obj = s3.get_object(Bucket="data-bucket", Key="file.parquet")

# Message Queues (Kafka)
from kafka import KafkaConsumer

consumer = KafkaConsumer(
    "events",
    bootstrap_servers=["localhost:9092"],
    group_id="data-eng-group"
)

Автоматизация и Infrastructure as Code

# Docker контейнер для ETL
FROM python:3.9

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY src/ .

ENV PYTHONUNBUFFERED=1

CMD ["python", "main.py"]

# Kubernetes deployment
from kubernetes import client, config

config.load_incluster_config()
api = client.BatchV1Api()

job_spec = {
    "apiVersion": "batch/v1",
    "kind": "Job",
    "metadata": {"name": "etl-job"},
    "spec": {
        "template": {
            "spec": {
                "containers": [{
                    "name": "etl",
                    "image": "my-registry/etl:latest",
                    "env": [
                        {"name": "DB_HOST", "value": "postgres-service"},
                    ]
                }]
            }
        }
    }
}

api.create_namespaced_job(namespace="default", body=job_spec)

Навыки и инструменты

Базы данных:

• PostgreSQL, MySQL, Greenplum
• Redshift, BigQuery, Snowflake
• MongoDB, Cassandra

Big Data:

• Apache Spark, Hadoop
• Kafka, Airflow
• Presto, Druid

Языки:

• Python (Pandas, PySpark)
• SQL (сложные запросы, оптимизация)
• Scala (Spark jobs)

Cloud:

• AWS (S3, Redshift, Glue, Lambda)
• Google Cloud (BigQuery, Dataflow)
• Azure (Synapse, Data Lake)

Infrastructure:

• Docker, Kubernetes
• Terraform, CloudFormation
• Jenkins, GitLab CI

Ключевые достижения

• Ускорение ETL в 10 раз оптимизацией JOIN-ов и индексов
• Построение data warehouse с 100+ таблиц для 500+ пользователей
• Real-time analytics для миллиона событий в час
• Снижение стоимости облака на 40% через оптимизацию партиционирования
• Аудит качества данных с 95% покрытием

Мой опыт показывает, что успешный Data Engineer должен понимать не только инструменты, но и архитектуру систем, производительность и бизнес-требования.