Apache Spark

Apache Spark 는 서버 여러 대에 흩어진 큰 데이터를 나눠 맡겨 한꺼번에 계산해 주는 엔진입니다. 계산 중간 결과를 디스크에 내려놓지 않고 메모리에 둔 채 다음 계산으로 넘깁니다. 그래서 같은 데이터를 여러 번 훑는 계산이 빨라집니다. 데이터를 오래 담아 두는 일은 하지 않고 다른 저장소에서 읽어 옵니다.

상세

Apache Spark 는 클러스터에서 도는 분산 처리 엔진입니다. 클러스터는 한 가지 일을 함께 하도록 묶은 서버 무리입니다. Spark 는 그 서버들에 계산을 나눠 맡기고 결과를 모아 줍니다. Apache Software Foundation 이 관리하는 오픈 소스 프로젝트입니다.

이 절은 Spark 가 앞 세대 계산 도구와 무엇이 다른지부터 봅니다. 그다음 Spark 가 데이터를 어떤 모양으로 들고 다니는지, 적어 둔 계산이 어떤 순서로 서버에 퍼지는지, 서버가 죽으면 어떻게 되살리는지를 봅니다. 마지막으로 Spark 가 내려놓은 것을 봅니다.

단계마다 디스크를 오가던 앞 세대

Hadoop 은 서버 수백 대에 데이터를 나눠 두고 그 위에서 계산까지 돌리는 오픈 소스 프로젝트 묶음입니다. Hadoop 의 계산 도구는 MapReduce 입니다. MapReduce 는 큰 데이터를 조각마다 따로 계산한 뒤 결과를 모으는 계산 방식입니다.

MapReduce 작업 하나는 결과를 HDFS(Hadoop Distributed File System, 하둡 분산 파일 시스템)에 쓰고 끝납니다. HDFS 는 서버 여러 대의 디스크를 묶은 파일 시스템입니다. 큰 파일은 일정한 크기의 조각으로 잘라 여러 서버에 나눠 담습니다. 이 조각이 HDFS 의 블록입니다.

계산이 여러 단계면 다음 작업이 그 파일을 디스크에서 다시 읽어 시작합니다.

이 방식이 특히 느린 계산이 둘 있습니다. 하나는 머신러닝의 학습처럼 같은 데이터를 수십 번 훑으며 값을 조금씩 고치는 반복 계산입니다. 다른 하나는 사람이 조회를 던지고 결과를 기다리는 대화형 분석입니다. 둘 다 매번 디스크를 오가느라 계산보다 읽고 쓰는 데 시간을 더 씁니다.

Spark 는 여러 단계를 한 작업으로 묶습니다. 단계 사이의 중간 결과는 메모리에 둔 채 다음 단계로 넘깁니다. 디스크에는 처음 읽을 때와 마지막에 쓸 때만 갑니다.

아래 그림은 계산 두 단계를 두 방식으로 돌린 모습입니다. 위가 MapReduce, 아래가 Spark 입니다.

flowchart TD
    subgraph MR["MapReduce · 작업 둘"]
        M1["HDFS 에서 읽는다"] --> M2["계산 1"]
        M2 --> M3["HDFS 에 쓴다"]
        M3 --> M4["HDFS 에서 다시 읽는다"]
        M4 --> M5["계산 2"]
        M5 --> M6["HDFS 에 쓴다"]
    end
    subgraph SP["Spark · 작업 하나"]
        S1["HDFS 에서 읽는다"] --> S2["계산 1"]
        S2 -->|메모리에 둔 채 넘긴다| S3["계산 2"]
        S3 --> S4["HDFS 에 쓴다"]
    end
    M6 ~~~ S1

RDD

Spark 가 데이터를 다루는 기본 단위는 RDD(Resilient Distributed Dataset, 탄력적 분산 데이터셋)입니다. RDD 는 여러 서버에 나뉘어 담긴 큰 컬렉션 하나입니다. 10억 줄짜리 로그 파일을 읽으면 줄 10억 개를 담은 RDD 하나가 됩니다.

RDD 는 파티션 여러 개로 쪼개져 있습니다. 파티션은 데이터 조각 하나입니다. 서버 한 대가 한 번에 맡는 계산 단위이기도 합니다. HDFS 파일을 읽으면 앞에서 본 HDFS 블록 하나가 보통 파티션 하나가 됩니다.

