Spark Upgrade 및 성능 비교 - Spark 1.1.0 vs Spark 1.2.0 ( on Apache Mesos )

Spark 1.1.0 이 작년 9월 경 버전업 된 이후로 1.1.1 업그레이드가 한번 더 있었고, 1.2.0 업그레이드도 최근에 있었다. Apache Mesos 와의 연동 작업도 세팅할 겸. Production 의 Spark 버전을 업그레이드 해 주었으며, 업그레이드 전 후 성능 비교도 해 보았다.

1. Spark Download
1. wget http://mirror.apache-kr.org/spark/spark-1.2.0/spark-1.2.0-bin-hadoop1.tgz
2. tar xvzf spark-1.2.0-bin-hadoop1.tgz
2. Library Upload
1. hadoop dfs -mkdir /user/spark
2. hadoop dfs -put ./spark-1.2.0-bin-hadoop1 /user/spark/
3. Apache Mesos 설치
1. Apache Mesos 설치는 [Apache Mesos 설치기] 요기에서....
4. Config Setting on Master
1. mv spark-env.sh.template spark-env.sh
2. vi spark-env.sh
   
   MESOS_NATIVE_LIBRARY=/data01/mesos/mesos-0.18.1/build/src/.libs/libmesos.so
   
   SPARK_EXECUTOR_URI=hdfs://master001.prod.moneymall.ssgbi.com:9000/user/spark/spark-1.2.0-bin-hadoop1.tgz
3. vi spark-defaults.conf
   spark.executor.uri              hdfs://master001.prod.moneymall.ssgbi.com:9000/user/spark/spark-1.2.0-bin-hadoop1.tgz
5. Spark Shell 실행

1. 먼저 shell 모드 실행
2. 위 처럼 1.2.0 버전으로 잘 떳음...
6. Spark Program 구동

1. 이어서 기존에 1.1.0 에서 돌던 Scala Batch 프로그램을 1.2.0 에서 수행하기 위해 위 처럼 인자를 바꾸고 수행.
2. executor memory 와 driver memory 값을 적절하게 주어야 Heavy 한 Batch Job 이 원할하게 수행 됨. 속도도 더 빨라 짐.
7. Spark 1.1.0 수행 속도 : 61초 

8. Spark 1.2.0 ( on Apache Mesos ) 수행 속도 : 35초

9. Spark 1.2.0 on Apache Mesos 업그레이드 고찰
   (1)   기존 JVM Spark 엔진은 설정 공유 메모리 만큼 JVM 데몬이 메모리를 점유 하고 있어서, Hadoop Job 과 Memory 경합이 있을 수 있음. 경합을 피하기 위해 OS Full Memory 를 쓰지 못하고, 보수적으로 Fix 점유 해놓은 메모리 만큼 만 공유하는 구조였음.
   Ø  바뀐 Mesos 는 C++ 엔진인지라, 별도로 Memory 할당이 필요 없고, OS 가 가지고 있는 물리 Memory를 Full 로 쓸 수 있음.
   Ø  공유 메모리 Pool 이 56G 에서 약 280G 정도로 5배 가까이 커짐.
   (2)   속도 향상.
   Ø  Memory 공유 방식 뿐만 아니라 CPU 공유 메커니즘도 포함 되면서, 위 결과 처럼 약 1.8배의 속도 향상이 있었음.

Scala - vim Syntax Highlight

Vi에서 Scala Syntax Highlight 사용하기.

mkdir -p ~/.vim/{ftdetect,indent,syntax} && for d in ftdetect indent syntax ; do curl -o ~/.vim/$d/scala.vim https://raw.githubusercontent.com/derekwyatt/vim-scala/master/syntax/scala.vim; done

Apache Mesos 설치기 1 - 설치 및 활용성에 대한 고찰

