EC2-Spark

여러 개의 Spark 를 설치하는 경우 클러스터 매니저 설정을 해야 한다.

실제 업무에서는 대부분 zepplin(jupyter notebooks)을 이용해서 사용

리눅스에서 코딩을 할 때는 wget 이나 git clone 또는 파일 전송하는 방법을 숙지하는 것이 중요함.

파일 사이즈가 큰 경우 파일 전송하는 방법을 사용함.

1. Python(anaconda), java, scala 설치: java 와 scala 설치는 필수
2. spark(java로 만들어짐) 설치
3. pyspark 설치

# 접속
sudo ssh -i [키 경로] [운영체제]@[Public DNS Address]

anaconda 다운로드

# 아나콘다 설치
wget https://repo.anaconda.com/archive/Anaconda3-2021.11-Linux-x86_64.sh

# 아나콘다 설치
bash Anaconda3-2021.11-Linux-x86_64.sh

# 설치하고 난 후 환경 변수 추가
vi ~/.bashrc
export PATH=/home/ubuntu/anaconda3/bin:$PATH

# 수정 내용 반영
source ~/.bashrc

jupyter notebook 환경 설정

jupyter notebook --generate-config

vi ~/.jupyter/jupyter_notebook_config.py

c.NotebookApp.allow_root = False
c.NotebookApp.ip='0.0.0.0'

source ~/.bashrc

주피터 노트북 실행: jupyter notebook

1. EC2 보안 그룹 내 8888 포트를 외부 접속 허용
2. 접속 후 토큰 번호 입력 서버 시 보여지는 토큰 번호 입력

java, scala 설치

# java 설치
sudo apt-get install -y openjdk-8-jdk

# 확인
java -version
javac -version

# 환경 변수 설정
vi ~/.bashrc

# Java set
export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64 
export PATH=$JAVA_HOME/bin:$PATH
export CLASS_PATH=$JAVA_HOME/lib:$CLASS_PATH

source ~/.bashrc

# 프로그래밍 언어 설치 Scala 설치
sudo apt install scala

# 환경 변수 설정
vi ~/.bashrc

# Scala set
export SCALA_HOME=/usr/bin/scala 
export PATH=$SCALA_HOME/bin:$PATH

source ~/.bashrc

Spark 설치

wget https://archive.apache.org/dist/spark/spark-3.2.4/spark-3.2.4-bin-hadoop3.2.tgz

# 압축 해제
tar xvf spark-3.2.4-bin-hadoop3.2 opt/spark

# Scala Set
vi ~/.bashrc

export SPARK_HOME=/opt/spark
export PATH=$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin
export PYSPARK_PYTHON=/home/ubuntu/anaconda3/bin/python3

import sys

from pyspark.sql import SparkSession
if __name__ == "__main__":

    #스파크 세션 생성
    spark = (SparkSession
            .builder
            .appName("PythonMnMCount")
            .getOrCreate())

    # 파일 경로 설정
    mnm_file = "./mnm_dataset.csv"

    # csv 파일을 읽어서 데이터프레임 생성
    mnm_df = (spark.read.format("csv")
            .option("header", "true")
            .option("inferSchema", "true")
            .load(mnm_file))
    #읽어낸 데이터 중에서 5개를 출력
    mnm_df.show(n=5, truncate=False)

    # 데이터 프레임을 이용한 데이터 처리
    count_mnm_df = (mnm_df.select("State", "Color", "Count")
            .groupBy("State", "Color")
            .sum("Count")
            .orderBy("sum(Count)", ascending=False))

    # 60개의 데이터 출력
    count_mnm_df.show(n=60, truncate=False)

    # 캘리포니아의 데이터 만 추출
    ca_count_mnm_df = (mnm_df.select("*")
            .where(mnm_df.State == 'CA')
            .groupBy("State", "Color")
            .sum("Count")
            .orderBy("sum(Count)", ascending=False))

    # 10개의 데이터 출력
    ca_count_mnm_df.show(n=10, truncate=False)

Spark RDD

1.x 버전 부터 존재

Apache Spark 에서 사용되는 기본 데이터 구조 중 하나

RDD는 분산된 컴퓨팅 환경에서 대규모 데이터를 처리하는데 사용되는 불변성을 가진 분산 컬렉션

연산을 수행하면 원본에 작업하는 것이 아니고 원본의 복제본에 작업해서 리턴하는 구조

• RDD는 Resilient Distributed Dataset 의 줄임말로 스파크의 기본 데이터 구조

핵심 특성

1. Dependency(의존성): 현재의 RDD가 어떻게 만들어졌는지 스파크에게 가르쳐주는 것으로 이 의존성을 가지고 연산을 수행해서 새로운 RDD를 만들어야 하는 경우 의존성을 사용해서 생성
   • 데이터에 대한 Meta Data 로 Schema 와 유사한 역할
2. Partition: 스파크에게 작업을 나눠서 병렬 연산을 수행할 수 있는 능력을 부여하고 이 정보를 통해서 가까이 있는 데이터를 처리할 수 있는 실행 머신에 작업을 전송
3. Compute Function: partition의 데이터를 하나씩 순차적으로 접근할 수 있는 iterator 형태로 만들어주는 함수

구조 확인 - spark 2.x

spark 1.x 버전에서는 map ㅣ안 filter 같은 함수를 이용해서 데이터 작업을 수행해서 코드를 읽기가 어렵고 spark 엔진이 이를 해석하는 시간이 소요

SQL 테이블이나 스프레드시트처럼 보이는 정형화 데이터 타입을 써서 데이터 표 형태로 구성할 수 있게 됨.

자주 사용되는 일상적인 패턴들을 이용해서 연산을 표현

• 각 언어 별로 직접 사용자 정의 함수를 만들어서 처리하는 경우가 많아서 언어 별로 코드가 달랐는데
• 고수준의 함수를 이용하도록 만들어서 언어가 달라도 코드가 거의 비슷해지도록 만들어서 언어가 달라도 코드는 거의 비슷
  • pandas의 DataFrame보다 Spark의 DataFrame을 사용하는 이유.