Orc Hwang's Wiki

개발 학습 기록과 기술 지식을 체계적으로 관리하는 개인 위키입니다.

Stage 7 · 실전 통합 — AIR 파이프라인 · concurrency 종합 · Job 간 동시성

ray air integration concurrency ray-job python ml
📖 시리즈: Ray-Essential (8 / 8)
← 이전: Stage 6 · Ray Serve 서빙 — 로컬 모델 배포 · 스케일/동시성
Ray AIR — Data→Train→Tune→Serve 통합 파이프라인과 Job 간 동시성 확장 Data전처리 Train분산 학습 Tune탐색 ServeHTTP 서빙 클라이언트 POST :8000 Job 간 동시성 ray job submit · --num-cpus KubeRay (전망) 동일 코드 · 클러스터 확장 concurrency 종합: 워커 풀 · map_batches · ray.wait · Job 내부/간
Ray AIR 통합 — **Data(전처리) → Train(분산 학습) → Tune(탐색) → Serve(HTTP 서빙)** 을 하나의 스크립트로 조립해 클라이언트가 `:8000`으로 호출합니다. 아래에는 **Job 간 동시성**(`ray job submit`으로 여러 잡이 클러스터 공유)과 **KubeRay 확장 전망**(같은 코드를 클러스터로)을 배치했습니다.

한눈에 보기

지금까지 배운 Ray의 기능(Core·Data·Train·Tune·Serve)을 하나의 엔드투엔드 파이프라인으로 조립해 마무리합니다. 이게 바로 Ray AIR(AI Runtime)의 비전 — “같은 파이썬 코드가 노트북에서도 클러스터에서도 그대로 돈다” — 입니다. 그리고 시리즈 전반의 concurrency 처리를 종합하고, Job 간 동시성과 KubeRay로의 확장을 조망합니다.

이번 포스트에서 다루는 핵심 질문은 세 가지입니다.

  1. Data→Train→Tune→Serve를 어떻게 한 스크립트로 조립하고 HTTP로 호출하나?
  2. 지금까지의 concurrency 처리(워커 풀·ray.wait·Job 내부)를 어떻게 한곳에 종합하나?
  3. Job 간 동시성(ray job submit)과 KubeRay로 어떻게 확장하나?

AIR 통합 — Data → Train → Tune → Serve

각 단계의 결과가 다음 단계를 먹이는 전형적인 조립입니다.

import ray
from ray import train, tune, serve
from ray.data import from_items
from ray.train import ScalingConfig, RunConfig
from ray.train.torch import TorchTrainer
from ray.train.torch import TorchCheckpoint

ray.init(num_cpus=4)

# ① Data — 전처리
train_ds = from_items([{"x": [i / 10, i / 10], "y": [i / 10]} for i in range(100)])

# ② Train + ③ Tune — Trainer를 Objective로 감싸 학습·탐색
def train_rf(config):                       # (예: 파이토치 대신 간단 회귀로)
    import numpy as np
    from ray import train
    n = config["n"]; coeff = config["coeff"]
    def model(x): return np.array(x["x"]) * coeff
    # ... 실제로는 TorchTrainer로 학습 ...
    train.report({"loss": 0.0})

search_space = {"n": tune.grid_search([10, 50]), "coeff": tune.choice([0.5, 1.0])}
tuner = tune.Tuner(
    tune.with_resources(train_rf, {"cpu": 0.5}),
    param_space=search_space,
    tune_config=tune.TuneConfig(metric="loss", mode="min"),
)
best = tuner.fit().get_best_result(metric="loss", mode="min")
print("최고 config:", best.config)

# ④ Serve — 학습 결과를 deployment로 배포
@serve.deployment
def predictor(request):
    return {"ok": True, **best.config}

serve.run(predictor.bind(), route_prefix="/")

클라이언트에서 최종 HTTP 호출:

curl "http://localhost:8000/?x=1"

완전한 실전에선 ②에서 Stage 5의 TorchTrainer + Stage 3의 get_dataset_shard를 사용하고, ④에서 체크포인트를 로드한 Serve deployment로 연결합니다. 여기서는 전체 흐름을 “한 스크립트”로 잇는 것에 집중합니다.

디자인 원칙 (AIR):

  1. 기능별로 쪼개 사용: 꼭 필요한 라이브러리만 골라 조립(항상 전체가 필요하진 않음).
  2. 통일된 설정 클래스: ScalingConfig/RunConfig/CheckpointConfig가 Train·Tune·Serve에 걸쳐 재사용.
  3. Predictor + Deployment: Stage 5의 Predictor와 Serve Deployment가 학습→서빙의 다리.