RDD 에는 불변성이 있습니다. 한번 만들면 내용을 바꿀 수 없다는 뜻입니다. 값을 바꾸고 싶으면 옛 RDD 에서 새 RDD 를 하나 더 만듭니다. 이 성질이 뒤에 나오는 장애 복구를 쉽게 만듭니다.

변환과 액션

RDD 에 거는 연산은 두 가지입니다. 변환(transformation)은 RDD 에서 새 RDD 를 만드는 연산입니다. 줄을 골라내는 filter, 줄마다 값을 바꾸는 map 이 변환입니다.

액션(action)은 계산 결과를 요구하는 연산입니다. 개수를 세는 count, 결과를 파일로 쓰는 saveAsTextFile 이 액션입니다. 액션을 부르면 그때 계산이 돕니다.

변환은 부르는 순간 계산하지 않습니다. 무엇을 할지 적어만 둡니다. 이렇게 결과가 필요해질 때까지 계산을 미루는 방식이 지연 평가입니다.

아래는 로그에서 에러 줄만 골라 날짜별로 세는 파이썬 코드입니다. sc 는 프로그램이 Spark 에 붙을 때 받는 연결 객체입니다. 줄마다 첫 칸이 날짜라고 가정합니다.

lines  = sc.textFile("hdfs:///logs/app.log")
errors = lines.filter(lambda l: "ERROR" in l)          # 변환: 아직 안 돈다
pairs  = errors.map(lambda l: (l.split()[0], 1))       # 변환: (날짜, 1)
counts = pairs.reduceByKey(lambda a, b: a + b)         # 변환: 날짜별로 더한다
counts.saveAsTextFile("hdfs:///out/error-counts")      # 액션: 여기서 돈다

네 줄의 변환은 적어 둔 계획일 뿐입니다. 마지막 줄의 액션이 불리고 나서야 파일을 읽기 시작합니다.

계산을 미루면 Spark 가 계획 전체를 보고 나서 돌릴 수 있습니다. filter 와 map 은 한 줄을 읽을 때마다 연달아 적용합니다. 에러 줄만 모은 중간 RDD 를 따로 만들어 두지 않습니다.

드라이버와 익스큐터

Spark 프로그램 하나는 프로세스 두 종류로 나뉘어 돕니다. 드라이버는 사용자 프로그램이 도는 프로세스입니다. 적어 둔 계획을 쪼개 어느 서버가 무엇을 할지 정합니다.

익스큐터는 계산을 맡은 서버마다 뜨는 프로세스입니다. 이런 서버를 워커 서버라고 부릅니다. 익스큐터는 드라이버가 보낸 일을 받아 파티션을 계산합니다. 계산한 파티션도 익스큐터가 자기 메모리에 쥡니다.

익스큐터를 어느 서버에 몇 개 띄울지는 클러스터 관리자가 정합니다. 클러스터 관리자는 서버들의 프로세서 코어와 메모리를 여러 프로그램에 나눠 주는 관리자입니다. Spark 에 딸린 자체 관리자를 써도 됩니다. YARN(Yet Another Resource Negotiator)이나 Kubernetes 를 써도 됩니다.

flowchart TD
    D["드라이버 · 계획을 쪼갠다"]
    CM["클러스터 관리자"]
    subgraph W1["워커 서버 1"]
        E1["익스큐터"]
    end
    subgraph W2["워커 서버 2"]
        E2["익스큐터"]
    end
    D -->|익스큐터를 청한다| CM
    CM -->|띄운다| E1
    CM -->|띄운다| E2
    D -->|할 일을 보낸다| E1
    D -->|할 일을 보낸다| E2

드라이버는 클러스터 관리자에게 익스큐터를 청합니다. 익스큐터가 뜨면 드라이버가 익스큐터와 직접 주고받습니다. 클러스터 관리자는 자원을 나눠 준 뒤 계산에는 끼지 않습니다.

스테이지와 셔플

액션이 불리면 드라이버는 적어 둔 변환을 그래프 하나로 그립니다. RDD 가 점이고 변환이 화살표입니다. 화살표가 한 방향으로만 가고 되돌아오는 길이 없는 이런 그래프가 DAG(Directed Acyclic Graph, 방향 비순환 그래프)입니다.

