Installing Spark Standalone to a Cluster

#출처

http://spark.apache.org/docs/latest/spark-standalone.html

□ Place pre-built versions of Spark (All Server)

  - To install Spark Standalone mode, you simply place a compiled version of Spark on each node on the cluster. 

  - 설치 위치에 파일 배포 ( /platform/package/spark )

    > cd /platform/package

    > tar -zxvf /platform/temp/spark/spark-2.1.0-bin-hadoop2.6.tgz

    > ln -s spark-2.1.0-bin-hadoop2.6 spark  

□ SSH KEY 설정 (All Server)

  - Master 와 Worker간 자동 SSH접속을 위해 Key를 생성하여 cluster간 통신이 가능하도록 설정한다.

  - Note, the master machine accesses each of the worker machines via ssh. 

    By default, ssh is run in parallel and requires password-less (using a private key) access to be setup. 

    If you do not have a password-less setup, 

    you can set the environment variable SPARK_SSH_FOREGROUND and serially provide a password for each worker.

  - 참조 : http://bigdata-architect.tistory.com/8

□ SPARK CONFIG (All Server)

  - sudo vim /etc/profile

    > SPARK_HOME="/platform/package/spark"

    > export SPARK_HOME

  - source /etc/profile

  - echo $SPARK_HOME

□ Cluster Launch Scripts(Master Server)

  - Spark Standalone Cluster를 시작 / 중지 하기 위하여 Master 서버에서 Launch Scripts를 사용한다. 

  - Slave Node 목록(conf/slaves) 을 작성한다. 

    > cd /platform/package/spark/conf

    > cp slaves.template slaves 

  - sbin/start-all.sh, sbin/stop-all.sh 로 Cluster를 시작 / 중지 한다. 

    > $SPARK_HOME/sbin/start-all.sh

    > $SPARK_HOME/sbin/stop-all.sh    

  - To launch a Spark standalone cluster with the launch scripts, 

    you should create a file called conf/slaves in your Spark directory, 

    which must contain the hostnames of all the machines where you intend to start Spark workers, one per line. 

    If conf/slaves does not exist, the launch scripts defaults to a single machine (localhost), which is useful for testing. 

  - Once you’ve set up this file, you can launch or stop your cluster with the following shell scripts, 

    based on Hadoop’s deploy scripts, and available in SPARK_HOME/sbin:

sbin/start-master.sh - Starts a master instance on the machine the script is executed on.

sbin/start-slaves.sh - Starts a slave instance on each machine specified in the conf/slaves file.

sbin/start-slave.sh - Starts a slave instance on the machine the script is executed on.

sbin/start-all.sh - Starts both a master and a number of slaves as described above.

sbin/stop-master.sh - Stops the master that was started via the bin/start-master.sh script.

sbin/stop-slaves.sh - Stops all slave instances on the machines specified in the conf/slaves file.

sbin/stop-all.sh - Stops both the master and the slaves as described above.

      Note that these scripts must be executed on the machine you want to run the Spark master on, 

 not your local machine.

  - You can optionally configure the cluster further by setting environment variables in conf/spark-env.sh. 

     Create this file by starting with the conf/spark-env.sh.template, 

     and copy it to all your worker machines for the settings to take effect. The following settings are available:

□ Spark Cluster 실행 확인

  - an interactive Spark shell 

    $SPARK_HOME/bin/spark-shell --master spark://migaloo01:7077

  - Master WEB UI 

    http://migaloo01:8080 , http://192.168.10.101:8080 

  - Worker WEB UI 

    http://migaloo02:8081 , http://192.168.10.102:8081 

    http://migaloo03:8081 , http://192.168.10.103:8081 

    http://migaloo04:8081 , http://192.168.10.104:8081 

    http://migaloo05:8081 , http://192.168.10.105:8081        

