AI-воркфлоу на стероидах: Flyte 2.0 на EKS для тех, кому надоели бесконечные Airflow-даги

Когда Airflow начинает стонать под весом ваших моделей

Вы помните тот момент? Когда ваш AI-пайплайн из 12 шагов в Airflow начинает падать на третьем, потому что GPU-нода в EKS не успела прогреть драйверы. Когда вы тратите больше времени на отладку дагов, чем на улучшение модели. Когда мониторинг выглядит как археологические раскопки в логах CloudWatch.

Вот тогда и появляется Flyte от Union.ai. Не очередной "инструмент оркестрации", а полноценная платформа, которая понимает, что AI-воркфлоу — это не просто "запусти скрипт, подожди, сохрани результат". Это сложные графы вычислений с GPU, проверками данных, версионированием моделей и постоянными перезапусками.

1 Зачем вообще это нужно?

Потому что типичный сценарий выглядит так:

• Ваша команда обучила модель в Jupyter Notebook
• Вы запихнули логику в Python-скрипт и запускаете его на инстансе с GPU
• Потом добавили второй скрипт для предобработки данных
• Третий — для постобработки
• Четвёртый — для отправки в прод
• И теперь у вас 4 скрипта, которые нужно запускать в правильном порядке, с правильными параметрами, отслеживая версии данных и моделей

Flyte решает это одной декларацией на Python. Но не торопитесь с установкой — сначала поговорим об архитектуре.

Архитектура, которая не сломается в пятницу вечером

Flyte на EKS состоит из трёх ключевых компонентов, и если вы пропустите любой из них — будете страдать по ночам.

Самая частая ошибка — запихнуть всё в один pod. Flyte Admin начнёт тормозить, когда у вас будет 50+ параллельных воркфлоу. Propeller начнёт терять поды. Datacatalog упадёт, потеряв метаданные ваших моделей.

2 Развертывание: не повторяйте моих ошибок

Официальная документация предлагает использовать helm. Это работает, но есть нюансы, которые вы узнаете только на собственном горьком опыте.

# Не делайте так (это сломается через неделю)
helm install flyte flyteorg/flyte -n flyte

# Делайте так (февраль 2026, актуально)
# 1. Создаём namespace с правильными лимитами
kubectl create namespace flyte
kubectl label namespace flyte istio-injection=enabled  # если используете Istio

# 2. Кастомные values для продакшена
echo '
# flyte-values.yaml
flyteadmin:
  replicaCount: 3
  resources:
    requests:
      memory: "2Gi"
      cpu: "500m"
    limits:
      memory: "4Gi"
      cpu: "2000m"

propeller:
  replicaCount: 2
  config:
    maxWorkflowRetries: 3
    maxTaskRetries: 5

storage:
  type: s3
  s3:
    bucket: "your-flyte-artifacts-2026"  # уникальное имя!
    region: "eu-west-1"
    serviceAccount: "flyte-sa"

database:
  host: "flyte-db.cluster-identifier.eu-west-1.rds.amazonaws.com"
  port: 5432
  username: "flyteadmin"
  passwordSecret: "flyte-db-password"
' > flyte-values.yaml

# 3. Устанавливаем с правильными параметрами
helm install flyte flyteorg/flyte \
  -n flyte \
  -f flyte-values.yaml \
  --version 2.3.1  # самая свежая стабильная на февраль 2026

Видите эти memory limits? Без них Flyte Admin начнёт съедать всю память ноды при обработке сложных графов. А параметр maxWorkflowRetries — это ваша страховка от временных сбоев в AWS.

Python SDK 2.0: когда код выглядит как поэма

Старый Flyte SDK требовал декораторов, специальных типов, кучи boilerplate. В версии 2.0 (актуальной на февраль 2026) всё изменилось. Теперь это выглядит... элегантно.

# train_model.py
# Flyte Python SDK 2.3.0+
from flytekit import task, workflow, Resources
from flytekit.types.directory import FlyteDirectory
from flytekit.types.file import FlyteFile
from flytekitplugins.awssagemaker import SageMakerTrainingTask
import pandas as pd
from typing import Tuple

# Простая задача с указанием ресурсов
@task(
    limits=Resources(cpu="2", mem="4Gi", gpu="1"),
    retries=3,
    cache=True,  # Кеширование включено!
    cache_version="1.0"
)
def preprocess_data(
    raw_data: FlyteFile
) -> Tuple[pd.DataFrame, pd.DataFrame]:
    """Загрузка и предобработка данных с автоматическим кешированием."""
    import joblib
    from sklearn.model_selection import train_test_split
    
    df = pd.read_parquet(raw_data.path)
    # ... логика обработки
    
    train_df, test_df = train_test_split(df, test_size=0.2)
    return train_df, test_df