한때 메모리를 클러스터링 하여, NFS 공유 파일시스템 쓰듯이 공유 메모리 Pool 을 쓸 수 있게 해주는 Tachyon (http://tachyon-project.org)에 관심을 갖은 적이 있었다. (결과 적으로 몇번의 Deep 한 테스트 이후 현재는 Tachyon 을 쓰고 있진 않지만, IO 를 많이 쓰는 공유 저장소가 필요할 때, 여러 Layer 에서 손쉽게 쓸 수 있는 좋은 대안이라는 생각에는 변함이 없다. NFS 처럼 Mount 될 수 있으며, NFS 보다 수십 배에서 수백 배의 성능을 얻을 수 있다.)

Spark 엔진 또한 Tachyon 과 Collaboration 하여 Batch 성능을 향상해보려는 테스트를 한바 있었는데…. 그때가 작년 8~9월 이었다.

그로부터 3~4개월이 흘렀고….그 짧은 기간 동안에 Spark 는 3번의 버전업이 있었다.

최근에 Spark 진영은 Tachyon 과의 Collaboration 보다는 Hadoop 2.0 Yarn 과의 Collaboration , 그리고 Apache Mesos 와의 Collaboration 을 하는 방향으로 유행이 바뀌어 있다.

아래는 Apache Mesos 가 무엇인지 와 닿게 하는 기사 글 이다.

http://www.itworld.co.kr/news/89034

특징을 한마디로 요약하자면….

Tachyon 은 메모리를 공유해준것에 그쳤다면…

Mesos 는 CPU와 memory 와 storage 와 기타 다른 컴퓨트 리소스들을 추상화 하여 하나의 pool 로서 쓸 수 있게 해준 다는 것이다.

C++, Java , Python ,Scala 등의 어플리케이션을 Mesos가 제공하는 Framework 위해서 구동할 수 있으며, Hadoop , Spark , ElasticSearch 그리고 Jenkins 까지도, mesos 와 Collaboration 구동되게 데몬 자체를 Mesos Layer 위에 구동 설치가 가능하다.(이 경우 기존 클러스터링 설치보다 더 뛰어난 성능을 얻을 수 있다.)

아래는 필 받은김에 이번 주말에 설치해본 간략 설치기 이다.

1. Installing Mesos
• Spark 1.2.0 is designed for use with Mesos 0.18.1 
• http://mesos.apache.org/
• wget http://archive.apache.org/dist/mesos/0.18.1/mesos-0.18.1.tar.gz
• tar xvzf mesos-0.18.1.tar.gz
• root 계정으로 vi /etc/yum.repos.d/wandisco-svn.repo
• [WandiscoSVN] name=Wandisco SVN Repo baseurl=http://opensource.wandisco.com/centos/6/svn-1.8/RPMS/$basearch/ enabled=1 gpgcheck=0
• yum groupinstall -y "Development Tools"
• yum install -y python-devel zlib-devel libcrul-devel openssl-devel cyrus-sasl-devel cyrus-sasl-md5 apr-devel subversion-devel
• Java 및 maven계열 설치는 기 설치되어 있어서 skip....
• cd mesos홈
• ./bootstrap ( root 계정에서 일반 계정으로 돌아옴. & sudo 활용)
• mkdir build
• cd build
• ../configure
• make ( 요 단계 꽤 오래 걸림...., sudo 도 안되고 su 가 필요한 경우도 있음. )
• make check
• make install ( sudo 가 필요한 경우가 있었음.. )
• 위 과정을 Master 역할을 할 서버에 설치
• 위 과정을 Slave 역할을 할 전체 서버에 모두 설치
2. Mesos Cluster 구동
1. Master 구동 
1. ./bin/mesos-master.sh --ip=서버IP --work_dir=/dataXX/mesos &
2. Slave 구동
1. ./bin/mesos-slave.sh --master=서버IP:5050

Elasticsearch 사용기 1 - 리얼타임 분석엔진입장에서 elasticSearch vs Strom

elasticSearch 는 빅데이타 에코시스템의 부족한 부분 중 상당히 많은 부분을 체워 줄 수 있다. 물론 이 부분을 다른 여러 가지로 나누어 메꾸거나, Storm 이라는 걸쭉한 리얼타임 엔진으로 대부분을 커버 할 수도 있긴 하다.

그러나, storm 을 기 사용하고 있었고, elasticSearch 를 짧은 시간이나마 다시 운영 Hadoop에 적용하여, 이리 저리 실험적으로 사용해본 첫 느낌으로는.....

Real Time Analytics 에 있어 Storm 과 elasticSearch 는,
마치 Java Map/Reduce (on Spring Hadoop) 와 REPL 콘솔을 갖춘 Scala In-Memory M/R (on Spark) 의 차이와 같은 느낌이라고나 할까....(비유가 더 어려웠나???)

여튼 Storm 은 멀 할라 치믄 일단, 할게 많고, 공들여 개발한 다음 배포하고, 데이타를 원하는 포맷으로 이리 저리 필터 걸거나 조건을 넣어, 실제로 드려다 보기까지 귀찮은 작업이 이만 저만이 아니다... 그리고 원하는 데로 결과가 나오지 않아 검증이라도 하고싶거나, 혹은 뭔가 결과가 나오긴 했는데, 과연 맞게 나온 결과인지, 런타임 디버깅이라도 해보고 싶다면, 여간 난감한게 아니다. 그래서, 상용 솔루션이긴 하지만, 에스퍼(http://esper.codehaus.org/)를 Collaboration 을 해서 쓰거나, 아예 좀더 애자일 하게 Storm을 쓰기 위해 clojure  와 trident 기반으로 Storm 을 사용하는 경우(우리처럼...)도 많이 봐왔다.

elasticSearch 는 일단, 여러가지 plugin 중 head 하나만 가지고도  할수 있는게 상당히 많다. 코딩을 하기 전, 데이타를 이리저리 만지작 거리며, 제대로된 모델을 사전 검증하거나, 중간 단계 슈도 코드로 Real-Data 를 통하여 검증하는 것을 쉽게 할 수 있다는 장점이 있다.

검증을 REPL 에서 바로바로 할 수 있는 Scala on Spark 와 비슷한 느낌이라고나 할까... 게다가 결과도 Spark 이상으로 빠르다. 사실 간단한 로직의 경우는 훨씬 더 빠르다. 물론 리얼타임 검색 엔진이니까..빠른게 당연하지만...

단점이 없는 것은 아니다. elasticSearch 엔진의 queryString 은 복잡한 비지니스 요건을 전부 처리하는데에는 한계가 있다. 즉, 요녀석도 그냥 NoSQL 처럼 에코시스템 중 One of Them 인것이다.
즉, elasticSearch 는 여러 에코시스템 중 elasticSearch 가 장점이 있는 부분에 요긴하게 쓸수 있으며, 그 부분이 다른 NoSQL 에 비하여 RealTime 활용 쪽에서는 상당히 넓은 부분을 커버 할 수 있으며, plugin등 자체 에코시스템도 상당히 갖추어져 있어 응용 확장성도 뛰어나다.. 정도로 요약 가능할 듯 하다. 게다가 kibana 라는 걸쭉한 시각화 도구 까지...

아래는 SQL 과 elasticSearch  의 queryString 을 비교 수행해본 결과이다.

1. SQL 쿼리

• SELECT COUNT(*) FROM log GROUP BY browser ORDER BY COUNT(*) DESC

2. ElasticSearch 쿼리

• ElasticSearch 의 쿼리는 아래와 같다.

{

  "size": 0,

  "aggs": {

    "group_by_state": {

      "terms": {

        "field": "browser"

      }

    }

  }

}

결과는 아래처럼 나온다. 놀라운 것은 이것이 1초전 Insert 된 Real Time 이벤트까지 반영된 실시간 결과라는 점!!!!!!

3. 아래 처럼 쿼리를 좀 확장해서 쓸수도 있다. 정규표현식으로 필터를 그룹 주고, 그 결과로 그룹핑 된 PV  정도는 손쉽게 리얼타임 장표를 만들어 낼 수 있다.
(여기서의 리얼타임이란, 시간당 수백만 PV가 들어오는 트래픽에 대하여, 고객들이 들어오는 특정 페이지의 PV  를 1~2초 안의 최신성을 반영하여 시각화 가능하다 정도...사실 Scale Out 이 가능하므로, 트래픽이 큰 문제가 되지는 않는다.)

GET /test/data/_search?search_type=count{
  "aggs" :{
    "total_count" : {
      "global" : {}
    },
    "XX999" : {
      "filter" : {
        "regexp":{
            "product_code" : "[a-z]{2}[0-9]{3}"
        }
      } 
    },
    "X99999" : {
      "filter" : {
        "regexp":{
            "product_code" : "[a-z]{1}[0-9]{5}"
        }
      } 
    },
    "no_format" : {
        "missing" : {
            "field" : "product_code"   
        }
    }
  }
}

Spark 사용기 5 - SparkSQL 실무 활용기

Spark 를 실무에서 사용하기 시작한지 어언 2달...
아직까지는 Heavy Batch 일부와 Adhoc 개발 부분에서만 사용중이지만, 쓰면 쓸수록, 그 압도적인 성능에 감탄을 금할 수 없다.

물론 시행착오가 없었던 것은 아니다. 실무 사용하기 전 검증 단계에서 사용했던, Spark + tachyon + Shark 는  Memory 관련 에러가 빈번했었고, Public 가상화 환경에서의 태스트 이긴 하였으나, Spark + Yarn 의 태스트는 Spark StandAlone 에 비하여 성능이 더 떨어지는 현상도 접한 바 있다. ( Yarn 테스트는 아직 충분히 되지 않았다. Production 환경에서 여러가지 설정을 바꾸어가며 좀더 지켜봐야 하는 부분이다.)

오늘은 Spark + Scala M/R + SparkSQL 을 실무 사용하며, 발견한 SparkSQL 의 미숙함에 대하여 블로그 해보겠다.

우선 최근에 접했던 것은 Not IN 요건.
결과부터 말하자면, SparkSQL 은 Not IN 을 아직 지원하지 않는다. 물론, 간단한 요건이면, 로딩하는 순간에 filter 로 제외 시키고 불러 올 수 있다. 그렇게 처리 가능한 루틴은 매우 빠르게 Not IN 처리가 가능하다. 단, 아래 SQL 처럼 무언가 요건이 내포된 경우의 Not IN 은 Filter 로 처리하기 힘든 요건이다.

우선 아래 쿼리는 SparkSQL 이 지원하지 않는다는 것이 다소 놀랍지는 않았다. Hive 도 아래 쿼리를 지원하지 않음을 기 알고 있었기 때문이다. (hive0.13 이상은 지원하는지 체크해보지는 않았다... 우리 Production 기준으로 Hive 도 아래 Not IN 쿼리를 지원하지 않는다.)

[1] 첫번째 시도.

         val query3 = """
                   select
                        count(distinct pcId)
                   from
                        everyClickFlat_PC
                   where
                        mbrId = '' and

                        url like '%s' and

                        pcId not in (select distinct pcId from everyClickFlat_PC where mbrId != '' and url like '%s')
                   """.format(classRow.pc_url,classRow.pc_url)
          val results3 = sqlContext.sql(query3)

이렇게 수행하면 아래와 같은 에러가 발생한다.

java.lang.RuntimeException: [9.38] failure: string literal expected
                        pcId not in (select distinct pcId from everyClickFlat_PC where mbrId != '' and url like 'http://www.ssg.com/')
                                     ^
        at scala.sys.package$.error(package.scala:27)
        at org.apache.spark.sql.catalyst.SqlParser.apply(SqlParser.scala:60)
        at org.apache.spark.sql.SQLContext.parseSql(SQLContext.scala:73)
        at org.apache.spark.sql.SQLContext.sql(SQLContext.scala:260)
        at $iwC$$iwC.inline(<console>:220)
        at $iwC$$iwC$$anonfun$dayJob$1.apply(<console>:80)
        at $iwC$$iwC$$anonfun$dayJob$1.apply(<console>:80)
        at scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
        at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:108)
        at $iwC$$iwC.dayJob(<console>:80)
        at $iwC$$iwC$$anonfun$1.apply$mcVI$sp(<console>:56)
        at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:141)
        at $iwC$$iwC.<init>(<console>:41)
        at $iwC.<init>(<console>:345)
        at <init>(<console>:347)
        at .<init>(<console>:351)
        at .<clinit>(<console>)
        at .<init>(<console>:7)
        at .<clinit>(<console>)
        at $print(<console>)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:606)
        at org.apache.spark.repl.SparkIMain$ReadEvalPrint.call(SparkIMain.scala:789)
        at org.apache.spark.repl.SparkIMain$Request.loadAndRun(SparkIMain.scala:1062)
        at org.apache.spark.repl.SparkIMain.loadAndRunReq$1(SparkIMain.scala:615)
        at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:646)
        at org.apache.spark.repl.SparkIMain.interpret(SparkIMain.scala:610)
        at org.apache.spark.repl.SparkILoop.pasteCommand(SparkILoop.scala:782)
        at org.apache.spark.repl.SparkILoop$$anonfun$standardCommands$8.apply(SparkILoop.scala:309)
        at org.apache.spark.repl.SparkILoop$$anonfun$standardCommands$8.apply(SparkILoop.scala:309)
        at scala.tools.nsc.interpreter.LoopCommands$LoopCommand$$anonfun$nullary$1.apply(LoopCommands.scala:65)
        at scala.tools.nsc.interpreter.LoopCommands$LoopCommand$$anonfun$nullary$1.apply(LoopCommands.scala:65)
        at scala.tools.nsc.interpreter.LoopCommands$NullaryCmd.apply(LoopCommands.scala:76)
        at org.apache.spark.repl.SparkILoop.command(SparkILoop.scala:766)
        at org.apache.spark.repl.SparkILoop.processLine$1(SparkILoop.scala:616)
        at org.apache.spark.repl.SparkILoop.innerLoop$1(SparkILoop.scala:624)
        at org.apache.spark.repl.SparkILoop.loop(SparkILoop.scala:629)
        at org.apache.spark.repl.SparkILoop$$anonfun$process$1.apply$mcZ$sp(SparkILoop.scala:954)
        at org.apache.spark.repl.SparkILoop$$anonfun$process$1.apply(SparkILoop.scala:902)
        at org.apache.spark.repl.SparkILoop$$anonfun$process$1.apply(SparkILoop.scala:902)
        at scala.tools.nsc.util.ScalaClassLoader$.savingContextLoader(ScalaClassLoader.scala:135)
        at org.apache.spark.repl.SparkILoop.process(SparkILoop.scala:902)
        at org.apache.spark.repl.SparkILoop.process(SparkILoop.scala:997)
        at org.apache.spark.repl.Main$.main(Main.scala:31)
        at org.apache.spark.repl.Main.main(Main.scala)
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.lang.reflect.Method.invoke(Method.java:606)
        at org.apache.spark.deploy.SparkSubmit$.launch(SparkSubmit.scala:328)
        at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:75)
        at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

