[apache spark] [Spark Programming]1. Apache Spark 개요 및 설치
1. Apache Spark
- 하둡 기반의 맵리듀스 작업이 가진 단점들을 보완하기 위해서 만들어 진 프레임워크
- 하둡과 달리 인메모리(In-Memory) 기법을 활용한 데이터 저장 방식을 제공함으로써 머신러닝 등
반복적인 데이터 처리가 필요한 분야에서 높은 성능을 보여준다.
- 작업을 실행하기 전에 최적의 처리 흐름을 찾는 과정을 포함하고 있었기 때문에 성능 향상과
더불어 여러 개의 맵리듀스 잡을 직접 순차적으로 실행하는 수고를 덜어줬다.
- 이외에 실시간 스트리밍 데이터를 다루기 위한 스파크 스트리밍과 하이브와 연동도 가능한
스키마 기반 데이터 분석모듈인 스파크 SQL, R과 연동가능한 sparkR 등 데이터 처리 분야에
특화된 라이브러리도 지원해준다.
1) RDD(Resilient Distributed Dataset)
- 스파크가 제공하는 일종의 분산 데이터에 대한 모델이자 핵심 데이터 모델임.
- 다수의 서버에 걸쳐 분산 방식으로 저장된 데이터 요소들의 집합을 의미
- 동시에 병렬로 명령처리가 가능하고 장애가 발생할 경우 스스로 복구될 수 있는 내성을 가지고 있다.
* 스파크에서는 하나의 작업을 수행할 때 파티션 단위로 나눠서 병렬로 처리한다.
또한 작업이 진행되는 동안 재구성되거나 네트워크를 통해 다른 서버로 이동하는, 셔플링 현상이
발생할 수 있다.
- 다수의 서버에 나눠져서 처리되기 때문에 유실이 발생할 수 있지만 이를 위해 원래 상태로 복구할 수
있도록 RDD의 생성과정을 기록해 뒀다가 다시 복구해주는 기능을 가지고 있다.
- 또한 문제가 발생 시 전체작업을 처음부터 다시 실행하는 대신 문제가 발생한 RDD를 생성했던
작업만 다시 수행해서 복구를 진행한다.
- 리니지(Lineage)
스파크에서 RDD 생성 작업을 기록해 두는 것
(1) 생성 방법
- List, Set 같은 기존 프로그램의 메모리에 생성된 데이터를 이용하는 것
-> 즉시 테스트해 볼 수 있어 테스트 코드 작성 등에 유용하게 사용됨
ex. Collection 을 이용
[Java Code]
JavaRDD<String> rdd = sc.parallelize(Arrays.asList("a", "b", "c", "d", "e"));
[Scala Code]
val rdd = sc.parallelize(List("a", "b", "c", "d", "e"))
[Python Script]
rdd = sc.parallelize(["a", "b", "c", "d", "e"])
- 로컬 파일시스템이나 하둡의 HDFS 같은 외부 저장소에 저장된 데이터를 읽어서 생성하는 방법
-> 다양한 유형의 데이터 소스로부터 데이터를 읽고 RDD를 생성할 수 있다.
ex. 파일로부터 생성
[Java Code]
JavaRDD<String> rdd = sc.textFile("<path_to_file>");
[Scala Code]
val rdd = sc.textFile("<path_to_file>")
[Python Script]
rdd = sc.textFile("<path_to_file>")
- 기존에 생성돼 있는 RDD로부터 또 다른 RDD를 생성하는 방법
- createRDD 같은 함수가 제공되는 것은 아니지만 기존 RDD의 모든 요소에 1을 더하는 등의
연산을 적용하면 "한 번 만들어지면 수정불가하다" 는 성질을 이용해 새로운 RDD 가 생성되는 것
ex. 기존 RDD로부터 새로운 RDD 생성
[Java Code]
JavaRDD<String> rdd1 = rdd.map(new Function<String, String>() {
@Override
public String call(String v1) throws Exception {
return v1.toUpperCase();
}
});
* JAVA 8 의 경우 다음과 같이 한다.
JavaRDD<String> rdd1 = rdd.map(v -> v.toUpperCase());
[Scala Code]
val rdd1 = rdd.map(_.toUpperCase())
[Python Script]
rdd1 = rdd.map(lambda s : s.upper())
(2) 연산
- 크게 Transformation. Action 으로 분류된다.
- Transformatoion
어떤 RDD에 변형을 가해 새로운 RDD를 생성하는 연산
기존 RDD는 바뀌지 않은 상태에서 변형된 값을 가진 새로운 RDD 가 생성된다.
=> RDD에 대한 생성 계보만 만들면서 쌓아 두고 액션 연산이 호출되면 한꺼번에 수행됨
때문에 본격적인 작업 실행 전에 데이터가 어떤 방법과 절차에 따라 변형되어야 하는지
알 수 있다는 장점을 가진다.
- Action
해당 연산의 결과로 RDD가 아닌 다른 값을 반환하거나 아예 반환하지 않는 연산을 의미함
* sum(), stddev() 등의 메소드는 RDD 의 모든 요소가 숫자형인 경우에 사용이 가능함
* groupByKey() 메소드의 경우 키와 값싸으로 구성된 RDD 에서만 사용가능
2) DAG(Directed Acyclic Graph)
- 그래프 이론에서 사용되는 용어로 여러 개의 꼭짓점 또는 노드와 그사이를 이어주는 방향성을 지닌
선으로 구성되고 그래프를 구성하는 꼭짓점, 노드 에서 출발하더라도 다시 원래의 꼭짓점으로
돌아오지 않도록 구성된 그래프 모델을 의미한다.
- 스파크에서 각 단계마다 최적화된 작업을 수행하기 위해 사용해야되는 데이터 처리용 라이브러리가
다를 수 있기 때문에 서로 다른 라이브러리를 잘 조합해서 사용할 수 있도록 일련의 작업 흐름을
나타내는 워크 플로우
(1) DAG 스케쥴러
- DAG 생성을 담당하는 부분
- 동작 방식
- 전체 작업을 스테이지라는 단위로 나누어 실행하고 각 스테이지를 다시 여러 개의 태스크로
나누어 실행
- 드라이버의 메인 함수에서 스파크 애플리케이션과 스파크 클러스터의 연동을 담당하는 스파크
컨텍스트라는 객체를 생성하고 이를 이용해 잡을 실행하고 종료하는 역할을 수행한다.
- 스파크컨텍스트를 통해 RDD의 연산 정보를 DAG스케줄러에게 전달하면 스케줄러는 해당 정보를
이용해 실행 계획을 수립한 후 클러스터매니저에게 전달한다.
- 전체 데이터 처리 흐름을 분석해 네트워크를 통한 데이터 이동이 최소화되도록 스테이지를
구성한다.
- 스케줄러의 역할: 대상 데이터의 크기를 줄여 셔플로 인한 부하를 최소화하는 것
(2) 좁은 의존성 vs. 넓은 의존성
- 좁은 의존성
기존 RDD와 신생 RDD의 관계가 작은 경우를 의미( ex. 1:1 대응인 경우)
- 넓은 의존성
기존 RDD가 여러 신생 RDD와 관계를 맺는 경우를 의미 (ex. 1:N 대응인 경우)
3) 람다 아키텍쳐
- 네이션 마츠가 제안한 모델로 빅데이터 처리를 위한 시스템을 구성하는 방법 중 하나이다.
- 빅데이터의 활용 분야가 넓어지면서 기존과 같은 대량의 데이터 처리는 몰론, 실시간 로그 분석과
같은 실시간 처리도 매우 중요해지는 상황이기 때문에 이를 만족 시키기 위한 아키텍쳐로 채택됨
-크게 일괄처리 계층과 속도 계층으로 나뉨
- 과정
- 새로운 데이터는 일괄 처리 계층과 속도 계층 모두에 전달된다.
- 일괄처리 계층은 원본 데이터를 저장하고 일정 주기마다 한 번씩 일괄적으로 가공해서 배치 뷰를
생성한다.
- 속도 계층은 들어오는 데이터를 즉시 또는 매우 짧은 주기로 처리해 실시간 뷰를 생성한다.
- 서빙 계층은 실시간 뷰와 배치 뷰의 결과를 적절히 조합해 사용자에게 데이터를 전달한다.
- 서빙 계층을 거치지않고 배치 뷰 또는 실시간 뷰를 직접 조회할 수도 있다.
-> 일괄 처리 작업을 통해 데이터를 처리하되 아직 배치 처리가 수행되지 않은 부분은 실시간 처리를
통해 보완한다는 개념이다.
-> 이 경우 속도 계층의 처리 결과는 다소 정확하지 않을 수 있지만 추후에 일괄 처리 작업을 통해
다시 보정하는 형태로 운영될 수 있다,
2. Spark 설치
1) JDK
- Java8 의 경우 람다식을 사용하므로 이 후에 다뤄질 예정이기 때문에 Java8 버전이상으로
설치한다.
- 설치 후 설치 위치(PATH)를 환경 변수에 적용시켜준다.
2) Maven 설치
- 자바 프로젝트 통합 관리 도구로서 자바 개발자에게는 익숙한 도구이다.
받는다.
- 설치 후 설치 위치(PATH)를 환경 변수에 적용시켜준다.
- install new software -> http://alchim31.free.fr/m2e-scala/update-site 에서 설치 진행
- Eclipse 에서 Maven 프로젝트 실행 시 pom.xml 에 추가해야 되는 내용
[pom.xml]
...
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>스칼라 라이브러리 버전</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_코어 버전</artifactId>
<version>스파크 버전</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_스파크SQL 버전</artifactId>
<version>스파크 버전</version>
</dependency>
...
...
ex.
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.11.11</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.3.3</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.11</artifactId>
<version>2.3.3</version>
</dependency>
...
3) Spark 설치
때문에 다음과 같이 받았다.
- 설치하게되면 앞선 프로그램들과 마찬가지로 시스템 환경 변수에 설치 경로(PATH)를 저장한다.
- 정상작동하는지 확인하기위해 간단한 테스트를 시행한다.
- 추가적으로 ./bin/spark-shell 을 실행시켜본다.
- 아래 결과에 대한 실행 결과는 다음과 같다.
[Scala code]
val file = sc.textFile("file://[스파크 설치 위치(절대경로로 입력할 것!)]/README.md")
val words = file.flatMap(_.split(" "))
val result = words. countByValue
result.get("For"). get
[실행 결과]
res1: Long = 3
- 마지막으로 웹 브라우저에서 [자신의 IP 주소]:4040 으로 환경이 제공되는지 확인한다.
4) 파이썬 설치
- 스파크에서는 파이썬 2.6 / 3.4 이상의 버전을 지원한다.
- 설치 이후 환경변수에 설치 경로를 추가시킨다.
5) 스칼라 IDE 설치
- 스파크는 스칼라, 파이썬, R 을 지원한다. 그 중 스칼라는 eclipse를 기반으로 IDE 가 제공된다.
스칼라와 스칼라IDE 모두 설치한다.
가능하다.
6) 파이썬 플러그인 설치
- Help - Install New Software - Add - Name 에 "PyDev" 입력 - PyDev 플러그인 설치
- Preference - PyDev - Interpreters - Python Interpreter 에서 새로운 인터프리터를 설치
- 바로 아래의 Libraries 탭을 눌러 스파크 라이브러리를 추가한다.
New Egg/Zip(s) - [python 설치 경로]/lib/py4j-0.10.4-src.zip 을 설치한다.
- 마지막으로 Environment 에서 SPARK_HOME과 PYSPARK_PYTHON 변수를 생성한다.
SPARK_HOME 은 스파크 설치 위치, PYSPARK-PYTHON은 파이썬 설치 위치를 입력한다.
2. 스파크 실행해보기
- 아래의 코드를 이용하여 spark 를 실행해보자
[WordCount.jar]
package Java.Spark;
import java.util.Arrays;
import java.util.Iterator;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;
public class WordCount {
public static void main(String[] args)
{
if (ArrayUtils.getLength(args) != 3)
{
System.out.println("Usage: WordCount <master> <input> <output>");
return;
}
// SparkContext 생성
JavaSparkContext sc = getSparkContext("WordCount", args[0]);
try
{
JavaRDD<String> inputRDD = getInputRDD(sc, args[1]);
//JavaPairRDD<String, Integer> resultRDD = process(inputRDD);
JavaPairRDD<String, Integer> resultRDD = processWithLambda(inputRDD);
handleResult(resultRDD, args[2]);
}
catch(Exception e)
{
e.printStackTrace();
}
finally
{
sc.stop();
}
}
public static JavaSparkContext getSparkContext(String appName, String master)
{
SparkConf conf = new SparkConf().setAppName(appName).setMaster(master);
return new JavaSparkContext(conf);
}
public static JavaRDD<String> getInputRDD(JavaSparkContext sc, String input)
{
return sc.textFile(input);
}
// Java 7
public static JavaPairRDD<String, Integer> process(JavaRDD<String> inputRDD)
{
JavaRDD<String> words = inputRDD.
flatMap(new FlatMapFunction<String, String>(){
public Iterator<String> call(String s) throws Exception
{
return Arrays.asList(s.split(" ")).iterator();
}
});
JavaPairRDD<String, Integer> wcPair = words
.mapToPair(new PairFunction<String, String, Integer>() {
public Tuple2<String, Integer> call(String s) throws Exception
{
return new Tuple2(s, 1);
}
});
JavaPairRDD<String, Integer> result = wcPair
.reduceByKey(new Function2<Integer, Integer, Integer>() {
public Integer call(Integer v1, Integer v2) throws Exception {
return v1 + v2;
}
});
return result;
}
// Java 8
public static JavaPairRDD<String, Integer> processWithLambda(JavaRDD<String> inputRDD)
{
JavaRDD<String> words = inputRDD.flatMap((String s) -> Arrays.asList(s.split(" ")).iterator());
JavaPairRDD<String, Integer> wcPair = words.mapToPair((String w) -> new Tuple2(w, 1));
JavaPairRDD<String, Integer> result = wcPair.reduceByKey((Integer c1, Integer c2) -> c1 + c2);
return result;
}
public static void handleResult(JavaPairRDD<String, Integer> resultRDD, String output)
{
resultRDD.saveAsTextFile(output);
}
}
- 위의 코드는 Java 7 과 Java 8 모두 가지고 있으며 실행시에는 Java 8 Lambda 식을 활용한 코드가 사용된다.
- pom.xml 은 파일을 다운 받는 것을 추천한다.
- 실행은 HDFS 가 설치 되었다는 전제하에 다음의 명령어를 실행하며 만약 HDFS 가 설치 안되었다면 아래의
형식에 맞춰 사용하기 바란다.
[형식]
<spark-home-dir>/bin/spark-submit \
--class <package주소>.WordCount \
<jar 파일 경로>/<jar파일명>.jar \
local[*] \
<spark-home-dir>/README.md \
<spark-home-dir>/testresult
[실행 예시 - HDFS 설치 시]
./bin/spark-submit \
--class Java.Spark.WordCount \
hdfs://master:9000/jar/spark/WordCount.jar \
local[*] \
hdfs://master:9000/input/README.md \
hdfs://master:9000/output/WordCountResult
[실행 화면 - 확인]
[hadoop@master spark-2.3.0-bin-hadoop2.7]$ hadoop fs -cat /output/WordCountResult/part*
(package,1)
(this,1)
(Because,1)
(Python,2)
(cluster.,1)
(its,1)
([run,1)
(general,3)
(have,1)
(pre-built,1)
(YARN,,1)
(locally,2)
(changed,1)
(locally.,1)
(sc.parallelize(1,1)
(only,1)
(several,1)
.....
[출처] https://m.blog.naver.com/slykid/221186930260
