Apache Spark : Spark 소개 및 구조

 

목차 LIST

 

 

 

스파크의 간략한 역사

스파크는 UC 버클리의 RAD 연구실(AMPLab)의 연구 프로젝트로 2009년에 시작됐다. 연구원들은 이전에 하둡 맵리듀스로 일한 경험이 있었는데, 맵리듀스가 반복적인 대화형 연산 작업에는 비효율적인 것을 발견했다. 그래서 스파크는 시작부터 인메모리 저장 장치 및 효과적인 장애 복구 지원 같은 개념에 기반하여 대화형 쿼리와 반복성 알고리즘에 빠르게 동작하도록 설계되었다.

 

스파크란

아파치 스파크는 가장 활발하게 개발되고 있는 병렬 처리 오픈소스 엔진이며, 표준 도구로 자리잡고 있다.

널리 쓰이는 네 가지 언어(파이썬, 자바, 스칼라, R)를 지원하며 SQL 뿐만 아니라 스트리밍, 머신러닝에 이르기까지 넓은 범위의 라이브러리를 제공한다. 스파크는 단일 노트북 환경에서부터 수천 대의 서버로 구성된 클러스터까지 다양한 환경에서 실행할 수 있다. 이런 특성으로 빅데이터 처리를 쉽게 시작할 수 있고 엄청난 규모의 클러스터로 확장해나갈 수 있다.

 

(도서) 스파크 완벽 가이드

 

 

아파치 스파크는 대규모 데이터 처리를 위한 빠르고 안정적이며 내결함성(fault tolerance)이 있는 분산 컴퓨팅 프레임워크이다.

 

컴퓨팅 엔진

스파크는 저장소 시스템의 데이터를 연산하는 역할만 수행할 뿐 영구 저장소의 역할은 수행하지 않는다. 그 대신 다양한 저장소(Azure Storage, Amazon S3, Apache Hadoop, Apache Cassandra, Apache kafka 등)를 지원한다. 

 

라이브러리

스파크는 엔진에서 제공하는 표준 라이브러리와 오픈소스 커뮤니티에서 서드파티 패키지로 제공하는 다양한 외부 라이브러리를 지원한다. 사실 스파크의 표준 라이브러리는 여러 오픈소스 프로젝트의 집합체이다. 스파크는 SQL과 구조화된 데이터를 제공하는 스파크 SQL, 머신러닝을 지원하는 MLlib, 스트림 처리 기능을 제공하는 스파크 스트리밍과 새롭게 선보인 구조적 스트리밍 그리고 그래프 분석 엔진인 GraphX 라이브러리를 제공한다. 기본 라이브러리 외에 다양한 저장소 시스템을 위한 커넥터부터 머신러닝을 위한 알고리즘까지 수백 개의 외부 오픈소스 라이브러리가 존재하며, 외부 라이브러리 목록은 https://spark-packages.org/ 에서 확인할 수 있다.

 

기능적 API

초기 스파크의 혁신은 획기적인 메모리 사용과 풍부한 기능적인 API에 의해 이루어졌다.

"기능적"은 함수형 프로그램의 특징을 의미하는데, 함수형 프로그래밍은 데이터의 불변성, 상태의 부수 효과 없이 계산을 수행하는 함수의 사용을 강조한다. 

Spark API는 이러한 함수형 프로그래밍 패러다임을 따르는며, 데이터 처리를 위한 고수준의 추상화를 제공한다. 이를 통해, 데이터셋에 map, filter, reduce와 같은 함수형 연산을 데이터셋에 적용할 수 있다. 해당 API를 사용함으로써, 복잡한 데이터 변환과 분석을 명확하고 간결한 코드로 구현할 수 있다.

 

