이 글은 명시한 공식 문서 버전을 기준으로 한 운영 가이드다. 실제 EMR의 버전·동시 세션 수·시작 시간·유휴 메모리를 측정한 결과는 포함하지 않는다.

들어가며

EMR 노트북에서 첫 셀을 실행하면 준비 시간이 길고, 다음 셀부터는 비교적 빠르게 실행되는 경험을 하게 된다. Livy를 사용하는 환경에서는 첫 실행 시 원격 Spark 실행 환경을 준비하고 이후 코드를 같은 환경에서 처리하기 때문이다.

Apache Livy는 HTTP로 Spark 코드와 애플리케이션을 제출하고 결과를 조회하는 서버다. 이 글은 EMR on EC2에서 Spark를 YARN으로 실행하는 대화형 Livy 세션을 기준으로 설명한다. EMR Serverless의 Livy 엔드포인트는 인증과 설정 방식이 달라 마지막에 따로 다룬다. AWS Livy 소개

Livy 세션이 starting에서 멈추거나 생성에 실패할 때

POST /sessions의 HTTP 요청 성공과 Spark 실행 환경 준비 완료는 다르다. 응답에서 받은 session ID로 상태를 조회하고 idle이 되었는지 확인한다. starting이 길어지면 같은 세션의 YARN app ID와 로그를 함께 확인해 원인을 좁힌다.

1. 생성 요청 자체가 실패한다면 HTTP 응답과 Livy 서버 로그에서 요청이 서버에 도달했는지, 인증·요청 설정 오류가 있는지 확인한다.
2. 세션은 생겼지만 starting이라면 YARN 애플리케이션 생성 여부와 Diagnostics를 확인한다. ACCEPTED 대기는 AM 한도·큐 자원·노드 배치 조건과 연결해 조사한다.
3. 세션이 dead이거나 조회 시 404라면 Driver 종료, 세션 타임아웃, 명시적 삭제, Livy 재시작 이력을 확인한다. 타임아웃 값을 늘리기 전에 어느 계층에서 종료됐는지 구분한다.

조회에는 GET /sessions/{sessionId}, GET /sessions/{sessionId}/state, GET /sessions/{sessionId}/log를 사용한다. 구체적인 요청 예제와 상태별 점검표는 아래에 정리했다. Livy REST API

Livy 세션은 무엇을 유지하는가

일반적인 YARN 구성에서 대화형 Livy 세션 하나는 하나의 Spark 애플리케이션과 연결된다. 그 안에 코드를 해석하는 인터프리터와 SparkContext가 있고, Driver와 Executor가 작업을 수행한다. 셀을 실행할 때마다 새 애플리케이션을 만드는 대신 같은 세션에 코드 조각인 statement를 제출한다. Livy 세션 생성 구현

노트북 또는 HTTP 클라이언트
  └─ Livy 서버: 세션 생성, statement 접수, 결과 조회
      └─ Livy 세션 → YARN 애플리케이션
          ├─ Driver: 인터프리터, SparkContext, SparkSession
          └─ Executor들: 분산 데이터 처리

같은 PySpark 인터프리터에 아래 두 코드를 차례로 제출하면 두 번째 코드에서 앞서 만든 변수를 사용할 수 있다.

# 첫 번째 statement
events = spark.range(1000)

# 두 번째 statement
print(events.count())

여기서 events는 DataFrame을 가리키는 변수다. 변수를 유지한다고 모든 데이터가 메모리에 캐시되는 것은 아니다. cache()나 persist()로 저장한 결과와 단순한 DataFrame 참조를 구분해야 한다.

SparkSession과 같은 개념인가

spark.newSession()은 SQL 설정과 임시 뷰 등을 분리하지만 SparkContext와 테이블 캐시는 공유한다. 새 Livy 세션이나 새 YARN 애플리케이션을 생성하는 API가 아니다. 따라서 사용자별 자원 격리를 newSession()만으로 구현할 수는 없다. SparkSession.newSession 문서

대화형 sessions와 batches 선택하기