[2] 두번째 시도



그냥 시도 이다. 쿼리가 정확하게 Not In 과 동일한 원하는 결과물을 가져다 줄지 검증은 아래처럼 너무 오래걸려서 검증해보지도 못하고 kill 하였다...

 val query3 = """
                   select
                        count(distinct A.pcId)
                   from
                        everyClickFlat_PC as A, everyClickFlat_PC as B
                   where
                        A.mbrId = '' and

                        A.url like '%s' and

                        B.mbrId != '' and

                        B.url like '%s' and

                        A.pcId != B.pcId

                   """.format(classRow.pc_url,classRow.pc_url)
          val results3 = sqlContext.sql(query3)

이렇게 했더니, 에러는 안나지만.... 무쟈게 오래 걸린다. 5분정도 기다리다가 kill..

[3] 세번째 시도
이번 역시 그냥 시도이다.. Not In 과 비스무리한 결과라도 얻을라고 했던 Just 시도 이다..

         val query3 = """
                   select
                        count(distinct A.pcId)
                   from
                         (select distinct pcId from everyClickFlat_PC where mbrId = '' and url like '%s') as A, 

                         (select distinct pcId from everyClickFlat_PC where mbrId != '' and url like '%s') as B

                   where

                        A.pcId != B.pcId

                   """.format(classRow.pc_url,classRow.pc_url)
          val results3 = sqlContext.sql(query3)