드라이버는 이 그래프를 스테이지 여러 개로 자릅니다. 스테이지는 파티션마다 멈추지 않고 이어서 돌릴 수 있는 변환의 묶음입니다. 자르는 기준은 파티션끼리 데이터를 주고받아야 하느냐입니다.

filter 와 map 은 파티션 하나가 자기 안에서 끝납니다. 줄 하나를 보는 데 다른 파티션이 필요 없습니다. 이런 변환은 한 스테이지 안에 이어 붙습니다.

reduceByKey 는 다릅니다. 같은 날짜의 줄이 여러 파티션에 흩어져 있어서 먼저 한곳으로 모아야 더할 수 있습니다. 파티션 사이에서 데이터를 다시 나눠 옮기는 이 일이 셔플입니다. 셔플 앞뒤에서 스테이지가 갈립니다.

flowchart TD
    subgraph S1["스테이지 1 · 파티션마다 따로 돈다"]
        A["textFile · 읽는다"] --> B["filter · 에러 줄만"]
        B --> C["map · (날짜, 1)"]
    end
    subgraph S2["스테이지 2"]
        R["reduceByKey · 날짜별로 더한다"] --> O["saveAsTextFile · 쓴다"]
    end
    C -->|셔플 · 같은 날짜를 한 파티션으로| R

스테이지 하나를 파티션 하나에 대해 돌리는 일이 태스크입니다. 파티션이 100개면 스테이지 1은 태스크 100개가 됩니다. 드라이버는 태스크를 익스큐터들에 나눠 보냅니다.

셔플은 Spark 에서 가장 비싼 단계입니다. 보내는 쪽 익스큐터는 넘길 데이터를 자기 서버의 로컬 디스크에 먼저 씁니다. 받는 쪽은 그것을 네트워크로 가져옵니다. 셔플 없이 끝나는 계산일수록 빠릅니다.

셔플 뒤에는 한 키가 한 파티션에 모입니다. 어떤 날짜에만 에러가 몰려 있으면 그 파티션 하나만 크게 부풉니다. 다른 태스크는 다 끝났는데 태스크 하나만 오래 도는 이 현상을 데이터 스큐라고 부릅니다.

캐시

액션을 두 번 부르면 Spark 는 기본으로 계산을 처음부터 두 번 돌립니다. 액션마다 파일 읽기부터 다시 합니다.

여러 번 쓸 RDD 는 cache() 를 불러 익스큐터 메모리에 남겨 둘 수 있습니다. 처음 계산할 때 파티션을 메모리에 담아 둡니다. 다음 액션부터는 거기서 바로 읽습니다. 머신러닝 학습처럼 같은 데이터를 수십 번 훑는 계산이 이것으로 빨라집니다.

메모리에 다 안 들어가면 못 담은 파티션은 필요할 때 다시 계산합니다. 메모리 대신 디스크에 담아 두도록 고를 수도 있습니다.

계보로 되살리기

서버 수백 대를 쓰면 어딘가의 익스큐터는 계산 도중에 죽습니다. 그 익스큐터가 쥐고 있던 파티션도 같이 사라집니다.

Spark 는 파티션을 복사해 두지 않습니다. 대신 RDD 마다 자기가 어느 RDD 에서 어떤 변환으로 만들어졌는지를 적어 둡니다. 이 기록이 계보(lineage)입니다. 앞에서 본 코드라면 counts 의 계보는 파일 읽기 → filter → map → reduceByKey 입니다.

파티션을 잃으면 Spark 는 계보를 따라 그 파티션만 다시 계산합니다. RDD 가 불변이라 같은 입력에 같은 변환을 걸면 같은 결과가 나옵니다. 그래서 다시 계산한 파티션이 잃은 것과 같습니다.

아래 그림은 셔플 앞 스테이지에서 pairs 의 파티션 2 를 잃은 경우입니다. Spark 는 그 파티션의 계보만 거슬러 올라가 HDFS 블록 2 부터 filter 와 map 을 다시 겁니다. 파티션 1 쪽은 건드리지 않습니다. 그림은 계보를 RDD 마다 한 층씩 나눠 그렸습니다. 실제 계산에서는 앞에서 본 대로 filter 와 map 을 한 줄마다 연달아 적용하므로, errors 가 따로 만들어져 쌓이지는 않습니다.