예를 들어 데이터를 읽고 분포를 확인한 뒤 조건을 바꾸는 분석에는 대화형 세션이 편리하다. 반면 매일 같은 입력 규칙으로 실행하는 ETL은 실행 상태가 독립적인 batch 제출이나 EMR Step으로 구성하는 편이 재실행을 관리하기 쉽다. 이는 두 API의 실행 방식에 따른 설계 판단이다. Livy REST API

비동기 실행 흐름과 상태 읽기

세션 생성 요청이 성공해도 Spark 실행 환경이 즉시 준비된 것은 아니다. 생성 응답의 ID를 저장하고 idle 상태까지 기다린 뒤 statement를 제출한다. Livy 실행 예제

POST /sessions
  → 세션 starting → idle
  → POST /sessions/{sessionId}/statements
  → statement waiting → running → available
  → output.status 확인
  → 후속 statement 제출 또는 DELETE /sessions/{sessionId}

전체 상태 목록은 Livy REST 객체 정의에서 확인할 수 있다.

특히 available을 실행 성공으로 해석하면 안 된다. 코드가 예외를 발생시켜도 출력이 준비되면 available이 될 수 있고, 실제 성공 여부는 output.status의 ok 또는 error로 구분한다. 오류 출력에는 ename, evalue, traceback이 포함될 수 있다. HTTP 성공, statement 완료, 사용자 코드 성공을 각각 확인해야 한다. Livy 인터프리터 실행 구현

Python으로 생성부터 종료까지 실행하기

다음 예제는 Livy가 설치된 EMR on EC2 클러스터와 Python requests 패키지를 전제로 한다. 기본 포트는 8998이다. 로컬에서 실행하려면 접근 가능한 Primary 노드로 SSH 터널을 먼저 연결한다. Livy 연결 예제

# 키 파일과 호스트를 실제 값으로 교체한다.
ssh -i /path/to/key.pem -N \
  -L 127.0.0.1:8998:localhost:8998 hadoop@primary-host

아래 코드는 HTTP·Kerberos 인증을 별도로 요구하지 않는 실습 환경을 가정한다. TLS나 Kerberos가 활성화되어 있으면 해당 클러스터의 인증과 인증서 설정을 적용해야 한다. X-Requested-By는 CSRF 검사에 사용하는 헤더이며 인증 수단은 아니다. Livy 서버 설정

import time
import uuid

import requests

LIVY_URL = "http://127.0.0.1:8998"
client = requests.Session()
client.headers.update({"X-Requested-By": "livy-session-example"})
session_id = None
session_name = f"livy-demo-{uuid.uuid4().hex[:8]}"


def api(method, path, **kwargs):
    response = client.request(
        method, LIVY_URL + path, timeout=(5, 30), **kwargs
    )
    response.raise_for_status()
    return response.json() if response.content else None


def wait_for(path, ready, failed, wait_seconds):
    deadline = time.monotonic() + wait_seconds
    while time.monotonic() < deadline:
        result = api("GET", path)
        state = result["state"]
        if state in ready:
            return result
        if state in failed:
            raise RuntimeError(f"{path}: {result}")
        time.sleep(2)
    raise TimeoutError(f"Polling timeout: {path}")


try:
    print(f"Creating session: {session_name}")
    created = api("POST", "/sessions", json={
        "kind": "pyspark",
        "name": session_name,
        "queue": "default",
        "driverMemory": "2g",
        "executorMemory": "2g",
        "executorCores": 1,
        "numExecutors": 2,
        "conf": {"spark.dynamicAllocation.enabled": "false"},
    })
    session_id = created["id"]
    session_path = f"/sessions/{session_id}"
    print(f"Livy session ID: {session_id}")
    ready = wait_for(
        session_path, {"idle"},
        {"error", "dead", "killed", "success", "shutting_down"}, 600,
    )
    print(f"YARN app ID: {ready.get('appId')}")

    statement = api("POST", session_path + "/statements", json={
        "code": "events = spark.range(1000)\nprint(events.count())"
    })
    statement_path = session_path + f"/statements/{statement['id']}"
    print(f"Statement ID: {statement['id']}")
    completed = wait_for(
        statement_path, {"available"}, {"error", "cancelled"}, 300,
    )
    output = completed.get("output") or {}
    if output.get("status") != "ok":
        raise RuntimeError(f"Statement execution failed: {output}")
    print(output.get("data", {}).get("text/plain", ""))