concurrency 종합 — 지금까지의 동시성 한곳에

Ray의 동시성은 그림처럼 계층으로 이뤄져 있습니다. 이 시리즈가 배운 전부를 정리하면:

계층 도구 시리즈에서
Task 수준 (무상태 병렬) @ray.remote + ObjectRef, ray.wait 스트리밍 Stage 1
Actor 수준 (상태 병렬) 액터, ActorPool, max_concurrency · async/threaded Stage 2
데이터 수준 (파이프라인 병렬) map_batches(Task/Actor 전략), block 샤딩 Stage 3
공유 상태 안전 액터 메서드 직렬화, 오브젝트 스토어 Stage 2 · 1
Job 간 (별도 프로그램) ray job submit, --num-cpus/--memory 이 단계

워커 풀 크롤러로 이 계층을 관통하는 예시 — Concurrency Essential의 Stage 6 워커 풀을 Ray로 다시 쓰면:

import httpx, ray
from ray.util import ActorPool

ray.init(num_cpus=4)

@ray.remote
class Crawler:
    def fetch(self, url):
        r = httpx.get(url, timeout=5)
        return url, r.status_code, len(r.content)

pool = ActorPool([Crawler.remote() for _ in range(4)])   # 고정 워커 풀
urls = ["https://wiki.orchwang.dev"] * 8

# map_unordered: 완료되는 대로 결과 소비 (스트리밍)
for url, status, size in pool.map_unordered(lambda w, u: w.fetch.remote(u), urls):
    print(url, status, size)

이것은 파이썬 ThreadPool+asyncio, Go 고루틴+채널로 구현했던 동일한 워커 풀을 Ray 액터 풀로 표현한 것입니다 — Concurrency 시리즈의 결론을 Ray가 분산 일반화하는 지점이 바로 여기입니다.

Job 간 동시성 & 확장 — 한 대 이상으로

ray job submit — 여러 잡이 클러스터 공유

지금까지는 하나의 프로그램(Job) 안에서 동시성을 다뤘습니다. 다른 각도의 동시성은 여러 Job이 하나의 Ray 클러스터를 공유하는 것입니다.

# head 노드 기동 후
ray start --head --num-cpus 8

# 여러 Job을 별도 프로그램으로 제출 (각자 runtime_env)
ray job submit --address http://127.0.0.1:8265 -- python script_a.py
ray job submit --address http://127.0.0.1:8265 -- python script_b.py
  • 각 Job은 독립 프로그램으로, 자체 의존성(runtime_env)을 가집니다.
  • 여러 Job이 같은 클러스터의 자원을 공유합니다 = “Job 간 동시성”.
  • 자원 상한으로 한 Job이 다른 Job을 굶기지 않게: 
    ray job submit --address http://127.0.0.1:8265 --num-cpus 4 --memory 2G -- python script_a.py
  • 대비: Job 내부 동시성 → 태스크/액터/max_concurrency. Job 간 동시성 → 여러 ray job submit이 클러스터 공유(자원 상한으로 보호).
  • ⚠️ Job은 클러스터 수명에 묶여 있습니다 — 클러스터가 죽으면 실행 중인 Job들도 죽습니다.

KubeRay 전망 — 같은 코드를 클러스터로 (선택)

이 시리즈가 “한 대”에 집중한 이유는, Ray가 노드 수와 무관하게 같은 코드를 돌리기 때문입니다. 규모가 필요해지면:

  • KubeRay로 쿠버네티스에 RayCluster(head + worker pod), RayJob(배치), RayService(연속 서빙)를 배포.
  • 위에서 만든 AIR 파이프라인 코드를 그대로 클러스터에 제출 — 노트북에서와 동일.

한 노드에서 배운 모든 것이 클러스터에서 그대로 성립한다는 것, 그것이 이 시리즈의 마지막 메시지입니다.

Summary

  • Ray AIR는 Data→Train→Tune→Serve를 한 스크립트로 조립해 HTTP로 최종 서빙합니다 — 같은 코드가 노트북·클러스터에서 그대로.
  • concurrency는 계층입니다: Task → Actor(max_concurrency) → Data(map_batches) → Job 간(ray job submit). 워커 풀 크롤러는 액터 풀로 관통합니다.
  • Job 간 동시성은 여러 ray job submit이 클러스터를 공유하며, --num-cpus/--memory로 자원을 제한합니다.
  • KubeRay로 규모를 키워도 코드는 그대로 — 로컬에서 배운 전부가 확장의 토대입니다.

다음 학습 (Next Learning)