특히, Spark의 탄력적 분석 데이터셋(RDD: Resilient Distributed Dataset)와 DataFrame API는 이러한 함수형 프로그래밍 개념을 활용한다. 이러한 API들을 통해 데이터에 대한 변환(transformations)과 액션(actions)을 정의할 수 있으며, spark는 이를 클러스터 전반에 걸쳐 효율적으로 실행한다. 이러한 방식으로, Spark는 메모리 내 계산을 최적화하고 대규모 데이터 처리 작업을 빠르고 효율적으로 수행할 수 있도록 한다.

 

 

SQL

스파크의 두 번째 전환점은 스파크 SQL 및 데이터프레임(이후에는 데이터셋으로 됨)의 도입이다. 스파크 SQL은 스키마가 있는 모든 데이터셋에 SQL을 지원한다. SQL 데이터베이스를 쿼리할 때와 같은 방식으로 CSV, Parquet 또는 JSON 데이터셋을 쿼리할 수 있다.

성능 관점에서 SparkSQL은 쿼리 최적화 프로그램과 물리적 실행 엔진을 스파크로 가져옴으로써 적은 리소스를 사용하는 동안 더 빠르게 실행될 수 있게 만들었다.

 

통합 엔진

스파크는 데이터 분석에 대한 polyglot(폴리그롯, 여러 프로그래밍 언어로 코드를 작성하는 관행)접근과 호환되는 배치와 스트리밍 기능을 제공하는 통합 분석 엔진으로 스칼라, 자바, 파이썬 및 R 언어 API를 제공한다.

 

스파크의 통합 프로그래밍 모델은 개발자가 배치 및 스트리밍 워크로드를 모두 해결하기 위해 단 하나의 새로운 패러다임을 배우면 된다는 것을 의미한다.

 

분산 스파크 구조

분산 모드 스파크는 하나의 Coordinator와 여러 개의 분산 작업 노드로 구성되는 마스터/슬레이브 구조를 사용한다.

Coordinator는 '드라이버' 라고도 부른다. 드라이버는 Executor라고 불리는 다수의 분산 작업자들과 통신한다.

 

드라이버는 개별 자바 프로세스로 실행되며, 각 익스큐터 또한 독립된 자바 프로세스로 동작한다. 이 드라이버와 여러 Executor들이 결합하여 스파크 애플리케이션을 형성한다.

 

클러스터 매니저

스파크는 한 노드(node)에서 수천 노드까지 효과적으로 확장할 수 있도록 설계되었다. 이를 달성하기 위해 유연성을 극대화하면서 스파크는 하둡의 YARN, 아파치 메소스, 스파크에서 지원하는 가벼운 구현 클러스터 매니저인 '단독 스케쥴러(Standalone Scheduler)'등 다양한 클러스터 매니저 위에서 동작할 수 있다.

YARN이나 메소스 클러스터가 설치되어 있다면 이 위에 스파크 애플리케이션을 실행시킬 수 있고, 아무것도 설치되지 않은 머신이라면 단독 스케줄러를 사용할 수 있다.

 

SparkContext

Spark의 핵심 구성 요소로, Spark 애플리케이션의 entry point 역할을 한다. 이 객체를 통해 Spark 클러스터로 연결되며, Spark 작업을 실행하기 위한 기본 설정과 리소스 관리, 작업 스케줄링, RDD 생성 및 작업 분배 등의 기능을 수행한다.

즉, SparkContext는 Spark 애플리케이션의 핵심 역할을 하며, 애플리케이션과 Spark 클러스터 간의 모든 통신을 관리한다.

 

SparkContext는 프로그래머가 Spark와의 모든 상호 작용을 시작하는 곳으로 애플리케이션 초기에서 SparkContext 객체를 생성하고, 이 객체를 사용하여 데이터를 불러오거나 변환하고 액션을 수행한다. 애플리케이션이 종료될 때는 SparkContext를 사용하여 클러스터와의 연결을 종료하고 리소스를 정리한다.

 

스파크 동작 과정

아래 그림은 스파크가 클러스터에서 어떻게 실행되는지를 나타낸다.

 