이 역시 무쟈게 오래 걸린다.. 이 역시 8분 정도 기다리다가 kill...

[4] 네번째 시도

사실 Not In 은 Left Semi Join 을 이용하면 동작할것임은 어느정도 예상하고는 있었다. Not In 을 지원하는 RDBMS 에서 조차, 성능상 이유로, Not In 보다는 semi join 을 많이 사용하곤 하므로...
left semi join 사용 했더니 약 5초만에 결과가 나왔다. 참고로 저 테이블은 300Gb 가 넘는 크기의 테이블이다. (물론 성능 극대화를 위하여, 파티션 처리를, 먼저 전처리 해두었다.)

        val query3_3 = """
                   select
                          count(distinct results3_1.pcId)
                   from

                         results3_1 left semi join results3_2 on results3_1.pcId != results3_2.pcId

                   """
          val results3_3 = sqlContext.sql(query3_3)

          pc_uv2 = results3_3.collect()(0)(0).toString.toDouble

[5] 결론

물론 이번에는 실험해보지 않았지만, SparkSQL 을 우회해서 사용하지 않고, Scala 로 그냥 구현하면 Not In 을 훨씬 빠르게 구현할 수 있을 것이나, 300Gb 가 5초 정도면, SparkSQL 도 쓸만한 속도이다... 시도해보진 않았지만, 같은 쿼리를 Hive 환경에서 테스트 했다면, 수분 혹은 수십분도 걸릴 수 있었을 것이다.

