(빅데이터분석 with Spark) Ch02. Introduction to Data Analysis with Scala and Spark

 

 

목차

 

 

 

1. Introduction to Data Analysis with Scala and Spark

 

Docker 환경, Spark 버전은 2.4.6을 사용하여 실습을 진행할 것이다. (Docker 사용법 등은 추후에 따로 다룰 것이다.)

 

1. "cmd" (mac은  terminal)를 관리자권한으로 실행한다.

2. 원하는 directory에 workspace를 만든다
mkdir<your workspace directory>

3. 다음을 실행하여 docker에 가상환경을 설치 및 실행한다. (모두 입력 후 enter → spark, hadoop 등 여러 개를 포함함)
docker run -h "localhost" -v <your workspace directory>:/root/workspace -p 8888:8888
-p 8080:8080 -p 8088:8088 -p 4040:4040 -p 50070:50070 titania7777/spark2_notebook:1

4. 'http://localhost:8888' 링크를 열어 웹 브라우저, 특히 jupyter notebook에서 실습을 진행한다.

 

 

 

 

1) Scala for data scientists

 

Spark는 R이나 Python API도 존재하지만,  Spark 기본언어인 Scala로 진행하는 것이 빠르고 최신 기능의 update도 바로 적용이 된다.

 

Scala 언어로 Spark를 사용할 때의 장점

• Performance Overhead를 줄여줌
  • R이나 Python으로 알고리즘을 실행하게 되면 다른 환경으로 코드와 데이터를 넘겨주어야 하는데, 가끔 이 과정에서 데이터를 잃는 등의 오류가 발생할 수 있다.
• 최신 버전 접근 용이성
  • R이나 Python 기반 API들은 Scala보다 늦게 업데이트가 이루어진다.
• Spark를 좀 더 효율적으로 이해할 수 있음
  • Scala 언어로 Spark를 사용할 줄 알면 시스템의 흐름을 보다 정확하게 알 수 있다.

 

Scala Tutorials : https://www.tutorialspoint.com/scala/index.htm

 

 

 

(1) Identifiers

 

variable, function 등의 이름의 첫 글자는 언더바(_) 또는 알파벳이어야 함

• age, salary, _value, __1_value → o
• $salary, 123abc, -salary → x
• abstract, case,  catch → x (keyword와 겹침)

 

 

 

(2) Keywords

 

Keywords in Scala

 

 

 

(3) Comments

 

주석은 C언어와 같다

// : 1줄 주석처리

/*

...   : 여러 줄 주석처리

*/

 

 

 

(4) Import Packages

 

Python과 비슷하지만, 중괄호를 통해 여러 Package들을 한꺼번에 Import할 수 있다.

 

• import a._ : a의 모든 package import
• import A.{a, b} : A의 a, b package import

ex)

import scala.xml._
import scala.collection.mutable.HashMap
import scala.collection.immutable.{TreeMap, TreeSet}

 

 

 

(5) Data Types

 

• 정수 데이터 타입
  • Byte : 8 bit signed value (-128 ~ 127)
  • Short : 16 bit signed value (-32768 ~ 32767)
  • Int : 32 bit signed value (-2147483648 ~ 2147483647)
  • Long : 64 bit signed value (-9223372036854775808 ~ 9223372036854775807)
• 실수 데이터 타입
  • Float : 32 bit IEEE 754 single-precision float
  • Double : 64 bit IEEE 754 double-precision float
• 기타
  • Char : 16 bit unsigned Unicode Character (U+0000 ~ U+FFFF)
  • String : Sequence of Chars
  • Boolean : True or False
  • Unit, Null, Nothing, Any, AnyRef 등

 

 

 

(6) Variable

 

※ 세미콜론(;)은 있어도 되고, 없어도 됨

 

var : variable (변수 → 값 변경 가능)

val : constant (상수 → 값 변경 불가)

 

var myVar : String = "Foo"