1. 스파크를 클러스터 환경에서 실행하기 위해 SparkContext는 여러 종류의 Cluster Manager(스파크 자체 독립적인 클러스터 매니저, Mesos, YARN, Kubernetes)에 연결할 수 있다.
2. SparkContext는 Cluster Manager에게 자원 할당을 요청하고, Cluster Manager는 요청에 따라 필요한 자원을 할당한다.
3. SparkContext는 할당받은 자원이 있는 노드에서 executor 프로세스를 시작하라는 명령을 내린다. executor 프로세스들은 실제 데이터 처리와 저장 작업을 수행한다.
   
4. 그 다음, 스파크는 애플리케이션 코드—즉, SparkContext로 전달된 Jar 또는 Python 파일—를 executor에게 전송한다.
5. 마지막으로, SparkContext는 executor에게 구체적인 작업을 수행하도록 지시한다. Executor는 이 지시에 따라 분산된 데이터 작업을 병렬로 수행한다.

 

https://spark.apache.org/docs/latest/cluster-overview.html

 

 

 

즉, 클러스터 관리자로부터 자원을 할당받고 실행자를 생성한 뒤, 스파크는 이러한 실행자들에게 사용자 코드를 전송하여 분산 데이터 처리 작업을 수행하게 한다. 이 과정을 통해, 스파크는 대규모 데이터 세트를 효과적으로 처리할 수 있는 분산 컴퓨팅 환경을 제공한다.

 

스파크의 저장소 계층

스파크는 하둡 분산 시스템(HDFS)이나 하둡 API가 지원하는 다른 저장 시스템(로컬 파일 시스템, S3, Cassandra, Hive 등)에 있는 모든 파일로부터 분산 데이터 모음을 만들 수 있다. 여기서 중요한 점은 스파크는 반드시 하둡을 필요로 하지 않는다. 단순히 하둡 API를 사용하는 저장 시스템을 지원하는 것뿐이다.

스파크는 텍스트 파일, SequenceFiles, Avro, Parquet 뿐 아니라 다른 하둡의 InputFormat(하둡이 입력 파일 포맷 지원에 사용하는 Java 클래스)이 지원하는 파일까지 지원한다.

 

스파크 사용해보기

Spark는 Coordinator 역할을 하는 드라이버 프로그램으로 구성된다. 이 드라이버 프로그램은 main 함수를 포함하며, 클러스터 상의 분산 데이터셋을 정의하고 해당 데이터셋에 연산 작업을 수행한다. 드라이버 프로그램의 예로는, Spark shell이나 scala 애플리케이션이 있다.

드라이버 프로그램들은 연산 클러스터에 대한 연결을 나타내는 SparkContext 객체를 통해 스파크에 접속한다. shell에서 이 객체는 자동으로 sc라는 변수에 만들어진다.

(base) $ bin/pyspark 
>>> sc
<SparkContext master=local[*] appName=PySparkShell>

 

SparkContext 객체를 하나 만들었다면 이것으로 RDD를 생성할 수 있다. 

아래 예제를 보면, 텍스트 파일의 각 라인을 표현하는 RDD를 만들기 위해 sc.textFile()을 호출한다. 그러고 나면 그 라인에 count()와 같은 다양한 연산을 수행할 수 있다.

>>> lines = sc.textFile("README.md")
>>> lines.count()
125                                                                             
>>> lines.first()
'# Apache Spark'

 

이러한 연산들을 수행하기 위해 드라이버 프로그램들은 보통 executor라고 불리는 다수의 노드를 관리한다. 예를 들어, 앞선 예제 코드에서 count()를 클러스터에서 실행하면 각 클러스터의 각 머신들은 파일의 서로 다른 부분을 처리하게 된다. (하지만 나는 로컬 모드로 spark shell을 실행했으므로 스파크는 하나의 머신에서 모든 작업을 처리한다.)

 