Hive 나 SparkSQL 이 RDBMS 에 비하여 지원하지 않는 아쉬운 기능 중, 실무에서 제일 쉽게 접하는 것중 첫번째 정도는 inline query 가 아닐 까 한다. 두번째 정도가 Not In 일것 같다. inline 쿼리는 scala 로 처리하면, 매우 빠르게 구현이 가능했고, Not In 은 SparkSQL 이 직접 지원하진 않지만, Left Semi Join 이 대안 정도로 쓸만 한것 같다가 오늘의 결론이다...

Elasticsearch 설치기 1 - 기본 설치 & plug-in 설치

Elasticsearch 설치는 매우매우 쉽다.

(1) 우선 공식 홈페이지에서 zip 파일을 다운로드 받는다.

http://www.elasticsearch.org/overview/elkdownloads/

(2) 그리고 압축을 푼다.

(3) 그리고 아래처럼 실행하면 끝.... 사실 아래처럼 백그라운드 실행을 할 필요는 없다. 그러나, 최초 Getting Started 데로, 실행시 프롬프트로 빠져나가지 않았고, 프롬프트로 돌아오면 데몬이 종료됨을 발견하여, 2번째 시도에서는 아래처럼 Ctrl+z 로 나와서 backgroud 프로세스로 돌려 주었었다. 물론 그렇게 하여도, 데모나이즈 화가 잘되는 것 같았지만, 좀더 자료를 찾아보니, -d 옵션을 주면 데몬화 실행이 됨을 알 수 있었다..