finally:
    if session_id is not None:
        try:
            api("DELETE", f"/sessions/{session_id}")
        except requests.RequestException as exc:
            print(f"Cleanup failed; inspect session {session_id}: {exc}")
    client.close()

공식 예제의 비동기 조회 흐름에 대기 제한, 오류 검사, 정리 절차를 추가했다. Executor 수를 읽기 쉽게 하기 위해 동적 할당을 껐으며, 이 값은 운영 권장값이 아니다. Queue에서 Driver·Executor 메모리와 overhead를 수용할 수 있어야 한다. Livy 예제, Spark on YARN 리소스 설정

이 예제의 polling 제한은 클라이언트가 기다리는 시간이다. 제한에 도달하면 finally에서 이 예제가 만든 세션의 종료를 요청한다. 서버에서 실제로 종료되었는지는 세션 상태와 YARN 애플리케이션 상태로 확인한다.

[!WARNING] 생성 POST가 서버에서 처리된 뒤 응답만 유실되면 클라이언트에는 session ID가 없을 수 있다. 이 경우 예제의 finally로 정리할 수 없다. 세션 이름과 생성 시각, 서버 로그로 생성 여부를 확인한다. Statement POST도 응답 유실만 보고 재전송하면 데이터 쓰기가 중복될 수 있으므로 요청 식별 정보를 기록하고, 쓰기 작업의 중복 방지 정책을 별도로 설계한다.

세션 재사용과 리소스 비용

세션을 재사용하면 애플리케이션 초기화와 라이브러리 준비를 반복하지 않고 변수와 캐시를 이어서 사용할 수 있다. 대신 세션이 오래 살아 있을수록 이전 실행의 상태가 다음 실행에 영향을 주기 쉽다.

별도 Spark 애플리케이션은 자체 Executor 프로세스를 갖는다. 동적 할당은 유휴 Executor를 줄일 수 있지만 Driver까지 종료하는 기능은 아니다. 캐시와 셔플 유지 조건에 따라 Executor 회수도 달라진다. 따라서 Livy의 idle을 자원 사용량 0으로 해석하면 안 된다. Spark Job Scheduling

세션 개수를 늘릴 때는 애플리케이션별 Driver·AM 비용과 YARN Queue의 동시 수용 능력을 함께 계산해야 한다. 많은 유휴 세션이 새로운 세션의 AM 할당을 막는지 확인하는 것도 운영 점검 항목이다.

같은 Livy 세션에 HTTP 요청을 여러 개 보낸다고 statement들이 모두 동시에 실행되는 것도 아니다. Apache Livy 0.8.0의 일반 REPL 실행 경로는 statement를 단일 실행 스레드에 순서대로 넣는다. Spark 작업 자체의 분산 처리와 Livy statement 실행 순서는 구분해야 한다. Livy REPL 구현

분석이 끝난 캐시와 결과는 어떻게 정리할까

다음은 한 세션 안에서 실행하는 작은 예제다. 같은 DataFrame을 여러 번 분석한 뒤 필요 없어진 캐시를 명시적으로 해제한다.

events = spark.range(1000).cache()
try:
    print(events.count())
    events.createOrReplaceTempView("session_events")
    spark.sql("SELECT COUNT(*) FROM session_events").show()
finally:
    spark.catalog.dropTempView("session_events")
    events.unpersist(blocking=True)

unpersist()는 저장된 DataFrame 블록을 제거한다. Python 변수의 참조를 지우는 del events와 역할이 다르며, 캐시 해제만으로 Livy 세션이나 Driver가 종료되지는 않는다. DataFrame.unpersist

결과 전달도 크기에 맞춰 설계한다. show()나 제한된 샘플은 노트북으로 조회하고, 큰 결과는 S3 등의 저장소에 기록한 뒤 경로를 전달하는 방식이 적합하다. toPandas()는 결과 전체를 Driver 메모리에 모으므로 분산 처리한 데이터가 크면 마지막 결과 변환에서 실패할 수 있다. DataFrame.toPandas

동적 할당과 EMR 노드 증설은 다른 단계다