마지막으로, 사용자가 특정 데이터 처리 작업을 위해 함수를 정의하고, 이 함수를 스파크 API(ex. map, filter, reduct, join)의 인자로 전달하여 실행한다. 이렇게 전달된 함수는 클러스터 내의 여러 노드에 분산되어 저장된 데이터에 대해 병렬로 실행된다. 아래 예시를 살펴보자.

>>> lines = sc.textFile("README.md")
>>> pythonLines = lines.filter(lambda line: "Python" in line)
>>> pythonLines.first()
'high-level APIs in Scala, Java, Python, and R, and an optimized engine that'

 

스파크 API (ex. filter)는 사용자 정의 함수를 기반으로 하는 연산을 클러스터 상의 여러 노드에서 병렬로 실행한다. 즉, 스파크는 사용자가 작성한 코드(lines.contains("Python"))를 클러스터의 executor에게 전송하여, 각 노드에서 동시에 실행되도록 한다. 따라서 단일 프로그램이 클러스터 내의 다수 노드에서 동시에 작동하며, 대규모 데이터 처리를 효과적으로 병렬화 할 수 있다. 

 

참고) 파이썬 스크립트를 실행하려면 스파크에 포함되어 있는 bin/spark-submit shell 스크립트를 이용해야만 한다.
spark-submit 스크립트는 파이썬에서 스파크를 연동하는 데 필요한 작업들을 포함하고 있으며, 스파크의 파이썬 API가 동작할 수 있는 환경을 설정한다.

사용 예)
bin/spark-submit my_script.py

 

독립적으로 스파크 사용해보기

파이썬 스크립트를 실행하려면 스파크에 포함되어 있는 bin/spark-submit shell 스크립트를 이용해야만 한다.
spark-submit 스크립트는 파이썬에서 스파크를 연동하는 데 필요한 작업들을 포함하고 있으며, 스파크의 파이썬 API가 동작할 수 있는 환경을 설정한다. 사용 예시는 아래와 같다.

bin/spark-submit my_script.py

 

자바나 스칼라에서는 Maven 의존성 필드에 spark-core artifact를 추가하여 사용한다.

(Maven : 자바 기반 언어들이 공개된 저장소의 라이브러리를 가져와 링크하게 해주는 패키지 관리 툴)

<!-- https://mvnrepository.com/artifact/org.apache.spark/spark-core -->
<dependency>
    <groupId>org.apache.spark</groupId>
    <artifactId>spark-core_2.13</artifactId>
    <version>3.5.1</version>
</dependency>

 

SparkContext 초기화하기

한 애플리케이션이 스파크에 연동되려면 관련 스파크 패키지를 import 하고 SparkContext 객체를 생성해야 한다. 먼저, SparkConf 객체를 만들어 필요한 설정을 하고 SparkContext를 생성한다. 아래는 스칼라에서 스파크를 초기화하는 예제이다.

 

SparkContext를 초기화 할 때 전달하는 두 가지 인자가 있다.

• 클러스터 URL : 아래 예제에서 local 이라고 쓰인 부분으로 스파크에게 어떤 식으로 클러스터에 접속할지 알려준다. local은 한 개의 쓰레드나 단일 로컬 머신에서 돌 때 따로 접속할 필요가 없음을 알려주는 특수 값이다.
• 애플리케이션 이름 : 아래 예제에서 My App이라고 쓰인 부분으로 클러스터 UI에서 해당 이름으로 애플리케이션을 구분할 수 있다.

// SparkContext 사용
val conf = new SparkConf().setMaster("local").setAppName("My App")
val sc = new SparkContext(conf)

// SparkSesion 사용
val conf = new SparkConf()

val sc = SparkSession.builder
  .config(conf)
  .appName("My App")
  .enableHiveSupport()
  .getOrCreate()

 

 

(더 알아보기) sparkContext vs sparkSession

Apache Spark : SparkContext vs SparkSession

 

 

 

References

https://spark.apache.org/docs/latest/cluster-overview.html

러닝 스파크 구버전