# Задача с GPU для обучения модели
@task(
    limits=Resources(cpu="4", mem="16Gi", gpu="1", storage="50Gi"),
    environment={
        "CUDA_VISIBLE_DEVICES": "0",
        "TF_FORCE_GPU_ALLOW_GROWTH": "true"
    }
)
def train_model(
    train_data: pd.DataFrame,
    model_config: dict
) -> FlyteDirectory:
    """Обучение модели с GPU поддержкой."""
    import tensorflow as tf
    import os
    
    # Автоматическое использование GPU через EKS
    strategy = tf.distribute.MirroredStrategy()
    
    with strategy.scope():
        model = create_model(**model_config)
        model.fit(train_data, epochs=50)
    
    # Сохраняем в FlyteDirectory для версионирования
    output_dir = "/tmp/model_artifacts"
    model.save(f"{output_dir}/model.h5")
    
    return FlyteDirectory(path=output_dir)

# Главный воркфлоу
@workflow
def ml_pipeline(
    data_path: FlyteFile,
    config: dict = {"layers": [128, 64], "dropout": 0.3}
) -> FlyteDirectory:
    """Полный пайплайн от данных до модели."""
    train_df, test_df = preprocess_data(raw_data=data_path)
    model_dir = train_model(train_data=train_df, model_config=config)
    
    return model_dir

Обратите внимание на cache=True и cache_version. Это магия Flyte 2.0. Если входные данные не изменились — задача не выполняется, возвращается предыдущий результат. Экономит часы GPU-времени.

Интеграция с AWS: где спрятаны все грабли

Flyte на EKS без интеграции с AWS сервисами — как Ferrari без бензина. Красиво, но никуда не едет.

3 S3 для артефактов (не забудьте про lifecycle)

Каждый артефакт в Flyte (данные, модели, метрики) летит в S3. К февралю 2026 стандартная конфигурация выглядит так:

# flyte-s3-policy.yaml
# IAM Policy для Service Account Flyte
apiVersion: v1
kind: ConfigMap
metadata:
  name: flyte-aws-config
  namespace: flyte
data:
  aws.region: "eu-west-1"
  aws.s3.endpoint: "https://s3.eu-west-1.amazonaws.com"
  storage.metadata.prefix: "metadata"
  storage.data.prefix: "data"
  # Критически важный параметр для больших моделей
  storage.maxDownloadMBs: "100"  # лимит скачивания в MB/s
---
# Lifecycle правило для автоматической очистки
# (чтобы S3 bucket не съел все деньги)
aws s3api put-bucket-lifecycle-configuration \
  --bucket your-flyte-artifacts-2026 \
  --lifecycle-configuration '
  {
    "Rules": [
      {
        "ID": "DeleteOldArtifacts",
        "Status": "Enabled",
        "Prefix": "data/",
        "Expiration": { "Days": 90 },
        "NoncurrentVersionExpiration": { "NoncurrentDays": 30 }
      }
    ]
  }'

Без lifecycle правила через три месяца вы получите счёт на несколько тысяч долларов за хранение 10ТБ промежуточных данных, которые уже никому не нужны.

4 Secrets Manager для конфигов моделей

Хранить API keys и конфигурации моделей в коде — преступление в 2026 году. Flyte умеет работать с AWS Secrets Manager прозрачно:

from flytekit import task, Secret

@task(
    secret_requests=[
        Secret(
            group="openai",
            key="api_key",
            mount_requirement=Secret.MountType.FILE
        ),
        Secret(
            group="huggingface",
            key="token",
            mount_requirement=Secret.MountType.ENV_VAR
        )
    ]
)
def call_external_apis(data: pd.DataFrame) -> dict:
    """Задача с автоматическим подтягиванием секретов из AWS."""
    # Секреты автоматически монтируются в /etc/secrets/openai/api_key
    # или в переменные окружения
    
    with open("/etc/secrets/openai/api_key", "r") as f:
        openai_key = f.read().strip()
    
    # Или через переменные окружения
    hf_token = os.environ.get("HUGGINGFACE_TOKEN")
    
    # ... вызов API
    return results

Настройка в EKS требует IAM Roles for Service Accounts (IRSA). Если пропустите этот шаг — задачи будут падать с ошибками доступа.

Масштабирование: когда одна модель превращается в сотню

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

Flyte 2.0 предлагает два подхода, и выбор зависит от того, насколько вы цените своё время.

# Пример Map Task для параллельного обучения
from flytekit import task, workflow, map_task
from typing import List

@task
def train_single_model(
    data_chunk: pd.DataFrame,
    params: dict
) -> float:
    """Обучение модели на одном чанке данных."""
    # ... обучение
    return accuracy

@workflow
def parallel_training(
    data_chunks: List[pd.DataFrame],
    param_grid: List[dict]
) -> List[float]:
    """Параллельное обучение множества моделей."""
    # Map Task автоматически распараллелит
    accuracies = map_task(
        train_single_model,
        concurrency=10,  # максимум параллельных задач
        min_success_ratio=0.8  # допускаем 20% неудач
    )(
        data_chunk=data_chunks,
        params=param_grid
    )
    
    return accuracies