Spark 동적 할당은 애플리케이션의 Executor 수를 조정하고, EMR의 클러스터 확장은 실행할 EC2 노드 용량을 조정한다. Executor를 더 요청해도 YARN에 자원이 없으면 즉시 늘어나지 않는다. 동적 할당에는 External Shuffle Service나 Shuffle Tracking 등 지원되는 셔플 보존 방식이 필요하다. EMR 릴리스에 이미 구성된 방식을 확인한 뒤 설정한다. Spark 동적 자원 할당

minExecutors, maxExecutors와 유휴 회수 설정은 Executor에 적용된다. 캐시·셔플 조건 때문에 회수가 지연될 수 있으며, 세션 개수와 Driver 비용을 줄이려면 별도의 Livy 종료 정책이 필요하다.

EMR에서 배포 모드와 설정 적용 시점

livy.spark.deploy-mode는 Livy가 시작하는 Spark Driver의 위치를 결정한다. YARN cluster 모드에서는 AM 프로세스 안에서 Driver가 실행되고, client 모드에서는 제출 측에서 실행된다. Livy 서버와 Spark Driver는 구분되는 프로세스다. Spark on YARN 배포 모드

AWS는 EMR Studio용 클러스터에서 cluster 모드를 사용해 Primary 노드의 메모리 부담을 낮추는 구성을 안내한다. 다음은 클러스터 생성 시 전달할 EMR configuration JSON 예시다. EMR Studio 클러스터 설정

[
  {
    "Classification": "livy-conf",
    "Properties": {
      "livy.spark.deploy-mode": "cluster",
      "livy.server.session.timeout": "6h"
    }
  }
]

Driver가 실제로 Core와 Task 중 어디에 배치되는지는 YARN의 AM Node Label Expression 등에도 영향을 받는다. Livy 세션에서 이 정책을 지정하려면 세션 생성 요청의 conf에 spark.yarn.am.nodeLabelExpression 같은 Spark 속성을 전달하고, 클러스터의 라벨과 Queue 설정을 함께 확인한다. Spark의 YARN 라벨 설정

Driver 메모리 같은 시작 설정은 세션 생성 시 지정한다. 이미 Driver가 실행된 뒤 노트북에서 spark.conf.set()을 호출해 프로세스 메모리를 늘릴 수는 없다. Sparkmagic 기반 노트북에서는 아래처럼 세션 설정을 전달할 수 있다. AWS Driver 메모리 설정 예제

%%configure -f
{
  "driverMemory": "4g",
  "executorMemory": "4g"
}

기존 세션을 재시작하는 설정 변경에서는 변수와 임시 상태를 잃을 수 있다. 세션을 다시 만들기 전에 필요한 결과를 외부 저장소에 기록한다.

Python 패키지는 어디에 설치해야 하는가

노트북의 로컬 Python, 원격 Driver, Executor의 Python 환경은 다를 수 있다. 노트북에서 pip install을 했다고 원격 PySpark 코드에서 바로 import할 수 있는 것은 아니다. 특히 UDF 안에서 import가 실패한다면 Executor 환경을 확인한다.

Livy의 파일 전달 필드는 REST API, Python 환경 패키징의 제약은 Spark Python Package Management를 기준으로 확인한다.

아래는 자체 모듈을 배포하는 세션 생성 요청 예시다. amzn-s3-demo-bucket 경로에 실제 ZIP을 올리고, 원격 실행 환경이 해당 객체를 읽을 수 있어야 한다. ZIP 내부의 import 경로도 확인한다.

{
  "kind": "pyspark",
  "name": "analysis-with-modules",
  "pyFiles": ["s3://amzn-s3-demo-bucket/libs/analysis_helpers.zip"]
}

타임아웃은 적용 대상을 구분한다

ttl과 heartbeatTimeoutInSecond는 의미가 다르다. 전자는 비활성 세션의 유지 시간이고 후자는 heartbeat가 끊긴 세션 판별에 사용한다. 짧은 heartbeat 제한을 지정하려면 클라이언트의 heartbeat 동작도 확인해야 한다. Livy 세션 요청 필드

