Apache Spark

https://aws.amazon.com/ko/what-is/apache-spark/ 포스트를 참고해서 작성했습니다. 

 

 

1. 왜 사용하는가?

 

- Apache Spark 는 빅데이터를 빠르게 처리하기 위해 사용하는 분산 처리 엔진이다. 등장 배경을 이해하기 위해선 Hadoop MapReduce를 알아야 한다(이하 HM). HM은 병렬 분산 알고리즘을 사용해서 빅 데이터 세트를 처리하기 위한 프로그래밍 모델인데, 매번 디스크 I/O가 일어나기 때문에 속도가 느리다는 단점이 있었다. Spark는 인 메모리 처리를 수행하고, 작업의 단계 수를 줄이고, 여러 병렬 작업에서 데이터를 재사용하여 MapReduce의 한계를 해결하기 위해 만들어졌다.

 

- Spark는 데이터를 메모리에 올려놓고(Cache) 여러 연산에서 재사용할 수 있도록 하는 구조를 사용한다. -> 여러번 반복해서 계산하는 작업에 유리하다. 이를테면 머신러닝, 실시간 분석 같은 작업이다. 

 

- Spark는 하나의 프레임워크 안에서 배치 처리, SQL, 스트리밍 처리, 머신러닝, 그래프 분석까지 지원한다. 하나의 기술 스택 위에서 다양한 작업을 할 수 있다는 것이 장점이다 .

 

- Java, Scala, Python, R 등과 호환 가능한 범용성 높은 API가 존재한다. 

 

정리 

1. 메모리 기반 동작 - 빠르다

2. 한 프레임워크 내에서 다양한 분석이 가능 -> 아키텍처 단순화

3. API가 잘 되어있어 개발 생산성이 높다. 

 

2. 어떻게 동작하는가 

 

Apache Spark는 클러스터 환경에서 분산 계산을 수행하는 엔진으로 동작한다. 사용자는 로컬 코드에서 Spark API를 이용하여 연산을 정의하고, Spark는 이 연산을 클러스터 전체에 배포해서 병렬로 실행한다.

 

다음 단계로 동작을 이해할 수 있다. 

 

2-1. Application 정의 

- 개발자는 Python, Scala, Java, R 중 하나의 언어로 Spark 애플리케이션을 작성한다. 이 애플리케이션은 데이터를 어떻게 불러오고, 어떤 변환을 적용하고, 어떤 결과를 낼지에 대한 논리적인 파이프라인을 포함한다.

 

2-2 드라이버 프로세스 

- 애플리케이션을 실행하면 드라이버 프로그램이 올라가고, 드라이버는 전체 작업의 계획을 세운다. 드라이버는 사용자가 정의한 연산들을 모아서 DAG(Directed Acyclic Graph) 형태의 논리적 실행 계획을 만든다. 이 과정에서 연산 간 의존 관계와 병렬 실행 가능한 부분이 분석된다.

 

2-3 클러스터 매니저와 실행기 

- Spark는 YARN, Mesos, Kubernetes, 또는 자체 스탠드얼론 클러스터 매니저 위에서 실행될 수 있다. 클러스터 매니저는 노드들에 리소스를 할당하고, Spark는 각 노드에서 Executor 프로세스를 띄워서 실제 연산을 수행한다.

 

2-4 RDD / DataFrame / Dataset과 Lazy Evaluation

- Spark는 내부적으로 RDD(Resilient Distributed Dataset) 또는 그 위에 추상화된 DataFrame과 같은 구조를 사용해서 데이터를 표현한다. 사용자가 map, filter, groupBy 같은 변환 연산을 호출해도, Spark는 즉시 실행하지 않고 게으른 평가(lazy evaluation) 방식으로 연산 그래프만 쌓아 둔다. 최종적으로 count, show, write 같은 액션 연산이 호출될 때 비로소 전체 DAG를 최적화하여 실제 작업을 실행한다.

 

2-5 단계(Stage)와 작업(Task) 분할