Параметр concurrency=10 — это не просто число. Это баланс между скоростью и нагрузкой на EKS. Поставьте слишком большое — убьёте ноды. Слишком маленькое — будете ждать часами.

Мониторинг и дебаг: что смотреть, когда всё падает

Flyte Dashboard — это красиво, но для продакшена нужны метрики в Prometheus и алерты в Slack. К февралю 2026 стандартный стек выглядит так:

# flyte-monitoring.yaml
apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: flyte-monitor
  namespace: flyte
spec:
  selector:
    matchLabels:
      app: flyteadmin
  endpoints:
  - port: http-metrics
    interval: 30s
    path: /metrics
  - port: propeller-metrics
    interval: 15s  # чаще, потому что больше событий
---
# Ключевые метрики для алертов
alert: FlyteWorkflowFailures
expr: rate(flyte:workflow:failed[5m]) > 0.1
for: 5m
labels:
  severity: critical
annotations:
  summary: "{{ $labels.workflow_name }} failing at {{ $value }} failures/min"
  description: "More than 10% of Flyte workflows are failing"
---
alert: FlyteTaskQueueBacklog
expr: flyte:propeller:queue_size > 100
for: 10m
labels:
  severity: warning
annotations:
  summary: "Flyte task queue backlog {{ $value }} tasks"

Но метрики — это половина дела. Вторая половина — логи. И здесь есть подвох: логи Flyte в EKS по умолчанию идут в stdout контейнеров. Без Fluentd или аналогичного решения вы их потеряете при перезапуске пода.

Миграция с Airflow/Luigi/кустарных скриптов

Если у вас уже есть работающий пайплайн, не надо переписывать всё с нуля. Flyte умеет оборачивать существующий код постепенно.

Стратегия, проверенная на трёх проектах:

1. Начинаем с самого "болезненного" шага — обычно это обучение модели с GPU
2. Оборачиваем его в @task, оставляя остальной пайплайн как есть
3. Запускаем параллельно: старый пайплайн и новый шаг в Flyte
4. Сравниваем результаты, настраиваем ресурсы
5. Переходим к следующему шагу
6. Когда все шаги перенесены, собираем из них @workflow

Самая частая ошибка — попытка перенести всё сразу. Потратите месяц, получите сломанный пайплайн и разочарование в технологии.

Стоимость: сколько это стоит на самом деле

Давайте без иллюзий. Flyte на EKS — не бесплатно. Но и Airflow с GPU-нодами тоже не дёшев. Разница в том, что Flyte экономит на операционных расходах.

Реальная картина на февраль 2026 для middle-scale проекта:

• EKS control plane: $73/месяц (фиксировано)
• 3 ноды m5.2xlarge для Flyte Admin/Propeller: ~$600/месяц
• GPU ноды (g5.2xlarge) по требованию: $1.20/час, но только когда идут тренировки
• S3 для артефактов: ~$50-200/месяц в зависимости от lifecycle правил
• RDS PostgreSQL для метаданных: ~$150/месяц

Итого: ~$1000-1500/месяц базовых расходов плюс GPU по факту использования. Дорого? Да. Но дешевле, чем три инженера MLOps, которые постоянно чинят сломанные пайплайны.

А что если... (FAQ от тех, кто уже обжёгся)

Что делать, если задача зависла и не завершается?
Flyte 2.0 имеет timeout на уровне задачи и workflow. Но если задача зависла без возможности отменить (например, заняла GPU), нужно убивать pod вручную через kubectl. Добавляйте всегда timeout в @task.

Как версионировать модели правильно?
Используйте FlyteDirectory с автоматическим versioning. Каждый запуск workflow создаёт новую версию артефакта. Плюс добавляйте теги через metadata.

Можно ли запускать Flyte задачи из Jupyter Notebook?
Да, через flytekit remote. Но не злоупотребляйте — это для дебага, не для продакшена.

Что лучше: одна большая задача или много маленьких?
Много маленьких. Легче дебажить, кешировать, масштабировать. Исключение — когда задача должна быть атомарной из-за GPU memory.

Как интегрировать с CI/CD?
Flyte имеет CLI для регистрации workflows. Добавляйте в pipeline команду `pyflyte register` после успешных тестов.

Что дальше? (Spoiler: Union.ai готовит сюрприз)

По слухам из коммитов в репозитории Union.ai, к концу 2026 года ожидается интеграция Flyte с IncidentFox — AI SRE для автоматического исправления падающих воркфлоу. Представьте: ваш пайплайн падает из-за нехватки памяти, а AI уже увеличивает limits и перезапускает задачу до того, как вы получите алерт.

Но пока это слухи. А сегодня у вас есть рабочий инструмент, который превращает хаос AI-экспериментов в предсказуемый инженерный процесс. Главное — не забудьте про lifecycle правила в S3. Серьёзно, счета приходят неожиданно.