Livy 0.8.0 설정 템플릿에는 livy.server.session.timeout-check.skip-busy가 있으며 기본값은 false로 표기되어 있다. 실행 중인 세션도 비활성 판단에 따라 종료 대상이 될 수 있다는 설명이 있으므로, 긴 작업이 실행 중이라는 이유만으로 타임아웃에서 제외된다고 가정하면 안 된다. EMR 배포 버전의 지원 여부와 실제 설정을 확인한다. Livy 타임아웃 설정

livy-conf JSON은 EMR 구성 입력 형식이다. 서버의 /etc/livy/conf/livy.conf를 직접 확인할 때는 다음과 같은 속성 형식으로 읽는다.

livy.spark.deploy-mode = cluster
livy.server.session.timeout = 6h

설정 변경에는 Livy 재시작이 필요할 수 있다. 무조건 타임아웃을 끄기보다 분석 시간과 유휴 자원 비용에 맞춰 유지 시간을 정한다. AWS 세션 타임아웃 안내

장애가 나면 세 가지 ID로 추적하기

요청 로그에 Livy session ID, YARN app ID, statement ID를 함께 남기면 어느 계층에서 멈췄는지 찾기 쉽다. Statement ID는 세션 안에서 구분하고, Livy ID와 YARN ID를 같은 값으로 취급하지 않는다.

아래의 12와 app ID는 실제 값으로 바꾼다.

curl --fail --silent --show-error \
  'http://127.0.0.1:8998/sessions/12/log?from=0&size=100'

yarn application -status application_1234567890000_0042
yarn logs -applicationId application_1234567890000_0042

Livy 로그는 세션 수준의 단서이고, Driver·Executor의 상세 원인은 YARN 로그와 Spark UI에서 확인한다. yarn logs로 조회할 수 있는 범위는 클러스터의 로그 집계 설정과 권한에 따른다. Livy 로그 API, Spark YARN 디버깅

Statement 취소는 /sessions/{sessionId}/statements/{statementId}/cancel로 요청하고, 전체 세션 종료는 DELETE /sessions/{sessionId}로 요청한다. 취소가 이미 완료한 외부 데이터 쓰기를 되돌려주지는 않는다. Livy 취소·종료 API

Livy 서버 재시작 뒤 세션 복구도 자동 보장으로 취급하지 않는다. Apache Livy에는 recovery mode와 상태 저장소 설정이 있지만, 이는 실행 중인 세션에 다시 연결하기 위한 기능이다. Driver가 종료된 뒤 Python 변수와 메모리 상태를 복원하는 체크포인트 기능은 아니다. Livy recovery 설정

EMR Serverless에서는 무엇이 다른가

EMR Serverless Livy 엔드포인트는 관리형 HTTPS 엔드포인트다. 위의 SSH 터널과 일반 HTTP 요청 예제를 그대로 적용하지 않는다.

• 요청에 SigV4 서명이 필요하다.
• 세션 생성 시 conf에 emr-serverless.session.executionRoleArn을 전달한다.
• 엔드포인트 접근 권한과 실행 역할을 전달할 iam:PassRole 권한을 확인한다.
• 세션 timeout과 Serverless 애플리케이션의 auto-stop은 별도 설정이다.

AWS 문서에 따르면 Serverless 세션의 기본 비활성 제한은 1시간이며, emr-7.8.0부터 생성 요청의 ttl로 변경할 수 있다. EC2의 livy-conf 설정 방식과 구분해야 한다. EMR Serverless Livy 엔드포인트

결론

Livy 세션을 운영할 때는 코드 실행 결과와 실행 환경의 수명을 함께 관리해야 한다. 탐색 중에는 세션을 재사용하되, 사용이 끝나면 명시적으로 종료하고 유휴 세션이 쌓이지 않도록 유지 시간을 정한다.

클라이언트는 준비 상태를 기다리고 statement의 실제 실행 결과를 검사해야 한다. 세션 생성이나 데이터 쓰기 요청의 재시도는 중복 실행 가능성을 고려하고, 장애 추적에는 Livy와 YARN의 ID를 함께 남긴다.

관련 글

• Spark on EMR의 ApplicationMaster — Driver 관계, 메모리, 대기와 재시도
• EMR YARN Node Label Expression — CORE·TASK와 관리 모드
• YARN Capacity Scheduler 실전 가이드