- Spark는 전체 DAG를 Stage와 Task 단위로 쪼갠다. 셔플(shuffle)이 필요 없는 연속된 변환들은 하나의 Stage로 묶이고, 셔플 경계에서 Stage가 끊어진다. 각 Stage는 여러 개의 Task로 나뉘어 파티션 단위로 클러스터의 Executor에 병렬로 배포된다.

 

2-6 메모리 기반 처리와 캐시

- Spark는 데이터를 메모리에 유지할 수 있기 때문에, 같은 데이터를 여러 번 사용하는 알고리즘에서 큰 이점을 가진다. 개발자는 cache() 또는 persist()를 호출해서 중간 결과를 메모리에 저장하고, 이후 연산에서 재사용할 수 있다. AWS 문서에서도 Spark가 인메모리 처리를 통해 MapReduce의 다단계 디스크 I/O 문제를 해결했다고 설명한다.

 

요약 : Spark는 사용자가 정의한 연산을 DAG로 분석하고, 이를 Stage와 Task로 나누어 Executor에 병렬로 배포하며, 메모리 기반 처리와 캐시를 활용해 빠르게 계산을 수행하는 방식으로 동작한다. 

 

3. 장점

 

3-1 속도 : 인메모리 캐시, 최적화된 쿼리 실행 엔진으로 인해 낮은 지연 시간을 가진다. 

 

3-2 개발자 친화적 : 고수준 API를 다양한 언어 환경에서 제공한다. -> 분산 처리를 직접 구현할 필요 없이 알아서 다 해준다. 

 

3-3 다양한 워크로드 : Batch 처리, 실시간 스트리밍 분석, 머신러닝, 그래프 처리까지 모두 지원한다. 별도의 시스템을 여러 개 운영하는 부담을 줄일 수 있다. 

 

3-4 연동성 : DFS, Amazon S3, Amazon Redshift, Cassandra, MongoDB 등 다양한 외부 스토리지 시스템과 연동하여 분석을 수행할 수 있다. 

 

4. 세부 아키텍처 - Spark Core, Spark SQL, Spark Streaming, MLlib, GraphX

 

4-1 Spark Core : Spark 플랫폼의 기초이다. 메모리 관리, 장애 복구, 스케줄링, 작업 배포 및 모니터링, storage 시스템과의 상호작용을 담당한다. 개발자는 고수준 API를 통해 Spark Core와 상호작용한다.

 

4-2 Spark SQL :  Spark SQL은 분산 SQL 쿼리 엔진으로, 구조화된 데이터에 대한 SQL 기반 질의를 수행할 수 있게 해준다. 이 모듈은 비용 기반 최적화, 컬럼 지향 스토리지, 코드 생성 등의 기술을 활용해서 매우 빠른 대화형 쿼리를 제공하며, 표준 SQL과 HiveQL을 사용할 수 있게 한다. 다양한 데이터 소스(HDFS, JSON, Parquet, Hive, 외부 DB 등)와 연동된다.

 

4-3 Spark Streaming : Spark Streaming은 실시간 데이터 스트림을 처리하기 위한 모듈이다. 전통적인 Spark Streaming은 입력 스트림을 작은 배치 단위(micro-batch)로 나누어 처리하는 방식으로 동작하며, 동일한 API로 배치 처리 코드와 스트리밍 코드를 함께 재사용할 수 있게 해준다. 이는 Twitter, Kafka, Flume, HDFS 등 다양한 소스에서 데이터를 받아 실시간 분석을 수행할 수 있도록 한다.

 

4-4 MLlib : MLlib는 Spark 위에서 구동되는 머신러닝 라이브러리이다. 이 라이브러리는 분류, 회귀, 클러스터링, 협업 필터링, 패턴 마이닝 등 다양한 알고리즘을 포함하며, 대규모 데이터셋에 대한 머신러닝을 효율적으로 수행하도록 설계되어 있다. MLlib는 R이나 Python 등으로 학습한 모델을 저장하고, Java나 Scala 기반 파이프라인에서 이를 로드하여 사용할 수 있는 구조를 제공한다.

 

