To Data Engineer

Spark Stream이란 ?

Real-time processing / Stream processing

- IOT 세상이 다가오면서 데이터의 크기도 중요하지만 속도도 부각

- 대량의 데이터를 빠르게 처리하기 위한 기술들이 실시간 처리 / 스트림 터리

- 실시간 처리는 데이터 처리의 목표 또는 제약 사항 방식 (마감 시각이 있다.)

- 스트림 처리는 데이터 처리 방식 (끊임없이 흘러가는 데이터에 대한 처리 방식)

Basic Concept

- 초 단위로 데이터 처리

- 마이크로 배치의 연속으로 구현 (DStream)

- kafkam Twitter, File System, TCP Socket 등등에서 데이터 수집

- RDD에 제공되는 Operation과 시간 기반의 Operation (예를 들면 sliding window)를 제공

Discretized System

- DStream이란 시간 흐름에 따른 순차적 데이터를 의미

- RDD 개념을 갖고 있는 Spark와 유사하게 Spark Streaming은 DStream 또는 Discretized Streams라고 불리는 추상 개념을 가지고 있다.

실습 예)

SPARK SQL이란 ?

: 구조화/반구조화 된 데이터를 처리하는 것     

sql에 대한 기본 지원을 제공, 외부 소스 모두에 저장된 데이터 조회가 간소화 됨, RDDs와 관계형 테이블의 경계가 사라짐.

- Import relational data from Parquet files and Hive tables

- Run SQL queries over imported data and existing RDDs

- Easily write RDDs out to Hive tables or Parquet files

다양한 interface 지원

> SQL, HiveSQL queries & DataFrame API (java, python, scala R) & DataSet API(scala, java)

다양한 Input source를 지원

>RDD & 임시테이블

 JSON 데이터 셋

 Parquet file

 HIve Table

 ODBC/JDBC 서버와의 연동

DataFrame이란 ?

: columns 들로 조직된 분산된 데이터 모음 ( 관계형 데이터베이스와 같은 개념) - 데이터 프레임이란 관계형 데이터 베이스의 테이블과 R, python의 데이터 프레임과 같은 개념이다. (데이터 프레임은 2차원 배열 구조로 구성되어 있다.)

DF can be constructed from structured data files, tables in Hive, external databases, or existing RDDs

python에서 DF를 만드는 방법    
 

• Ability to scale from kilobytes of data on a single laptop to petabytes on a large cluster

• Support for a wide array of data formats and storage systems

• State-of-the-art optimization and code generation through the Spark SQL Catalystoptimizer

• APIs for Python, Java, Scala, and R (in development via SparkR)

외부 데이터 소스를 읽어오면 데이터 프레임 형태로 읽음 (예제)

Running SQL Queries Programmatically

RDD란?

RDD Operator

- Tranceformation : 데이터의 흐름, 계보를 만듬

- Action Transformation에서 작성된 계보를 따라 데이터 처리하여 결과를 생성함 (lazy-execution)

Actions == transformations의 결과를 가져오는 과정

선 디자인 후 연산이 시작된다 >>>>> 미리 계획이 되어 있었기 때문에 연산 속도가 매우 빠르다.

Narrow dependency / Wide dependency

Narrow dependency 형태로 코딩해야 한다.

- 책상 한자리에서 다 처리할 수 있는 일은 모아서 하는 것이 좋다는 개념

- 네트워크를 안타고 메모리의 속도로 동작해서 빠르다.

- 파티션이 부서져도 해당 노드에서 바로 복원 가능하다.

- map, filter, union, join with inputs co-partitioned

Wide dependency

- 여러 책상에 있는 자료를 훑어와야 한다는 개념

- 네트워크의 속도로 동작해서 느리다.

- 노드끼리 셔플이 일어나야 한다.

- 파티션이 부서지면 계산 비용이 비싸다.

- groupByKey, join with inputs not co-partitioned

>>>>dependency의 의미

메모리 안에서 처리 될 일들을 먼저 처리하고 네트워킹을 타거나 IO를 타는 일을 나중에 하도록 코딩을 해아한다는 의미.