□ Launching Spark Application On YARN

  - spark shell 

    > $SPARK_HOME/bin/spark-shell  #실행됨

    > $SPARK_HOME/bin/spark-shell --master spark://migaloo01:7077     #실행 중 죽음

    > $SPARK_HOME/bin/spark-shell --master yarn --deploy-mode client  #실행안됨

    > $SPARK_HOME/bin/spark-shell --master yarn --deploy-mode cluster #지원안함   

  - spark application

    > hdfs dfs -ls /

    > hdfs dfs -mkdir /platform    

    > hdfs dfs -mkdir /platform/temp/

    > hdfs dfs -put /platform/temp/UserPurchaseHistory.csv /platform/temp

    > hdfs dfs -ls /platform/temp/

    # Hadoop Resource Manager에는표시되지 않지만 실제로는 처리

    > $SPARK_HOME/bin/spark-submit --class org.migaloo.spark.test.ScalaApp --master spark://migaloo01:7077 /platform/temp/org.migaloo.spark.test-0.0.1-SNAPSHOT.jar

    > $SPARK_HOME/bin/spark-submit --class org.migaloo.spark.test.ScalaApp /platform/temp/org.migaloo.spark.test-0.0.1-SNAPSHOT.jar

    > $SPARK_HOME/bin/spark-submit --class org.migaloo.spark.test.ScalaApp --master yarn --deploy-mode client /platform/temp/org.migaloo.spark.test-0.0.1-SNAPSHOT.jar

running spark application.txt

    # Hadoop Resource Manager에는 ACCEPTED 되나 실제 처리되지 않음

    > $SPARK_HOME/bin/spark-submit --class org.migaloo.spark.test.ScalaApp --master yarn --deploy-mode cluster --supervise  /platform/temp/org.migaloo.spark.test-0.0.1-SNAPSHOT.jar

    > $SPARK_HOME/bin/spark-submit --class org.migaloo.spark.test.ScalaApp --master yarn --deploy-mode cluster /platform/temp/org.migaloo.spark.test-0.0.1-SNAPSHOT.jar

□ 메모리 오류 발생 시 조치 

  - 메모리가 1G인 라즈베리에서 Spark Shell 실행 시 메모리 오류가 발생할 수 있다. 

  - 메모리에 저장할 공간이 모자랄 때 발생

  - memory overcommit을 허용해주면 된다.

  - sudo sysctl vm.overcommit_memory=1

  - 일반 커맨드 창에서 실행하면 됨

migaloo@migaloo01:/platform/package/spark$ ./bin/spark-shell --master spark://migaloo01:7077

Setting default log level to "WARN".

To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).

Java HotSpot(TM) Client VM warning: You have loaded library /platform/package/hadoop-2.6.5/lib/native/libhadoop.so.1.0.0 which might have disabled stack guard. The VM will try to fix the stack guard now.

It's highly recommended that you fix the library with 'execstack -c <libfile>', or link it with '-z noexecstack'.

17/01/22 10:52:35 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable

17/01/22 10:52:45 WARN DataNucleus.General: Plugin (Bundle) "org.datanucleus" is already registered. Ensure you dont have multiple JAR versions of the same plugin in the classpath. The URL "file:/platform/package/spark/jars/datanucleus-core-3.2.10.jar" is already registered, and you are trying to register an identical plugin located at URL "file:/platform/package/spark-2.1.0-bin-hadoop2.6/jars/datanucleus-core-3.2.10.jar."

17/01/22 10:52:45 WARN DataNucleus.General: Plugin (Bundle) "org.datanucleus.store.rdbms" is already registered. Ensure you dont have multiple JAR versions of the same plugin in the classpath. The URL "file:/platform/package/spark-2.1.0-bin-hadoop2.6/jars/datanucleus-rdbms-3.2.9.jar" is already registered, and you are trying to register an identical plugin located at URL "file:/platform/package/spark/jars/datanucleus-rdbms-3.2.9.jar."

17/01/22 10:52:45 WARN DataNucleus.General: Plugin (Bundle) "org.datanucleus.api.jdo" is already registered. Ensure you dont have multiple JAR versions of the same plugin in the classpath. The URL "file:/platform/package/spark/jars/datanucleus-api-jdo-3.2.6.jar" is already registered, and you are trying to register an identical plugin located at URL "file:/platform/package/spark-2.1.0-bin-hadoop2.6/jars/datanucleus-api-jdo-3.2.6.jar."

Java HotSpot(TM) Client VM warning: INFO: os::commit_memory(0x555c2000, 129228800, 0) failed; error='메모리를 할당할 수 없습니다' (errno=12)

#

# There is insufficient memory for the Java Runtime Environment to continue.

# Native memory allocation (mmap) failed to map 129228800 bytes for committing reserved memory.

# An error report file with more information is saved as:

# /platform/package/spark-2.1.0-bin-hadoop2.6/hs_err_pid2559.log

hs_err_pid2559.log