flowchart TD
    subgraph F["HDFS 파일"]
        B1["블록 1"]
        B2["블록 2"]
    end
    subgraph L["lines"]
        L1["파티션 1"]
        L2["파티션 2"]
    end
    subgraph E["errors"]
        E1["파티션 1"]
        E2["파티션 2"]
    end
    subgraph P["pairs"]
        P1["파티션 1 · 남아 있다"]
        P2["파티션 2 · 잃었다"]
    end
    B1 --> L1
    L1 -->|filter| E1
    E1 -->|map| P1
    B2 ==>|다시 읽는다| L2
    L2 ==>|filter 를 다시 건다| E2
    E2 ==>|map 을 다시 건다| P2

복사본을 두지 않으니 메모리를 한 벌만 씁니다. 계보가 길면 이 방식이 느려집니다. 변환을 수백 번 이은 RDD 는 되살리는 데도 그만큼 오래 걸립니다.

그럴 때는 RDD 체크포인트를 씁니다. 중간 RDD 를 HDFS 같은 저장소에 파일로 써 둡니다. 되살릴 때는 계보를 처음부터 따라가지 않고 그 파일에서 시작합니다. 계보가 거기서 끊기는 셈입니다.

DataFrame 과 Spark SQL

RDD 는 Spark 가 속을 모르는 객체 묶음입니다. 줄 하나가 문자열인지 여러 칸짜리 기록인지 Spark 는 알지 못합니다. 그래서 사용자가 적은 순서대로만 돌립니다.

DataFrame 은 열 이름과 타입이 있는 표 모양 데이터입니다. 로그라면 date · level · message 열을 가진 표가 됩니다. Spark 가 열을 알기 때문에 계산 순서를 스스로 고쳐 짤 수 있습니다.

Spark SQL 은 DataFrame 을 SQL(Structured Query Language)로 조회하게 해 주는 부품입니다. 같은 계산을 코드로 적든 SQL 로 적든 같은 실행 계획이 됩니다. 아래 둘은 같은 일을 합니다. spark 는 DataFrame 을 쓸 때 받는 연결 객체입니다.

df = spark.read.parquet("hdfs:///logs/parquet/")
df.filter(df.level == "ERROR").groupBy("date").count().show()

df.createOrReplaceTempView("logs")
spark.sql("SELECT date, count(*) FROM logs WHERE level = 'ERROR' GROUP BY date").show()

실행 계획을 고쳐 짜는 부품이 Catalyst 옵티마이저입니다. 위 조회라면 level 과 date 열만 필요하다는 것을 압니다.

그 지식은 파일을 읽을 때 쓰입니다. Apache Parquet 는 열마다 따로 모아 담는 파일 형식이라 Spark 가 두 열만 골라 읽습니다. message 열은 디스크에서 읽지도 않습니다.

그래서 요즘 Spark 코드는 대개 DataFrame 으로 씁니다. RDD 는 표 모양으로 담기 어려운 데이터를 다룰 때 씁니다.

한 엔진 위의 라이브러리

Spark 는 같은 엔진 위에 여러 일을 얹었습니다. 배치 계산, 흘러 들어오는 데이터 계산, 머신러닝이 같은 드라이버와 익스큐터 위에서 돕니다.

Structured Streaming 은 기본으로 마이크로 배치 방식으로 돕니다. 들어오는 데이터를 몇 초 단위 작은 묶음으로 끊습니다. 묶음마다 배치 계산을 한 번 돌립니다. 그래서 배치 코드와 거의 같은 코드로 스트림 처리를 합니다.

사용자 코드를 어느 언어로 쓰든 도는 엔진은 이 하나입니다. Spark 자신은 Scala 로 쓰였고 JVM(Java Virtual Machine, 자바 가상 머신) 위에서 돕니다. 사용자 코드는 Scala · Java · 파이썬 · R 로 씁니다. 파이썬으로 쓸 때 붙는 쪽이 PySpark 입니다.

PySpark 에서 DataFrame 연산은 JVM 안에서 돕니다. 파이썬 코드는 계획만 넘깁니다. 다만 사용자가 직접 짠 파이썬 함수를 줄마다 부르면 데이터를 JVM 과 파이썬 프로세스 사이로 옮겨야 합니다. 옮길 때마다 직렬화를 거칩니다. 그래서 같은 일을 DataFrame 연산으로 적을 때보다 느려집니다.