4-5 GraphX : GraphX는 대규모 그래프 데이터를 처리하기 위한 모듈이다. 이 모듈은 ETL, 탐색적 분석, 반복적 그래프 계산을 지원하며, 그래프 알고리즘(페이지랭크, 연결 요소 탐색 등)을 대규모 분산 환경에서 실행할 수 있게 한다.

 

이러한 모듈들이 모두 Spark Core 위에 올라가 있기 때문에, 하나의 클러스터와 하나의 API 세트로 다양한 분석 요구 사항을 충족할 수 있다. 

 

 

5. Python, PySpark 예시 

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg

# 1. SparkSession 생성
spark = SparkSession.builder \
    .appName("SimpleSparkExample") \
    .master("local[*]") \
    .getOrCreate()

# 2. 예제용 데이터 생성 (파이썬 리스트 → DataFrame)
data = [
    ("user1", "A", 100),
    ("user2", "A", 200),
    ("user3", "B", 300),
    ("user4", "B", 150),
    ("user5", "C", 400),
]

columns = ["user_id", "group", "amount"]

df = spark.createDataFrame(data, schema=columns)

# 3. DataFrame 내용 확인
print("=== 원본 데이터 ===")
df.show()

# 4. 그룹별 평균 amount 계산
grouped_df = df.groupBy("group").agg(
    avg(col("amount")).alias("avg_amount")
)

print("=== 그룹별 평균 amount ===")
grouped_df.show()

# 5. 결과를 파일로 저장 (예: Parquet)
# 실제 환경에서는 HDFS, S3 등으로 저장 가능
grouped_df.write.mode("overwrite").parquet("./output/group_avg_amount")

# 6. SparkSession 종료
spark.stop()

 

6. 더 쉬운 예시 

" 신용카드 거래 데이터를 하루 단위로 분석하는 배치 작업 " 

은행이나 카드사에는 하루에도 수천만 ~ 수억 건의 거래 로그가 발생한다.

예를 들어 다음과 같은 데이터가 S3, HDFS, DW 등 저장소에 쌓인다:

 

transaction_id card_id amount merchant timestamp

T10001	C123	35000	CU편의점	2025-11-20 10:01
T10002	C555	120000	배달의민족	2025-11-20 10:02
…	…	…	…	…

 

Spark는 S3, HDFS, DW, MySQL 등 어디서든 데이터를 읽어올 수 있다.

 

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("DailyBatchJob").getOrCreate()

# HDFS 또는 S3에 저장된 거래 로그 읽기
df = spark.read.parquet("s3://my-data/card_transactions/2025-11-20/")

 

데이터를 정재하는 것도 가능하다 ! 

 

df_clean = df.filter(df.amount > 0).dropna()

 

고객별 일일 총 결제 금액 집계를 한다고 가정하자.  다음과 같은 코드를 실행할 것이다. 

 

daily_sum = (
    df_clean.groupBy("card_id")
            .sum("amount")
            .withColumnRenamed("sum(amount)", "daily_total")
)

 

이런 작업은 Shuffle을 발생시키는 대표적인 작업이다. Spark에서는 이를 내부적으로 어떻게 처리하냐면, 

1. 각 노드에 분산된 데이터를 key 기준으로 정렬하고 병합한다

2. 하나의 노드에서 카드별 합계를 계산한다. 

3. 결과를 다시 분산 저장한다. 

 

 

만약 이상 거래 집계를 하고자 한다면, 

fraud_df = df_clean.filter(df_clean.amount > 1_000_000)

 

이런 방식으로 할 수 있다. 

 

가맹점별 매출 집계를 하고자 한다면 

merchant_stats = (
    df_clean.groupBy("merchant")
            .agg(
                F.count("*").alias("count"),
                F.sum("amount").alias("total_sales"),
                F.avg("amount").alias("avg_sales")
            )
)

 