• myVar이라는 String 데이터타입의 변수를 생성하고 "Foo"라는 값을 지정함

 

var myVar :Int;
val myVal = "Hello, Scala!";

• myVar이라는 Int 데이터타입의 변수 생성
• myVal 이라는 상수에 "Hello, Scala!"라는 string 값 지정

 

val (myVar1: Int, myVar2: String) = Pair(40, "Foo")

• myVar1 상수에는 40이라는 int 값을, myVar2 상수에는 "Foo"라는 string 값을 지정 (Int, String 지정해주지 않아도 됨)

 

 

 

(7) if-else-elseif

 

var x = 30;
 
if (x == 10)
{
    println("Value of X is 10");
}
else if (x == 20) {
    println("Value of X is 20");
}
else {
    println("This is else statement");
}
 
// 결과
// Value of X is 30
// x: Int = 30

 

 

 

(8) while

 

while(condition) {
	statement(s);
}

 

var a = 15;
 
// while loop execution
while(a < 20) {
    println("Value of a: " + a);
    a += 1;
}
 
// 결과
// Value of a: 15
// Value of a: 16
// Value of a: 17
// Value of a: 18
// Value of a: 19
// a: Int = 20

 

※ do-while : statement가 최소 한 번은 실행됨

do {
	statement(s);
}
while(condition);

 

 

 

(9) for loop with ranges

 

for (var x <- Range) {
	statement(s);
}

 

→ Range를 나타낼 때 'to'는 뒤에 오는 숫자를 포함하고, 'until'은 뒤에 오는 숫자를 포함하지 않음

for (i <- 1 to 5) {
    println("Value of i: " + i);
}
 
// 결과
// Value of i: 1
// Value of i: 2
// Value of i: 3
// Value of i: 4
// Value of i: 5
 
for (i <- 1 until 5) {
    println("Value of i: " + i);
}
 
// 결과
// Value of i: 1
// Value of i: 2
// Value of i: 3 
// Value of i: 4

 

→ 이중 for loop의 경우, 뒷 조건이 먼저 실행됨

for (i <- 1 to 3; j <- 1 to 3) {
    println("Value of i: " + i);
    println("Value of j: " + j);
}
 
// 결과
// Value of i: 1
// Value of j: 1
// Value of i: 1
// Value of j: 2
// Value of i: 1
// Value of j: 3
// Value of i: 2
// Value of j: 1
// ...

 

 

 

(10) for loop with collections

 

for (var x <- List) {
	statement(s);
}

 

val numList = List(1, 2, 3, 4, 5, 6);
 
// for loop execution with a collection
for (a <- numList) {
    println("Value of a: " + a);
}
 
// 결과
// Value of a: 1
// Value of a: 2
// Value of a: 3
// Value of a: 4
// Value of a: 5
// Value of a: 6
//
// numList: List[Int] = List(1, 2, 3, 4, 5, 6)

 

→ for loop with filters

for( var x <- List
	if condition1; if condition2...) { // condition 사이에 and or 등 연산자는 없음!
    statement(s);
}

 

→ for loop with yield

var retVal = for{ var x <- List
	if condition1; if condition2 ...
}
yield x

var a = 0;
val numList = List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
 
// for loop execution with a yield
var retVal = for{ a <- numList if a != 3; if a < 8} yield a
 
// Now print returned values using another loop
for (a <- retVal) {
    println("Value of a: " + a);
}
 
// 결과
// Value of a: 1
// Value of a: 2
// Value of a: 4
// Value of a: 5
// Value of a: 6
// Value of a: 7
//
// a: Int = 0
// numList: List[Int] = List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
// retVal: List[Int] = List(1, 2, 4, 5, 6, 7)

 

Function Definitions (Body를 포함하는 경우)

def functionName ([list of parameters]) : [return type] = {
	function body
    return [expr]
}

 

Function calling

functionName(list of parameters)

 

→ 인스턴스의 함수일 경우, C++과 동일하다.

[instance.]functionName(list of parameters)

 

