SparkSession

모든 스파크 애플리케이션은 가장 먼저 SparkSession을 생성

기존 코드 : new SparkContext 패턴을 사용 (과거 SparkContext, SQLContext 를 직접 생성)

최신 코드 : SparkSession의 빌더 메서드를 사용해 생성할것을 추천

SparkSession 클래스는 스파크2.x 버전 에서 사용이 가능하다

new SparkContext 에서 보다 안전하게 생성할 수 있다. 다수의 라이브러리가 세션을 생성하려는 상황에서 컨텍스트 충돌을 방지 할 수 있다.

예제

val spark = SparkSession.builder
.appName("My Spark Application")  // optional and will be autogenerated if not specified
.master("local[*]")      // only for demo and testing purposes, use spark-submit instead
.enableHiveSupport()              // self-explanatory, isn't it?
.config("spark.sql.warehouse.dir", "target/spark-warehouse")
.config("spark.som.config.option","some-value")
.getOrCreate()



SparkSession, SQLContext, HiveContext

과거 버전의 스파크에서는 SQLContext와 HiveContext를 사용해 DataFrame과 스파크 SQL을 다룰수 있었음.
과거 SparkContext는 스파크의 핵심 추상화 개념을 다루는데 중점을 두고 , SQLContext는 스파크의 핵심 추상화 개념을 다루는데 중점을 둠 
Spark2.x 이후 부터는 두 콘텍스트를 SparkSession으로 단일화함
SparkContext, SQLContext는 존재하지만 SparkSession을 통하여 접근이 가능하다.

멀티 세션

newSession

SparkContext를 공유하고 session 만 달라지는 방법, 개별의 session 마다 동일한 이름의 테이블 생성 가능.

동일한 jvm에 여러개의 스파크 컨텍스트를 갖는것은 권장되지 않으며 더 불안하고 하나의 스파크 컨텍스트가 충돌하면 다른 스파크에 영향을 줄수 있다.

newSession으로 만들어진 세션중 하나라도 stop/close 를 하게 되면 동일한 SparkContext를 사용하기 때문에 종료가 된다.

sparkSession.newSession();

참고

Spark 완벽 가이드(도서)