이런 식으로 할 수 있다. SQL을 사용하는 것과 별 다를 바 없다. 내부 처리가 엄청나게 효율적일 뿐이다. 

 

 

결과를 Warehouse 또는 S3에 저장하기 위해서 다음과 같은 코드를 사용할 수 있다. 

daily_sum.write.mode("overwrite").parquet("s3://reports/daily_sum/2025-11-20/")
merchant_stats.write.mode("overwrite").parquet("s3://reports/merchant_stats/2025-11-20/")

 

Airflow / Cron으로 배치 자동 실행 파이프라인을 만들 수 있다. 

 

 

7. 클러스터와 노드 

- Spark 에서 노드는 크게 세 종류가 있다. 

 

7-1 Driver Node : 메인 컨트롤 노드이다. Job을 여러 Task로 나누고, Worker Node에게 일을 분배한다. 

7-2 Worker Node : 실제로 데이터를 계산, 가공하는 노드이다. CPU + Memory를 사용해 Task를 실행한다. 

7-3 Cluster Manager : 클러스터 전체의 자원을 관리하고, Driver/Worker의 실행을 조율한다. 

 

Step 1 - Driver 시작

- 개발자가 “spark-submit”으로 Spark 프로그램 실행

- Driver 노드가 실행됨

 

Step 2 - 클러스터 매니저에게 자원 요청

- Driver는 Worker 노드 몇 개가 필요한지 요청

ex) Worker 4개 할당받음

 

Step 3 - Job을 Task로 쪼갬

- 10GB 데이터를 256MB씩 쪼개면 약 40개 블록

- Driver는 각 블록을 하나의 Task로 분리

- Task 40개 생성

 

Step 4 - Worker에게 작업 분배

- Worker Node 1: Task 1~10

- Worker Node 2: Task 11~20

- Worker Node 3: Task 21~30

- Worker Node 4: Task 31~40

→ 병렬 처리 시작

 

Step 5 - Worker가 데이터 연산 실행

- 필터, 그룹바이, 조인 등 DataFrame 연산 수행

- CPU 코어를 모두 사용하며 parallel execution

 

Step 6 - 결과 수집

- Worker → Driver로 결과 전송

- Driver가 최종 결과를 파일로 저장하거나 반환

 

여기서 클러스터는 여러 대의 컴퓨터를 묶어서 하나처럼 쓰는 시스템인 것이다. 그리고 노드는 클러스터를 구성하는 개별 컴퓨터이다. 노드는 물리적 서버, 가상 머신 등이 될 수 있다. 

 

8. 쿠버네티스와 Spark 

 

- 쿠버네티스는 어떻게 클러스터가 되는가 : 쿠버네티스는 여러 서버를 묶어서 하나의 거대한 컴퓨팅 시스템처럼 작동한다. 

 

Spark에서는 다음과 같이 클러스터 종류를 자체적으로 선택할 수 있는데, 

 

1. Standalone Cluster (Spark 자체 클러스터)

2. YARN Cluster (하둡 생태계)

3. Kubernetes Cluster 

 

 

쿠버네티스도 그 중 하나이다. 

 

쿠버네티스는 컨테이너 기반이기 때문에 클러스터 환경을 구성하기 쉽다는 장점이 있다. 

 

from pyspark.sql import SparkSession
from pyspark.sql.functions import col

spark = SparkSession.builder.appName("K8sSparkExample").getOrCreate()

# 예: S3 또는 HDFS에 저장된 거래 로그
df = spark.read.parquet("s3://bucket/transactions/2024/")

# 예: 2024년 1월 거래 합계 계산
result = df.groupBy("customer_id").sum("amount")

# 다시 S3에 저장
result.write.mode("overwrite").parquet("s3://bucket/processed/2024/")

spark.stop()


// 클러스터 실행 명령 

spark-submit \
  --master k8s://https://k8s-mycluster:6443 \
  --deploy-mode cluster \
  --name spark-k8s-job \
  --conf spark.executor.instances=4 \
  --conf spark.kubernetes.container.image=my-spark:latest \
  local:///app/job.py