ex)

def addInt(a:Int, b:Int) : Int = {
	var sum: Int = 0
    sum = a + b
    return sum
}

// 결과
// addInt: (a: Int, b: Int)Int

println("Returned Value: " + addInt(5,7));

// 결과
// Returned Value: 12

 

→ Anonymous function (익명 함수)

ex)

(x: Int) => x + 1
(x: Int, y: Int) => "(" + x + ", " + y + ")"
() => { System.getProperty("user.dir") }

 

 

(12) Collections

 

• Scala Lists
• Scala Sets
• Scala Maps
• Scala Tuples
• Scala Options
• Scala Iterators

 

 

 

(13) Class & Object

 

한 번 클래스를 선언하면 keyword 'new'를 통해 object를 생성할 수 있다. (C++과 유사함)

class Point(xc: Int, yc: Int) {
    var x: Int = xc
    var y: Int = yc
 
    def move(dx: Int, dy: Int) {
        x += dx
        y += dy
        println("Point x location : " + x);
        println("Point y location : " + y);
    }
}
 
val pt = new Point(10, 20);
 
pt.move(10, 10);
 
// 결과
// defined class Point
// pt: Point = Point@d3b8556
// Point x location : 20
// Point y location : 30

 

 

2) Spark Programming Model

 

Spark Programming의 단계

1. 입력 data set을 Preprocessing한다.
2. preprocessing의 결과를 분석한다.
3. local computation을 진행한다.

 

 

 

(1) Record Linkage

 

똑같은 object의 표현에도 data를 다양하게 나타낼 수 있다. 같은 entity이지만 서로 다른 entity인 것 처럼 다르게 표현될 수도 있고, 다른 entity이지만 서로 같은 value를 가져서 같은 것으로 인식할 수 있다.

→ Data의 'Cleansing'이 필요하다!

 

 

---

<Practice : Github 참조>

https://github.com/JJukE/BigdataAnalysis

JJukE - Overview

 

 

 

 

 

2. RDD, DataFrame

 

 

 

 

1) RDD (Resilient Distributed Dataset)

 

Spark는 빅데이터를 multiple cluster (node)의 분산된 data set으로 분석한다. → distributed!

 

 

 

(1) RDD란?

 

Spark에서 cluster 내부에서 여러 기계(node)들로 분산시킨 data들을 표현하기 위한 기본적인 abstraction

 

• 변형이 불가능함 (Read Only)
• 특정 동작을 하기 위해서는 새로운 RDD가 생성되어야 하며, 이것들이 모여 'RDD Lineage'를 이룸
• 이 RDD Lineage가 DAG(Directed Acyclic Graph)의 형태를 가지면서, Fault-tolerant 특성을 갖게 됨

 

 

 

(2) RDD 생성하는 두 가지 방법

 

Spark에서 RDD를 만드는 두 가지 방법

1. 외부 데이터 소스(.csv 파일 등)로부터 RDD 생성 → 'SparkContext' 사용
2. 이미 존재하는 RDD를 filtering 등을 통해 변형시켜 새로운 RDD 생성

 

 

 

 

2) DataFrame

 

DataFrame을 만들기 위해서는 'SparkContext'대신에 'Spark'라는 library를 사용한다.

 

 

(1) DataFrame이란?

 

DataFrame은 Python pandas, R에서의 DataFrame, SQL의 Data Table과 비슷한 개념이다.

DataFrame은 Database에 가까운 abstraction으로, Spark에서의 표라고 볼 수 있다.

하나의 기계에 저장된 local data가 아니라 Cluster의 분산된 data set을 표현할 수 있는 것이다.

 

RDD의 경우, 메모리나 디스크에 저장 공간이 충분하지 않으면 잘 동작하지 않는 등의 여러 한계가 존재해서, DataFrame을 도입하였다.

보통 행(row)은 각각의 data point를, 열(column)은 data의 feature(성질, 변수)를 나타낸다.