네트워킹, IO시간을 메모리와 번갈아 가면서 구동하면 속도가 많이 느려지므로 최대한 메모리 내에서 할 수 있는 일을 한 후에 wide denpendency를 활용해야 한다는 의미.

RDD Persistence (더 학습이 필요..)

- RDD가 Action으로 수행될 때마다 다시금 소스에서 부터 다시 로드되서 수행됨.

- RDD를 저장해놓고 사용하는 기능으로 persist()와 cache()라는 두 가지 오퍼레이셔을 지원

- 저장 옵션을 MEMORY_ONLY로 한 옵션과 동일

- 디폴트는 메모리에 저장하고, 옵션으로 디스크를 지정

- 메모리에 저장할 때, RDD를 RAW(원본 형식)으로 저장할 것인지 자바의 Serialized(직렬화)된 형태로 저장할 지 선택 

>>>오류 발생시 lineage의 맨 처음부터 수행하는 것이 아니라 중간 지점을 기억해 놓고 그 기억시점부터 다시 수행시키는 기술, 메모리를 사용하지만 속도가 더 빠른 것은 당연한 일

RDD란?

spark에서 가장 핵심인 RDD에 대해서 알아보자.

사실 slideshare의 하용호 데이터사이언티스트님의 자료로 부터 많은 걸 얻을 수 있었다. 이 분의 자료는 실로 대단하고 또 쉽다. 나도 넘버웍스 인턴에 지원해보고 싶지만... 아직 부족한듯 하다.

무튼 참고자료  
            http://www.slideshare.net/yongho/rdd-paper-review?qid=3ff4fd97-e003-46c3-aeb9-dcc3977cdf0d&v=&b=&from_search=1

RDD란?

- 분산되어 있는 변경 불가능한 객체 모음(분산되어 존재하는 데이터 요소들의 모임)

- 스파크의 모든 작업은 새로운 RDD를 만들거나 존재하는 RDD를 변형하거나 결과 계산을 위해 RDD에서 연산(함수나 메소드)을 호출하는 것 중의 하나로 표현

Hadoop MapReduce의 단점?

> Machine Learning에 적합하지 않다.

> 데이터 처리시 HDFS(Hadoop Distributed File System)을 거치기 때문에 IO에서 시간이 오래 걸린다.

Spark는?

> RAM에서 Read-Only로 처리해서 running time이 빠르다.                        

Data Sharing is slow in MapReduce

- 맵리듀스들 사이에엇 데이터를 재사용하는 방법은 외부 안전한 스토리지 시스템에 데이터르 쓰는 방법뿐이다. 

- 반복적이거나 Interactive 애플리케이션은 빠른 데이터 공유가 필요하다. 하지만 디스크 IO 응답시간, 직렬화시간 때문에 맵리듀스는 속도가 느리다. > 하둡 시스템에서 90% 이상의 시간이 HDFS로부터 데이터를 읽고 쓰는데 걸린다.
       (하둡의 작동 방법 : 데이터를 찾아서(namednode에 질의) HDFS서 읽은 후 연산하고 그 결과를 로컬에 저장한 후 합쳐서 HDFS에서 다시 업로드 하는 방식)

Data Sharing using Spark RDD(Resilient Distributed Datasets)

- it supports in-memory processing computation. 작업 전체의 대상이 되는 메모리의 상태로 저장된다(Sharable)
    > 네트워크나 디스크를 사용하는 것보다 10~100배 빠르다. >> 오류시 fault-tolerant 방식이 필요하다.(RAM에만 올려서 사용)

- immutable(read-only), partitioned collections of records

- 스토리지 >RDD변환 or RDD > RDD만 가능

##fault-tolerant 방식이란?
        : 컴퓨터 시스템이란 시스템 내의 어느 한 부품 또는 어느 한 모듈에 Fault(장애)가 발생하더라도 시스템 운영에 전혀 지장을 주지 않도록 설계된 컴퓨터 시스템