Spark Standalone: Differences between client and cluster deploy modes

#출처 

http://stackoverflow.com/questions/37027732/spark-standalone-differences-between-client-and-cluster-deploy-modes

http://spark.apache.org/docs/latest/submitting-applications.html

Client:

Driver runs on a dedicated server (Master node) inside a dedicated process. This means it has all available resources at it's disposal to execute work.

Driver opens up a dedicated Netty HTTP server and distributes the JAR files specified to all Worker nodes (big advantage).

Because the Master node has dedicated resources of it's own, you don't need to "spend" worker resources for the Driver program.

If the driver process dies, you need an external monitoring system to reset it's execution.

Cluster:

Driver runs on one of the cluster's Worker nodes. The worker is chosen by the Master leader

Driver runs as a dedicated, standalone process inside the Worker.

Driver programs takes up at least 1 core and a dedicated amount of memory from one of the workers (this can be configured).

Driver program can be monitored from the Master node using the --supervise flag and be reset in case it dies.

# Run on a Spark standalone cluster in cluster deploy mode with supervise
./bin/spark-submit \
  --class org.apache.spark.examples.SparkPi \
  --master spark://207.184.161.138:7077 \ # --master yarn
  --deploy-mode cluster \ # --deploy-mode client
  --supervise \
  --executor-memory 20G \
  --total-executor-cores 100 \
  /path/to/examples.jar \
  1000

Spark의 Streaming 처리 방식 - 마이크로 배치(micro-batch)

#출처 : Spark Streaming으로 유실 없는 스트림 처리 인프라 구축하기

http://readme.skplanet.com/?p=12465

실시간 처리(Real-time processing)와 스트림 처리(Stream processing)

실시간 처리는 데이터 처리의 목표 또는 제약 사항이라고 볼 수 있고, 

스트림 처리는 데이터 처리 방식이라고 볼 수 있습니다.

실시간이라는 용어에는 마감시각(deadline)이 있고, 마감시각 내에 주어진 연산을 완료하지 못하면 실패로 처리합니다. 조금 더 세분화하면 마감시각을 놓쳤을 때의 처리 결과에 따라 Hard/Firm/Soft real-time으로 구분합니다.      

스트림 처리는 범위가 한정되지 않고(unbounded) 끊임 없이 흘러가는(stream) 데이터에 대한 처리 방식입니다.

반대로 한정된(bounded) 데이터의 처리를 배치 처리(batch processing)라고 합니다. 

작은 배치 처리를 무한히 하는 방식도 스트림 처리에 포함되고, 이를 마이크로 배치(micro-batch)라고 합니다. 

잘 설계된 스트림 처리는 배치 처리를 포함한다고 까지 말할 수 있습니다.

결국, 스트림 처리가 끊임없이 흘러가는 데이터를 처리하다 보니 당연하게도 배치 처리에 비해 데이터 처리 결과를 빠르게 받아볼 수 있어서 실시간 처리라고 불릴 수도 있습니다. 

[중략...]

Spark Streaming의 선택

 스트림 처리를 신뢰도(reliability)에 따라 다음과 같이 세 가지 보장 방식으로 구분할 수 있습니다.

• At-most-once(최대 한 번): 데이터 유실이 있을 수 있어, 추천하지 않는 방식
• At-least-once(적어도 한 번): 데이터 유실은 없으나 재전송으로 인해 중복이 생길 수 있음. 대부분의 경우 충분한 방식
• Exactly-once(딱 한 번): 데이터가 오직 한 번만 처리되어 유실도 중복도 없음. 모든 상황에 대해 완벽히 보장하기 어렵지만 가장 바라는 방식

Core Storm과 Spark Streaming을 비교하면 Spark Streaming이 exactly-once를 보장하기 때문에 조금 더 믿을만하지만, 마이크로 배치처럼 동작하도록 추상화 계층을 추가한 Storm Trident에 비해서 Spark Streaming은 아직 부족한 점이 많았습니다. 

[중략...]

스트림 처리 프레임웍으로 선택한 Spark Streaming을 간단히 정리하면, 다음 그림과 같습니다.

그림 4. Spark Streaming 흐름
출처: http://spark.apache.org/docs/latest/streaming-programming-guide.html