Spark -newSession (https://medium.com/@achilleus/spark-session-10d0d66d1d24)

SparkSession - The Entry Point to Spark SQL (https://jaceklaskowski.gitbooks.io/mastering-spark-sql/spark-sql-SparkSession.html)

RDD 와 RDD 에서 호출하는 액션들에 대한 모든 의존성을 재연산하게 되는데, 호출하는 액션들에 대한 모든의존성을 재연산하게 된다.

이때 데이터를 여러번 스캔하는 반복알고리즘들에 대해서는 매우 무거운 작업일 수 있다.

여러번 반복 연산하는 것을 피하려면 스파크에 데이터 영속화(persist/persistence)를 요청할 수 있다.

RDD 영속화에 대한 요청을 하면 RDD를 계산한 노드들은 그 파티션들을 저장하고 있게 된다.

자바에서는 기본적으로 persist()가 데이터를 JVM 힙(heap) 에 직렬화되지 않는 객체 형태로 저장.

레벨 공간사용 CPU 사용시간 메모리에 저장 디스크에 저장 비고
MEMORY_ONLY 높음 낮음 예 아니오
MEMORY_ONLY_SER 낮음 높음 예 아니오
MEMORY_AND_DISK 높음 중간 일부 일부 메모리에 넣기에 데이터가 너무 많으면 디스크에 나눠 저장
MEMORY_AND_DISK_SER 낮음 높음 일부 일부 메모리에 넣기에 데이터가 너무 많으면 디스크에 나눠 저장.메모리에 직렬화된 형태로 저장
DISK_ONLY 낮음 높음 아니오 예

persist() 호출은 연산을 강제로 수행하지않는다.

메모리에 많은 데이터를 올리려고 시도하면 스파크는 LRU 캐시 정책에 따라 오래된 파티션들을 자동으로 버림.

예제 코드

import org.apache.spark.storage.StorageLevel

val result = input map(x => x*x)
result.persist(StorageLevel.DISK_ONLY)
println(result.count())
println(result.collect().mkString(","))

https://jaceklaskowski.gitbooks.io/mastering-apache-spark/spark-broadcast.html

SPARK 에서의 기본 행동(action) 연산자 및 변환(transformation)연산자(2)

sample, take, takeSample 연산으로 RDD 의 일부 요소 가지고 오기

sample

이전의 고객 ID 중 30% 를 무작위로 고른 샘플 데이터셋이 필요하다고 가정할때,

1. RDD 클래스에 sample 메서드 사용 가능

def sample(widthReplacement: Boolean,fraction:Double, seed:Long=Util.random.nextLong):RDD[T]

첫 번째, widthReplacement 는 같은 요소가 여러번 샘플링될수 있는지에 대한지정

true : 복원샘플링, false : 비복원 샘플링

복원 샘플링은 물고기를 잡았을 때 다시 물고기를 살려주고 다시 물고기를 잡는 상황이라고 볼 수있고,

비복원 샘플링은 반대로 물고기를 잡았을때, 물고기를 제외하고 다시 물고기를 잡는 상황이라고 이해하면된다.

두 번째, fraction 샘플링될 횟수의 기댓값을 의미

세 번째, seed 는 난수 생성에 사용되는 시드로, 같은 시드는 항상 같은 유산 난수를 생성하기 때문에 프로그램을 테스트 하는데 쓰임.

이전의 예제인 uniqueIds 의 값들 중 샘플링 테스트를 해보자.

scala> val uniqueIds = idsStr.distinct
uniqueIds: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[6] at distinct at <console>:28
​
scala> uniqueIds.collect
res4: Array[String] = Array(80, 20, 98, 15, 16, 31, 94, 77)
​
scala> val ss = uniqueIds.sample(false,0.3)
ss: org.apache.spark.rdd.RDD[String] = PartitionwiseSampledRDD[11] at sample at <console>:30
/*
sample 의 반환형은 RDD 
*/
scala> ss.count
res10: Long = 2
​
scala> ss.collect
res11: Array[String] = Array(20, 98)

takeSample

위의 sample 메서드는 확률을 통해 값을 가지고 왔지만, 갯수를 가지고 sample을 하려면 takeSmaple 을 사용

def takeSample(widthReplacement: Boolean, num:Int, seed: Long=Utils.random.nextLong):Array[T]

첫번째 인자 : 복원,비복원

두번째는 가지고올 갯수

세 번째, seed 는 난수 생성에 사용되는 시드로, 같은 시드는 항상 같은 유산 난수를 생성하기 때문에 프로그램을 테스트 하는데 쓰임

scala> val taken = uniqueIds.takeSample(false,5)
taken: Array[String] = Array(31, 77, 94, 15, 16)
/*
takeSample의 반환형은 Array 로 반환하게 된다.
*/
scala> uniqueIds.take(3)
res12: Array[String] = Array(80, 20, 98)
​

take 는 RDD 에서 갯수만큼 가지고 오는 연산자인데, 지정된 개수의 요소를 모을때까지 RDD 파티션 하나씩 처리해 결과를 반환한다.

(파티션을 하나씩 처리 한다는 것은 결국 연산이 전혀 분산이 되지 않는다는 것을 의미한다. 여러 파티션의 요소를 빠르게 가져오고 싶다면 드라이버의 메모리를 넘지 않도록 요소 개수를 적당히 줄이고 collect 연산자를 사용한다.)

RDD연산자의 종류는 transformation 과 action둘로 나뉘는데

transformation 은 새로운 RDD 를 생성

action 은 RDD 의 연산자를 호출함

스파크에서 transformation , 과 action둘의 지연 실행

spark 에 대해서 lazy evaluation 개념이 중요한데, 처음에는 lazy 에 대한 이해를 하지 못 한채 그냥 그렇구나 했는데, 개념은 다음과 같다.

transformation의 지연 실행은 action 연산자를 호출하기 전까지는 transformation 연산자의 계산을 실제로 실행 하지 않는 것을 의미한다.
이는 RDD에 action연산자가 호출되면 스파크는 해당 RDD 의 계보를 살펴본 후, 이를 바탕으로 실행해야하는 연산 그래프를 작성해서 action 연산자를 계산한다. 
결론은 transformation 연산자는 action 연산자를 호출했을때, 무슨 연산이 어떤 순서로 실행되어야 할지 알려주는 일종의 설계도 라고 할 수 있다.

책의 예제를 따라다 우연히 lazy evaluation 의 예제를 찾은 것같다.

scala> val lines = sc.textFile("/home/morris01/study/spark/data/client-ids.log")
lines: org.apache.spark.rdd.RDD[String] = /home/morris01/study/spark/data/client-ids.log MapPartitionsRDD[4] at textFile at <console>:24
​
scala> val idsStr = lines.map(line=>line.split(","))
idsStr: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[5] at map at <console>:26
​
scala> idsStr.foreach(println)
org.apache.hadoop.mapred.InvalidInputException: Input path does not exist: file:/home/morris01/study/spark/data/client-ids.log
  at org.apache.hadoop.mapred.FileInputFormat.singleThreadedListStatus(FileInputFormat.java:287)
  at org.apache.hadoop.mapred.FileInputFormat.listStatus(FileInputFormat.java:229)
  at org.apache.hadoop.mapred.FileInputFormat.getSplits(FileInputFormat.java:315)
​

위의 실행 예제가 에러가 난이유는 filepath가 잘못되어서 나온 에러이다.

그러나 idsStr.foreach를 실행하기전까지는 순수히 진행이 되는 것 같았다. 하지면 foreach 라는 action을 수행을 하면서 이전의 RDD 의 계보를 살펴보다가 잘못되어서 에러가 발생한것같다

RDD 연산자

원본 RDD 의 각 요소를 변환한 후 변환된 요소로 새로운 RDD를 생성하는 map 변환 연산자

scala> val numbers = sc.parallelize(10 to 50 by 10)
numbers: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[4] at parallelize at <console>:24
​
scala> numbers.foreach(x=>println(x))
10
20
30
40
50
​
scala> val numberSquared = numbers.map(num=>num*num)
numberSquared: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[5] at map at <console>:26
​
scala> numberSquared.foreach(x=>println(x))
100
400
900
1600
2500
​
scala> numberSquared.foreach(println)
100
400
900
1600
​

distinct, flatMap 연산자

예제 데이터는 물건을 구매한 ID 값을 가진 log 파일이다.

echo "15,16,20,20
77,80,94
94,98,16,31
31,15,20" > ~/client-ids.log

scala> val lines = sc.textFile("/home/morris01/study/spark/data/sparkinaction/client-ids.log")
lines: org.apache.spark.rdd.RDD[String] = /home/morris01/study/spark/data/sparkinaction/client-ids.log MapPartitionsRDD[7] at textFile at <console>:24
​
scala> val idsStr = lines.map(line=>line.split(","))
idsStr: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[8] at map at <console>:26
​
scala> idsStr.foreach(println)
[Ljava.lang.String;@77278a7d
[Ljava.lang.String;@6876c229
[Ljava.lang.String;@25f5ac40
[Ljava.lang.String;@2d06d673
/*
idsStr 에는 string 하나, 하나의 rdd 가생성되는걸로 예상했는데 
string 배열을 가진 RDD 가 생성되었다. 
*/
scala> idsStr.first
res5: Array[String] = Array(15, 16, 20, 20)
​
scala> idsStr.collect
res6: Array[Array[String]] = Array(Array(15, 16, 20, 20), Array(77, 80, 94), Array(94, 98, 16, 31), Array(31, 15, 20))
/*
collect 를 사용하여 새로운 배열을 생성 , RDD의 모든 요소를 이 배열에 모아서 반환
*/

이 배열을 단일 배열로 분해 하려면 flatMap을 사용하게된다.

flatMap은 RDD 모든 요소에 적용이 된다.

익명함수가 반환한 배열의 중첩구조를 한단계 제거하고 모든 배열의 요소를 단일 컬렌션으로 병합한다는것이 flatmap 과 map 의 다른 점이다.

scala 에 대한 지식중 TraversableOnce 에 대해서 꼭 알 필요가 있다.

이유는 flatMap의 시그니쳐는 다음과 같이 가지고 있기 때문이다.

def flatMap[U](f:(T)=>TraversableOnce[U]):RDD[U]

map으로 연산을 했던 것을 flatMap을 사용하게 되면 하나의 배열로 값을 불러올 수 있는 것을 확인 할 수 있다.

scala> val idsStr = lines.flatMap(line=>line.split(","))
idsStr: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[2] at flatMap at <console>:26
​
scala> idsStr.collect
res1: Array[String] = Array(15, 16, 20, 20, 77, 80, 94, 94, 98, 16, 31, 31, 15, 20)
/*
String 의 값을 Int 로 반환해주기 위해서는 _.toInt 메서드를 사용하면된다.
*/
scala> val idsInt = idsStr.map(_.toInt)
idsInt: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[3] at map at <console>:28
​
scala> idsInt.collect
res2: Array[Int] = Array(15, 16, 20, 20, 77, 80, 94, 94, 98, 16, 31, 31, 15, 20)
​

Distinct

구매 고객들의 아이디 값 들을 연산하기 쉽게 하나의 배열로 나타냈지만, 구매고객의 수를 구하려면 중복을 제거를 해주어야한다.

보통은 Scala의 Set 함수에 다시 넣을수도있겠지만, 간편하게 Distinct 를 사용하면 된다.

scala> val uniqueIds = idsInt.distinct
uniqueIds: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[6] at distinct at <console>:30
​
scala> uniqueIds.collect
res4: Array[Int] = Array(15, 77, 16, 80, 98, 20, 31, 94)
​
scala> val finalCount = uniqueIds.count
finalCount: Long = 8
​

예제 파일 github : https://github.com/spark-in-action/first-edition/blob/master/ch02/scala/ch02-listings.scala

SPARK(3) SparkContext

지난 시간의 Spark의 실행하기 위해

val conf = new SparkConf().setAppName("HelloWorld").setMaster("local[1]")
      .set("spark.executor.memory", "4g")
      .set("spark.driver.memory", "4g")
val sc = new SparkContext(conf)

다음과 같은 코드를작성 하였는데

풀이를 하면

SparkConf() 는 SparkContext를 생성 하기 위한 설정 파일이다.

setAppName()은 Spark어플리케이션의 이름

setMaster() 는 로컬 피씨에서 사용 하기 위해 local로 적어준것이고 [ N ] 은 실행할 스레드의 개수(core)를 의미한다.

로컬 피씨의 전체 의 core를 쓰고 싶다면 [*] 를 사용해주면 된다.

SparkContext 객체는 클러스터상에서 스파크 작업 실행을 관리 하는 객체이다.

SparkContext 는 많은 유용한 메서드를 제공하는데 , 많이 쓰이는 것은 탄력적 분산 데이터셋 을 생성하는 메서드들을 가장 자주 사용하게 된다.

탄력적 분산 데이터셋(RDD) 은 클러스터의 여러 노드에 파티션으로 나뉘어 분산되며, 각 파티션은 RDD전체 데이터중 일부를 담게 된다.

여기서 파티션의 의미는 스파크에서 병렬 처리되는 단위.

RDD를 생성하는 간단한 방법은 로컬 객체 컬렉션을 인수로 SparkContext의 parallelize 메서드를 실행 하는것.

val rdd = sc.parallelize(Array(1,2,3,4),4)

첫 번째 파라미터는 병렬 처리 하려는 객체 컬렉션을 나타내며, 두번째 인수는 파티션의 개수이다.

파티션내의 객체들에 대한 연산을 수행하게 될때, 스파크는 구동자 프로세스로 부터 객체 컬랙션의 일부를 가지고 온다.

RDD를 HDFS, 텍스트파일 를 포함한 디렉토리로부터 생성하기 위해서는

val rdd2 = sc.textFile('hdfs:///hadoopData/process01.txt') 

textFile 메서드를 사용한다. 단, 메서드의 인수로 디렉토리 이름을 입력하게되면 스파크는 그 디렉터리의 모든 파일을 RDD로 구성요소로 간주하게된다. 이렇게 parallelize, textFile등의 메서드의 코드 의 시점까지는 데이터를 읽어들이거나, 메모리에 올리는 일은 실제로 일어나지 않는다. 스파크는 파티션 내의 객체들에 대해 연산을 수항할때가 되서야 섹션=스플릿 단위로 읽어 RDD에 정의한 필터링 과 같은 집계같은 작업을 통해 가공을 함.

오류가 있거나 궁금한점있으면 같이 공유했으면 좋겠습니다.

SPARK (2) - Spark 환경 설정 및 scala trait

1. Spark gradle dependncy 추가

    compile group: 'org.scala-lang', name: 'scala-library', version: '2.11.1'
    compile group: 'org.apache.spark', name: 'spark-core_2.11', version: '2.1.0'
    compile group: 'org.apache.spark', name: 'spark-sql_2.11', version: '2.1.0'
    compile group: 'org.apache.spark', name: 'spark-mllib_2.11', version: '2.1.0'

build.gradle 파일 오른쪽 마우스 클릭 gradle refresh

main함수 안에 다음고 같이 적어주고

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext._
import org.apache.spark.SparkContext

object MainScala {
  def main(arg: Array[String]) {
	val conf = new SparkConf().setAppName("HelloWorld").setMaster("local[1]")
          .set("spark.executor.memory", "4g")
          .set("spark.driver.memory", "4g")
	val sc = new SparkContext(conf)

	println("=========================")
	println("Hello Spark")
	println("=========================")
	sc.stop()
  }
}

ScalaApplication 을 실행 해주면 실행 완료

RDD 프로그래밍 하기

graph LR
A[RDD 연산]-->B[transformation]
A[RDD 연산]-->C[action]

python으로 예를 들면

lines = sc.textFile("README.md")

Transformation : 존재하는 RDD에서 새로운 RDD를 만들어낸다.(예제는 python)

예를 들면 표현식과 일치하는 어떤 데이터든 걸러내는것이 있을 때

pythonLines = lines.filter(lamda line: "Python" in line)

Action : RDD를 기초로 결과 값을 계산하며, 그 값을 드라이버 프로그램에 되돌려주거나 외부 스토리지(예. HDFS)에 저장 하기도한다.

예를 들면 기존에서 이미 써본적이 있는 액션으로는 첫번째 요소를 되돌려주는 first가 있다.

pythonLines.first()

 다음시간에는 spark shell을 이용한 기초 RDD 조작방법에 대해 포스팅 하겠습니다.

SCALA Trait

spark 를 scala 언어로 사용하는데 있어서 scala에 대한 정리도 추가로 하겠습니다.

잘못된 정보가 있으면 댓글로 알려주세요.

scala trait 믹스인

믹스인 ?? 개념이 헷갈린다 .

우선 실습을 하자면

mathFunction trait 은 sumTest 함수는 int 형으로 반환해야한다는 것을 정의하고 있다.

또한 sum 은 mathFunction을 상속 받고 있다.

//mathFunction.scala

trait mathFunction {
  def sumTest(value:Int):Int
  def averageTest(values:Array[Int]):Int
}

class mathLecture extends mathFunction{
  
   override def sumTest(value:Int):Int = {
      3+value
   }
   
   override def averageTest(values:Array[Int]):Int={
     var sum = 0;
     for(i<-(0 to values.length-1)){
       sum += values(i)
     }
     sum/values.length
   }
   
}

아래의 코드는 mathFunction 의 trait 타입을 가지는 mathClass 값을 설정 하였다.

mathClass의 리터럴을 sum() 클래스를 지정해주면 , mathClass 는 mathFunction 타입으로 sumTest 함수를 정의만 하고있지만, sum class 에 정의된 sumTest 함수를 사용할수 있다.

//MainScala.scala
object MainScala {
  def main(arg: Array[String]) {
    val mathClass:mathFunction = new mathLecture()

    println("value +3 :  "+mathClass.sumTest(4))
    println("average : "+mathClass.averageTest(Array(1,2,3,4,5)))
  }
}

실행 결과

value +3 :  7
average : 3

참고 사이트

https://docs.scala-lang.org/ko/tutorials/tour/traits.html.html

sequential collection 에 사용하기 유용한 함수

• zipWithindex

val days = Array("Sunday","Month","Tuesday","Wednesday","Thursday","Saturday")

days.zipWidthIndex.foreach{
    case(day,count) => println(s"$count is $day")
}

결과 값

0 is Sunday
1 is Monday
2 is Tuesday
3 is Wednesday
4 is Thursday
5 is Friday
6 is Saturday

• zip 두 리스트의 원소들의 쌍으로 이루어진 단일 리스트를 반환

  List(1, 2, 3).zip(List("a", "b", "c"))
  

결과 값

List[(Int, String)] = List((1,a), (2,b), (3,c))