백그라운드 실행 - 이렇게 할 필요는 없다.

-d 옵션을 주면 데모나이즈 실행 모드 이다.

(4) 설치 확인은 아래처럼 Restful 접근을 해보면 가능하다.

(5) 뭔가 불안하리 만큼 너무 설치가 간단하다. Web-Console 이라도 띄워보기 위해 플러그인을 뒤져 보았다. marvel 이라고 하는 걸쭉한 모니터링 도구가 바로 검색 되었다. 아래처럼 plug-in 설치도 매우 쉽다. ( marvel 은 이름이 생소하였지만, 설치하고 나서 알고보니 kibana 위에 구성된 3rd party kibana 활용 dashboard 였다.. 요즘 뜨고 있는 ELK 의 K )

marvel 설치 후에는 elasticsearch 재실행이 필요하다.

(6) marvel web console 화면은 아래와 같은 주소로 띄울 수 있다. 9200 으로 방화벽 신청이 필요할 수 있다.

http://설치서버주소:9200/_plugin/marvel/
-> kibana dashboard 페이지로 direction 된다.

일단.... 오늘은 여기까지....
이제 좀 놀다가 자야겠다....

회의는 회사에서... R&D 는 집에서....

Spark 사용기 4 - cache() 쓰고 안쓰고 비교

최근에 만든 Spark 위의 Scala M/R 배치는 3개 HDFS File 을 가지고 연산을 수행하는데, 모두 Memory 에 로딩 가능한 정도의 크기이다. 이 경우 RDD 를 Cache 할 필요가 있을까?

그래서 실험을 해 보았다.

(1) Cache 안썻을 때
-> 14초

(2) Cache 썼을 때
-> 1차 시도: 45초
-> 2차 시도: 32초

1차 시도를 했을때 , 45초나 걸려서 다소 놀랐었다. 그때는 뭔가 효율적으로 자료구조를 메모리 DB 화 하기 위해 복잡한 insert 루틴을 수행하고 있나 보다 했었다. 그런데, 그런 논리라고 한다면, 2차 시도때 훨씬 빨라 져야 정상일 것이다.
그런데, 결과는 약간 빨라지긴 했지만, 그다지 안썻을 때에 비해 전혀 개선된 것 같지가 않았다.

(3) cache() 했을 때 값이 남아 있는지 여부?

Shell 콘솔을 껏다 키면, 살짝 간격을 두고 다 없어진다. 그럼 뭔차이일까?
오히려 더 느리기만 하고.... (persist 를 쓰면, keeping 이 되긴 하겠지만...)