끊임없이 들어오는 스트림 데이터를 Spark Streaming이 배치 간격(batch interval)마다 데이터를 나누고, 나눠진 배치 데이터를 Spark Core 엔진이 처리해서 배치 간격마다 결과를 내놓습니다. 즉, 스트림 처리를 작은 시간 간격을 갖는 배치 처리의 연속으로 전환하여 처리합니다. 이러한 방식을 마이크로 배치(micro-batch) 방식이라고 합니다. 또한 각각의 배치 처리는 Spark으로 분산 처리하기 때문에, 좋은 성능을 가지면서도 장애 복구가 가능한 Spark의 장점들을 모두 이어 받았습니다.

Spark Streaming이 마이크로 배치 방식을 택한 이유는 배치 처리용 Spark이 먼저 개발된 후에 Spark의 다양한 기능을 그대로 이용하면서 스트림 처리가 가능하도록 추가한 기능이 Spark Streaming이기 때문입니다. 초기 Spark Streaming의 설계는 Discretized Streams: Fault-Tolerant Streaming Computation at Scale 논문에서 확인할 수 있습니다.

마이크로 배치 방식은 continuous 처리 방식의 스트림 처리 시스템보다 latency가 큰 단점이 있지만, 배치 처리와 스트림 처리를 하나의 시스템에서 처리할 수 있는 장점이 되었습니다. 기존의 continuous 처리 방식과의 차이점은 https://databricks.com/blog/2015/07/30/diving-into-spark-streamings-execution-model.html 에서 조금 더 살펴볼 수 있습니다.

배치 처리와 스트림 처리를 하나의 시스템에서 코드를 재활용해서 사용할 수 있기 때문에, 최근 대용량 데이터 처리에서 많이 언급되는 람다 아키텍처(Lambda Architecture)에 적합한 방식이라고 할 수 있습니다. 람다 아키텍처에서는 배치 계층(batch layer)과 속도 계층(speed layer)이 따로 존재하는데, Spark Streaming은 람다 아키텍처를 하나의 시스템에서 구현할 수 있어, 한 걸음 더 나아간 방식이라고 할 수 있습니다.

How to set up a Spark project with Scala Eclipse IDE Maven

#출처 동영상 

https://www.youtube.com/watch?v=aB4-RD_MMf0

- Scala Eclipse IDE를 다운로드 받는다. 

  . http://scala-ide.org/download/sdk.html

- Eclipse에서 새로운 Maven Project를 생성한다.

  . File > New > Other > Maven > Maven Project

  . [New Maven Project] Create a simple project 옵션 체크 

  . GroupId, ArtifactId 입력 ex) org.migaloo.spark.test

- pom.xml 에 Spark 관련 설정을 추가한다. 

  . [출처] https://github.com/mcapuccini/spark-cp/blob/master/cp.examples/pom.xml

  . 실행할 Spark 버전에 맞게 pom.xml을 수정한다. 

  . 참고 : pom.xml

#Erro Case. NoSuchMethodError: org.apache.spark.SparkContext.ui()Lscala/Option

- Project에 Scala Nature 추가 

  . org.migaloo.spark.test Project 에 마우스 오른쪽 클릭 > Configure > Add Scala Nature 

- Project의 Scala Compiler 설정 변경 

  . org.migaloo.spark.test Project 에 마우스 오른쪽 클릭 > Scala Compiler

  . Use Project Settings 체크 

  . Scala Installation : Fixed Scala Installation 2.10.4(built-in) 선택 후 저장

- Project의 Maven Update 실행

  . org.migaloo.spark.test Project 에 마우스 오른쪽 클릭 > Maven > Update Maven Project

- Project의 Scala Library Container 삭제

  . org.migaloo.spark.test Project 에 마우스 오른쪽 클릭 > Properties > Java Build Path > Libraries Tab 

  . Scala Library Container 선택 후 Remove 버튼을 눌러 삭제한다. 

- Project의 폴더 명칭 변경

  . 폴더 선택 후 마우스 오른쪽 클릭 > refactor > rename

  . src/main/java -> src/main/scala

  . src/test/java -> src/main/scala

- 패키지 및 스칼라 파일 생성 후 main 함수 추가, Run As > Maven Install로 최종 jar 파일 생성 

- Spark에서 실행하여 결과 확인 

  . ./bin/spark-submit --class org.migaloo.spark.test.ScalaApp /platform/temp/org.migaloo.spark.test-0.0.1-SNAPSHOT.jar

#Project Sample

org.migaloo.spark.test.zip