Spark 가 내려놓은 것

Spark 는 큰 데이터를 여러 단계에 걸쳐 계산하는 일에 맞춰 몇 가지를 포기했습니다.

첫 줄은 Spark 가 데이터베이스가 아니라는 뜻입니다. 데이터를 담아 두고 한 줄씩 고쳐 쓰는 일은 하지 않습니다. 읽고 계산하고 결과를 다른 저장소에 씁니다.

둘째 줄의 대가는 운영에서 드러납니다. 익스큐터가 큰 메모리를 쥐면 가비지 컬렉션이 오래 멈춥니다. 셔플이나 캐시가 메모리를 넘치면 익스큐터가 메모리 부족으로 죽습니다.

넷째 줄은 Apache Flink 와 견주면 보입니다. Flink 는 들어오는 기록을 한 건씩 바로 흘려 처리하는 엔진입니다. Spark 의 마이크로 배치는 묶음 하나를 모으는 만큼 결과가 늦게 나옵니다.

Spark 를 고르는 경우

한 서버에 안 들어가는 데이터를 여러 단계에 걸쳐 가공하는 일에 맞습니다. 밤마다 로그를 모아 정리하고 집계해 데이터 웨어하우스에 넣는 배치 처리가 그런 일입니다. 같은 데이터를 여러 번 훑는 머신러닝 학습도 맞습니다.

데이터가 서버 한 대의 메모리에 들어가면 Spark 가 오히려 느립니다. 익스큐터를 띄우고 태스크를 나눠 보내는 준비가 계산보다 오래 걸립니다. 그럴 때는 한 서버에서 도는 pandas 나 데이터베이스가 맞습니다.

주문 한 건을 곧바로 읽고 고치는 서비스 데이터베이스 노릇은 못 합니다. 기록 한 건마다 밀리초 안에 반응해야 하는 스트림 처리라면 Flink 같은 엔진이 맞습니다.

맞물림

Spark 는 Hadoop 생태계 곁에서 자랐습니다. 같은 Apache Software Foundation 의 도구들을 부르거나 그 위에 섭니다. 아래 넷은 누가 누구를 부르는지, 붙어서 무엇을 얻고 무엇을 치르는지로 나눠 적었습니다.

HDFS 에 묻는 블록 위치

Spark 가 HDFS 를 부릅니다. HDFS 에서 파일 목록과 블록 목록을 관리하는 서버가 네임노드입니다. Spark 드라이버는 파일을 읽기 전에 네임노드에 블록마다 어느 서버에 있는지 묻습니다.

드라이버는 그 블록을 가진 서버의 익스큐터에 태스크를 먼저 보냅니다. 데이터를 네트워크로 옮기는 대신 계산을 데이터 쪽으로 보내는 것입니다. 이것을 데이터 지역성이라고 부릅니다.

작은 파일이 많으면 이 방식이 짐이 됩니다. 파일 하나가 적어도 파티션 하나가 되므로 작은 파일이 수십만 개면 태스크도 수십만 개가 됩니다. 태스크마다 드는 준비 시간이 계산보다 길어집니다. 이것이 HDFS 의 작은 파일 문제가 Spark 에서 드러나는 모습입니다.

YARN 에 청하는 컨테이너

YARN 은 Hadoop 클러스터의 서버 자원을 여러 계산 작업에 나눠 주는 관리자입니다. Spark 를 YARN 위에서 돌리면 Spark 가 YARN 에 자원을 청합니다.

YARN 은 서버마다 프로세서 코어와 메모리를 떼어 컨테이너라는 몫으로 내줍니다. Spark 는 받은 컨테이너 안에 익스큐터를 띄웁니다. 그래서 한 Hadoop 클러스터 위에서 MapReduce 작업과 Spark 작업이 같은 서버들을 나눠 씁니다.

YARN 위에서 조심할 것은 메모리 한도입니다. YARN 은 컨테이너가 받은 몫보다 메모리를 더 쓰면 그 컨테이너를 죽입니다. 익스큐터는 JVM 이 관리하는 메모리 말고도 JVM 바깥에서 쓰는 메모리가 있습니다. 이 몫을 넉넉히 잡지 않으면 익스큐터가 계산 도중에 죽습니다.