일단 짐작을 해 보자면, 클러스터를 Yarn 기반으로 구동 시키면, MR 코드가 Yarn 의 Job 을 수행시켰었다. 이 경우 Cache 를 적절하게 써주면, HDFS 가 아닌 Spark Memory 를 활용하며, 좀더 효율적인 연산을 하지 않을 까 기대했었다...  그 시나리오를 실험해보자, 경미하게 속도 개선이 있었지만, 역시 큰 개선은 없었다. 그러나, Job 성격에 따라 성능 향상이 많은 경우도 있을 것이다. 일단, Cache 는 그런 용도라고 짐작하고 넘어가야겠다.

Yarn 모드가 아닌 Spark Engine 자체로 Master 와 Slave 노드를 설정하고 Memory Instance 를 띄워서 Memory Cluster 를 구동한 경우에는 Memory Cluster 에 로딩 가능한 크기인 경우, 굳이 Cache 를 쓰지 않아도 메모리에서 모든 연산이 이루어지는 듯 하다. 아니, 위 실험 처럼 오히려 속도가 빠르다....
즉, 그러한 환경에서는 ( Stanalone Master & Slave Cluster 모드) Cache 를 쓰지 않는게 더 효율적인듯 하다. 가끔 pinning 할때만, persist 기능을 적절히 써주면 될듯 하다.

ps... persist 를 적절히 쓰면 10배 이상 성능이 개선된다고 한다... 아래 메뉴월 원문을 paste 해본다.  이런실험은 좀더 다양한 M/R 을 짜보고 종합적으로 실험해 보아야 할듯 하다. 일단, 오늘 코딩한 소스로는 아무것도 안쓴게 젤로 빠르다...

RDD Persistence

One of the most important capabilities in Spark is persisting (or caching) a dataset in memory across operations. When you persist an RDD, each node stores any partitions of it that it computes in memory and reuses them in other actions on that dataset (or datasets derived from it). This allows future actions to be much faster (often by more than 10x). Caching is a key tool for iterative algorithms and fast interactive use.

You can mark an RDD to be persisted using the persist() or cache() methods on it. The first time it is computed in an action, it will be kept in memory on the nodes. Spark’s cache is fault-tolerant – if any partition of an RDD is lost, it will automatically be recomputed using the transformations that originally created it.

In addition, each persisted RDD can be stored using a different storage level, allowing you, for example, to persist the dataset on disk, persist it in memory but as serialized Java objects (to save space), replicate it across nodes, or store it off-heap in Tachyon. These levels are set by passing a StorageLevel object (Scala, Java, Python) to persist(). The cache() method is a shorthand for using the default storage level, which isStorageLevel.MEMORY_ONLY (store deserialized objects in memory). The full set of storage levels is:

Storage Level	Meaning
MEMORY_ONLY	Store RDD as deserialized Java objects in the JVM. If the RDD does not fit in memory, some partitions will not be cached and will be recomputed on the fly each time they're needed. This is the default level.
MEMORY_AND_DISK	Store RDD as deserialized Java objects in the JVM. If the RDD does not fit in memory, store the partitions that don't fit on disk, and read them from there when they're needed.
MEMORY_ONLY_SER	Store RDD as serialized Java objects (one byte array per partition). This is generally more space-efficient than deserialized objects, especially when using a fast serializer, but more CPU-intensive to read.
MEMORY_AND_DISK_SER	Similar to MEMORY_ONLY_SER, but spill partitions that don't fit in memory to disk instead of recomputing them on the fly each time they're needed.
DISK_ONLY	Store the RDD partitions only on disk.
MEMORY_ONLY_2, MEMORY_AND_DISK_2, etc.	Same as the levels above, but replicate each partition on two cluster nodes.
OFF_HEAP (experimental)	Store RDD in serialized format in Tachyon. Compared to MEMORY_ONLY_SER, OFF_HEAP reduces garbage collection overhead and allows executors to be smaller and to share a pool of memory, making it attractive in environments with large heaps or multiple concurrent applications. Furthermore, as the RDDs reside in Tachyon, the crash of an executor does not lead to losing the in-memory cache. In this mode, the memory in Tachyon is discardable. Thus, Tachyon does not attempt to reconstruct a block that it evicts from memory.