- DAG (directed acyclic graph 디자인) > 코딩을 하는 것은 실제 계산 작업이 되는 것이 아니라 Lineage 계보를 디자인 해 가는 것 (빈 RDD를 만든다.)

일의 순서대로 빈 RDD를 만들어 가다가 가장 마지막 일이 수행되었을 때, 실제 일을 시작함
    > Lazy execution의 개념이 여기서 나옴 - 일을 미루다가 나중에 한 방에 처리

Spark 클러스터 설정/ 구축

1. 워커 노드에 SSH서버를 설치해서 마스터 노드에서 들어갈 수 있게 한다.

>sudo apt-get install oppenssh-server

2. 마스터 노드에서  키를 생성

> ssh-keygen

3. 워커 노드에 매번 비밀번호를 치고 들어가지 않게 하기 위해 RSA 키 복사

> ssh-copy-id -i ~/.ssh/id_rsa.pub user@x.x.x.x

4. conf 폴더에 slaves 템플릿을 복사해 파일 생성 워커 노드들을 잡아준다.

5. spark-env.sh도 템플릿을 복사해 파일을 생성한다.
   export SPARK_MASTER_IP = N.N.N.N  >>>>> 마스터 노드의 아이피
   export SPARK_WORKER_CORES = 2    >>>>> 워커 노드의 사용 가능한 총 CPU 코어 수
   export SPARK_WORKER_MEMORY = 2400m >>>>> 워커 노드의 전체 메로리 할당량
   export SPARK_WORKER_INSTANCE =1 >>>>> 워커 노드의 프로세서 갯수

6. 마스터 노드의 커뮤터도 워커 노드에 추가해 활용하기
   ./bin/spark-calss org.apache.spark.deploy.worker.Worker spark://10.0.0.25:7077

7. 스파크 실행
   MASTER=spark://10.0.0.25:7077 ./bin/pyspark

추후 실습을 다시 해서 추가내용으로 작성하도록 하자

- 설치 환경 ubuntu 10.15

1) 자바 설치

- 스파크 자체는 스칼라로 만들어졌으며, 자바 가상 머신 위에서 돌아간다.

- 스파크 설치를 위해서는 자바 6이상의 버전이 필요하다.

2) Spark 설치

- 먼저 spark 홈페이지에서 standalone 버전을 다운 받는다.

- 압출을 풀어 설치를 완료한다.

Spark 클러스터 구조

1. Driver Program : 스파크 프로그램. 여러 개의 병렬적 작업으로 나눠져 Worker Node에 있는 Executor에서 실행

2. SparkCotext : 메인 시작 지점. 스파크API를 활용하기 위해 필요하다. 클러스터의 연결을 보여주고 RDD를 만드는데 사용

3. Cluster Manager : Standalone, YARN. Mesos 등 클러스터 자원 관리자

4. Worker Node : 하드웨어 서버. 하나의 물리적 장치에 여러 개도 가능

5. Executer : 프로세스. 하나의 워커 노드에 여러 개 가능

Spark 스택 구조

1. 인프라 계층 (Standalone Scheduler, YARN, Mesos) : 먼저 스파크가 기동하기 위한 인프라는 스파크가 독립적으로 기동할 수 있는 Standalone, 하둡 종합 플랫폼인 YARN 또는 Docker 가상화 플랫폼인 Mesos 위에서 기동한다.

2. 스파크 코어 (Spark Core) : 메모리 기반의 분산 클러스터 컴퓨팅 환경인 스팍 코어가 그 위에 올라간다.

3. 스파크 라이브러리 : 스파크 코어를 이용하여 특정한 기능에 목적이 맞추어진 각각의 라이브러리가 돌아간다. 빅데이터를 SQL로 핸들링할 수 있게 해주는 Spark SQL, 실시간으로 들어오는 데이터에 대한 real-time streaming 처리를 해주는 Spark Streaming, 그리고 머신러닝을 위한 MLib, 그래프 데이터 프로세싱이 가능한 GraphX등이 있다.