Hive 메타스토어에서 찾는 테이블

Apache Hive 는 HDFS 의 파일을 SQL 로 조회하게 해 주는 도구입니다. Hive 테이블 하나는 HDFS 디렉터리 하나입니다. 테이블 이름과 열 목록, 디렉터리 경로의 짝은 Hive 메타스토어가 쥡니다.

Spark SQL 이 Hive 메타스토어를 부릅니다. 테이블 이름으로 조회를 받으면 메타스토어에서 경로와 열 목록을 찾습니다. 파일을 읽는 것도 계산도 Spark 가 합니다. 그래서 Hive 로 만든 테이블을 계산 엔진만 바꿔 Spark 로 조회할 수 있습니다.

약점은 테이블 정보가 메타스토어와 디렉터리 두 곳에 갈린다는 것입니다. Hive 테이블은 날짜 같은 열 값마다 하위 디렉터리를 나눠 담기도 합니다. Hive 에서는 이 하위 디렉터리를 파티션이라고 부릅니다. 앞에서 본 RDD 의 파티션과는 다른 것입니다.

날짜 하위 디렉터리를 새로 만들어 파일을 직접 넣으면 메타스토어는 그 디렉터리가 생긴 줄 모릅니다. 메타스토어를 따로 갱신해야 Spark 가 그 파일을 읽습니다.

Kafka 에서 읽어 가는 오프셋 범위

Apache Kafka 는 들어오는 메시지를 순서대로 쌓아 두고 여러 소비자가 읽어 가게 하는 메시지 저장소입니다. Kafka 는 메시지마다 순번을 매깁니다. 이 순번이 오프셋입니다.

Structured Streaming 이 Kafka 를 부릅니다. 마이크로 배치를 하나 돌 때마다 지난번에 읽은 오프셋 다음부터 지금까지 쌓인 메시지를 가져옵니다. Kafka 가 Spark 에 밀어 넣지 않고 Spark 가 가져갑니다.

어디까지 읽었는지는 Spark 가 자기 체크포인트 디렉터리에 적습니다. 이름은 같지만 「계보로 되살리기」의 RDD 체크포인트와는 다른 것입니다. 계보를 끊으려고 RDD 를 써 두는 곳이 아닙니다. 스트림을 어디까지 읽었는지 적어 두는 디렉터리입니다.

프로그램을 다시 켜면 거기서 이어 읽습니다. 체크포인트 디렉터리를 잃거나 지우면 Spark 는 어디서부터 읽을지 모르게 됩니다.

관련 항목

Spark 가 속하는 상위 분류

분산 처리 · 분산 시스템 · 빅데이터 · 오픈 소스 · Apache Software Foundation

Spark 를 이루는 구성 요소

RDD · DataFrame · Spark SQL · Structured Streaming · MLlib · GraphX · Catalyst 옵티마이저 · Spark 드라이버 · Spark 익스큐터 · PySpark

Spark 가 계산을 쪼개는 단위

DAG · 스테이지 · 태스크 · 파티션 · 셔플

Spark 가 기대는 계산 원리

지연 평가 · 불변성 · 리니지 · 체크포인트 · 인메모리 컴퓨팅 · 데이터 지역성 · 직렬화

Spark 가 자라난 Hadoop 생태계

Hadoop · HDFS · YARN · MapReduce · Apache Hive · Hive 메타스토어 · HBase

Spark 가 읽고 쓰는 저장소와 파일 형식

Amazon S3 · 오브젝트 스토리지 · 데이터 레이크 · Apache Kafka · Apache Parquet · 데이터 웨어하우스

Spark 에 서버 자원을 나눠 주는 클러스터 관리자

Kubernetes · Spark Standalone · 클러스터 · Apache Mesos

Spark 와 같은 일을 두고 겨루는 엔진

Apache Flink · Apache Beam · Trino · Dask · pandas

Spark 가 맡는 작업 방식

배치 처리 · 스트림 처리 · 마이크로 배치 · ETL · 머신러닝 · 처리량

Spark 를 쓰는 언어와 런타임

Scala · Java · Python · R · JVM

Spark 에서 자주 나는 운영 문제

데이터 스큐 · 메모리 부족 · 가비지 컬렉션 · 작은 파일 문제