'빅데이터 분석을 위한 스파크2 프로그래밍'책의 내용을 정리한 포스팅입니다.

RDD란?

스파크가 사용하는 핵심 데이터 모델로서 다수의 서버에 걸쳐 분산 방식으로 저장된 데이터 요소들의 집합을 의미하며, 병렬처리가 가능하고 장애가 발생할 경우에도 스스로 복구될 수 있는 내성을 가지고 있다. 즉, RDD란 스파크에서 정의한 분산 데이터 모델인데 내부에는 단위 데이터를 포함하고 있고 저장할 때는 여러 서버에 나누어 저장되며, 처리할 때는 각 서버에 저장된 데이터를 동시에 병렬로 처리할 수 있는 모델이다. 

RDD장점

데이터를 여러 서버에 나누어 저장하고, 처리하는 과정에서 일부 서버 혹은 데이터에 문제가 발생하더라도 스스로 에러를 복구할 수 있는 능력을 가지고 있는 데이터 모델이다.

RDD처리 방식

RDD에 속한 요소들은 파티션이라고 하는 더 작은 단위로 나눠질 수 있는데, 스파크는 작업을 수행할 때 바로 이 파티션 단위로 나눠서 병렬로 처리를 수행한다. 이렇게 만들어진 파티션은 작업이 진행되는 과정에서 재구성되거나 네트워크를 통해 다른 서버로 이동하는, 이른바 셔플링이 발생할 수 있다. 

이런 셔플링은 전체 작업 성능에 큰 영향을 주기 때문에 주의해서 다뤄야 하며, 스파크에서는 셔플링이 발생할 수 있는 주요 연산마다 파티션의 개수를 직접 지정할 수 있는 옵션을 제공한다.  (파티션의 수는 곧 데이터 처리에 참여하는 병렬 프로세스의 수이다. 즉, 하나의 데이터를 잘게 쪼개어 여러 개의 파티션을 만들면 여러 프로세스에서 동시에 작업을 처리해서 처리 속도가 증가할 수 있지만 이 정도가 지나치면 오히려 전체 성능을 떨어뜨리는 요인이 된다.

스파크의 장애시 RDD복구

하나의 RDD가 여러 파티션으로 나눠져 다수의 서버에서 처리되다 보니 작업 도중 일부 파티션에 장애가 발생해서 데이터가 유실될 수 있는데, 스파크는 손상된 RDD를 원래 상태로 다시 복원하기 위해 RDD의 생성 과정을 기록해 뒀다가 다시 복구해 주는 기능을 가지고 있다. RDD의  resilient라는 단어가 복구 능력을 의미하는데, 좀 더 정확하게 말하면 RDD에 포함된 데이터를 저장해 두는 것이 아니고 RDD를 생성하는 데 사용했던 작업 내용을 기억하고 있는 것이다. 그래서 문제가 발생하면 전체 작업을 처음부터 다시 실행하는 대신 문제가 발생한 RDD를 생성했던 작업만 다시 수행해서 복구를 수행한다.

정리하면, 스파크는 RDD가 생성되어 변경되는 모든 과정을 일일이 기억하는 대신 RDD를 한번 생성하면 변경되지 않는 읽기 전용 모델로 만든 후 RDD 생성과 관련된 내용만 기억하고 있다가 장애가 발생하면 이전에 RDD를 만들 때 수행했던 작업을 똑같이 실행해 데이터를 복구하는 방식을 사용한다. 이처럼 스파크에서 RDD 생성 작업을 기록해 두는 것을 리니지(linege)라고 한다.

도움이 되셨다면 광고도 한 번 클릭해주시는 센스^_^

하둡 데이터노드들에 색인시스템을 설치해 로그에서 관심사 뽑는 작업이 있었는데 전사 공통하둡클러스터로 넘어가게 되면 별도로 설치하기가 힘든 이유로 의존성을 제거해야하는 작업을 진행하게 되었다.

그 과정에서 키워드-관심사가 매핑되어 있는 데이터를 브로드캐스트 변수를 사용해 executor들에게 넘겨사용하는 과정에서 삽질했던 경험을 공유한다. 회사 업무로 진행해서 자세한 내용까지 포스팅하기는 힘들고 브로드캐스트에 대해 새로 알게된 내용들에 대해서만 기록해본다.

먼저 다음과 같이 브로드캐스트변수를 메인 클래스에 선언하고 해당 Broadcast.value를 executor들에게 넘겨서 사용하도록 하였다.

Driver에 Broadcast선언

문제는 값들이 정상적으로 넘어가 처리가 되었지만 실시간 배치(마이크로배치 : 1분)으로 처리되는 과정에 용납하기 힘든 delay가 발생하였다.

기존 스트리밍 처리 속도 (delay - no)

브로드캐스트 변수 적용 후

이정도 딜레이면 사용할 수 없다고 볼 수 있겠다....

원인을 찾으려고 삽질을 굉장히 많이 했었다.

원인은

스파크는 드라이버에서 각각의 executor들에게 작업을 분담시키고 executor들은 내부적으로 task를 만들어 수행하는데 broadcast를 executor에 넘기는 과정에서 모든 task마다 broadcast의 객체를 재생성하고 있었다. (실제 SuffixTree(매핑에 사용하는 객체)의 hashCode값을 찍어 보았다.

마이크로 배치잡으로 1분 간격으로 실행되는데 그때마다 15개의 executor들에서 250개정도의 task가 생기는데 각 task마다 객체를 다시 만들어 내고 있었던 것이다.....

문제해결

해결은 모든 task에서 객체를 생성하지 않고 각 executor들마다 한번만 생성하도록 싱글톤방식으로 수정하여 해결하였다.

주의 할 것은 브로드캐스트 변수의 값은 지역적이어야 하며 직렬화 가능해야 한다는 것이다!

다음과 같이 선언한 드라이버 내부에서 지역적으로 사용해야만 기대했던 방식대로 브로드캐스트를 사용할 수 있다.

이런식으로 action이 연산이 이루어지는 곳에 값을 넘겨 사용하게 되면 각 executor의 task마다 다시 생성한다는 사실....주의하자.

스스로의 기억을 위한 포스팅으로 내용이 자세하지 못한점 죄송합니당(꾸벅...)

Spark 스파크 지연 평과와 장애 내구성

스파크는 장애에 강하다라는 말을 쓰는데 이는 하드웨어나 네트워크 장애에도 작업이 완전히 실패하지 않고 데이터 유실이 일어나거나 잘못된 결과를 반환하지 않는다는 의미다.

스파크의 우수한 장애 내구성(fault-tolerance) 구현 방식은 각 파티션이 자신을 재계산하는 데 필요한 종속성 정보 등의 데이터를 갖고 있기 때문에 가능하다.

일반적인 분산 컴퓨터 패러다임은 데이터 변경을 일일이 로깅해 놓거나 노드들에 데이터를 복제해 놓는 방식으로 장애에 대비하는 반면에

스파크는 각 파티션이 복제에 필요한 모든 정보를 갖고 있으므로 각 RDD에서 데이터 변경 내역 로그를 유지하거나 실제 중간 단계들을 로깅할 필요가 없다.

만약 파티션이 유실되면 RDD는 재계산에 필요한 종속성 그래프에 대한 충분한 정보를 갖고 있으므로 더 빠른 복구를 위해 병렬 연산을 수행할 수도 있다.

메모리 영속화와 메모리 관리

맵리듀스와 비교해 스파크의 성능상 이점은 반복 연산이 들어 있는 사례에서 상당한 우위를 보인다. (다시 말해 모든 케이스에서 스파크가 월등히 빠른건 아니다, 실제 단순 작업의 경우에는 하둡 맵리듀스와 작업시간이 크게 안나는걸 여러번 경험했다.)

성능 향상의 많은 부분은 스파크가 메모리 영속화(in-memory persistence)를 활용하는 덕택이다. 스파크는 데이터가 거치는 각 단계마다 디스크에 기록하는 대신 이그제큐터의 메모리에 데이터를 로드해 놓을 수도 있다. 그러므로 파티션의 데이터에 접근이 필요할 대마다 메모리에서 꺼내 올 수 있다.  (스파크는 영속화를 위한 메모리 영역을 저장 장치처럼 따로 관리한다고 생각하면 된다.)

스파크는 메모리 관리에 대해 세 가지 옵션을 제공한다.

1. 메모리에 직렬화되지 않은 자바 객체

이 방식은 직렬화 하는 시간이 필요 없으므로 가장 빠르지만, 객체 그대로 저장하기 위해서 그를 표현하는 데이터도 같이 저장해야 하므로 메모리 공간 사용이 비효율적이다.

2. 메모리에 직렬화된 데이터

직렬화되지 않은 데이터를 읽는 것에 비해 직렬화된 데이터를 읽는 데에는 CPU가 더 많이 사용되므로 이 접근 방식은 더 느릴 것이다.

하지만 메모리 공간 사용 측면에서는 직렬화하지않고 저장하는 방식보다 뛰어나다. 자바의 기본 직렬화는 원본 객체보다는 효과적이지만 크리오(Kyro) 직렬화를 쓰면 공간 측면에서도 더욱 효과적이다. (무조건 Kyro쓰는 것을 권장한다.)

3. 디스크

각 executor의 램에 담기에 파티션이 너무 큰 RDD라면 디스크에 데이터를 쓸 수 있다. 이 전략은 당연히 반복 연산에는 속도 면에서 불리하다.

그러나 오래 걸리는 트랜스포메이션들이 반복되고, 가장 장애에 안전하고 또한 막대한 양의 연산을 해야 한다면 유일하게 선택할 수 있는 옵션이다.

해당 내용은 '하이 퍼포먼스 스파크(High Performance Spark)' 내용을 학습하다가 정리한 내용이다.

현상황  : Cloudera(클라우데라) 버전(CDH 5.5.1, Parcel), Spark버전(1.5) - jdk version 1.7

필요상황 : 기존 작업을 Spark1.5(jdk1.7) - jdk 1.8로 돌리기

준비상황 :  클러스터의 각 노드들에 jdk1.8이 설치되어 있어야 함.

spark-submit스크립트에 jdk1.8 path를 명시

--conf "spark.yarn.appMasterEnv.JAVA_HOME=/home1/irteam/jdk/jdk1.8.0_141" \
--conf "spark.driverEnv.JAVA_HOME=/home1/irteam/jdk/jdk1.8.0_141" \
--conf "spark.executorEnv.JAVA_HOME=/home1/irteam/jdk/jdk1.8.0_141" \

이렇게 driver와 executor의 JAVA_HOME은 명시가 되었고

해당 스크립트가 돌아가는 client의 JAVA_HOME은 export로 변경해준다.

기존 단순 spark-submit명령어에서 앞에 다음과 같이 추가

export JAVA_HOME=/home1/irteam/jdk/jdk1.8.0_141 && spark-submit \

이렇게 설정을 해주게 되면 jdk1.8로 빌드된 코드도 기존 spark로 돌릴 수 있게 된다.

이런식의 설정으로 클라우데라 스파크 버전 또한 2점대로 높여 사용할 수 있겠다.

카산드라 Connection Pooling에 따른 요청수와 요청 처리방법

How managed node contactPoints?

// (1)
//ContactPoints 하나라도 존재해야함!!!
Assert.isTrue(StringUtils.hasText(contactPoints), "At least one server is required");

public Builder addContactPoints(String... addresses) {
    for (String address : addresses)
        addContactPoint(address);  (2)
    return this;
}

(3)
public Builder addContactPoint(String address) {
    // We explicitly check for nulls because InetAdress.getByName() will happily
    // accept it and use localhost (while a null here almost likely mean a user error,
    // not "connect to localhost")
    if (address == null)
        throw new NullPointerException();
    try {
        addContactPoints(InetAddress.getAllByName(address)); (4)
        return this;
    } catch (UnknownHostException e) {
        throw new IllegalArgumentException("Failed to add contact point: " + address, e);
    }
}

(5) 
public Builder addContactPoints(InetAddress... addresses) {
    Collections.addAll(this.rawAddresses, addresses);
    return this;
}

(6)
contactpoints는 배열로 관리됨
private final List<InetAddress> rawAddresses = new ArrayList<InetAddress>();

How selected node when API server requests

RounRobinPolicy : 라운드로빈방식에 대해서는 따로 설명하지 않겠다. (단순 동일하게 각 노드에게 돌아가면서 요청하는 방식)

Cassandra 버전 별 허용 Request수

Cassandra versions 1.2 and 2.0 allow the clients to send up to 128 requests without waiting for a response per connection.
Higher versions of Cassandra ( that is 2.1 or greater) allow clients to send up to 32768 requests without waiting for a response.
단순히 이 계산식으로 계산을 했을 때
한 노드 당 14억(1415577600)정도의 요청을 처리 할 수 있음.
http://datastax.github.io/nodejs-driver/features/connection-pooling/

Cassandra 커넥션 맺는 flow 및 ProtocolVersion에 따른 차이

참고 : https://docs.datastax.com/en/developer/java-driver/3.6/manual/pooling/

참고 : https://docs.datastax.com/en/developer/java-driver/3.6/manual/native_protocol/

ContactPoints는 클러스터 모든 서버 목록을 적어줘야하나???

결론을 먼저 말하자면 '그렇지 않다'이다!!!

실질적으로 어플리케이션이 로드될 때 카산드라 드라이버는 컨택포인트에 적어진 노드 하나에 접근하게 되고 

그 노드로부터 실제 클러스터의 노드 정보등을 받아와 메모리상에 메타정보로 관리하게 된다.

따라서 클러스터가 10대로 구성되어 있더라도 ContactPoints에는 2~3대만 기입해줘도 문제가 없다.

다음의 내용을 참고하자.

자세한 내용이 궁금하다면 다음을 참고하자.

참고 : https://teddyma.gitbooks.io/learncassandra/content/client/which_node_to_connect.html  (Which Node to Connect)

Spark dataframe(스파크 데이터프레임)으로 작업 중 dataframe의 null값을 특정값으로 바꾸고 싶은 경우가 있다.

이 때 주의해야할 점은 dataframe의 컬럼의 자료형 타입에 맞게끔 변환해줘야 정상적으로 replace된다.

다음과 같은 데이터프레임(dataframe)이 있을 때 "bid_i"의 값을 0으로 변경하려고 다음을 실행

val result_df_q_new = result_df_q.na.fill(0, Seq("bid_i"))

위와 같은 명령을 수행하고 확인을 해도 정상적으로 null값이 0으로 변경되지 않은 걸 확인할 수 있었다.

원인은 bid_i의 자료형 타입에 맞지않게 변경하려했기 때문이다.

위에서 보듯이 "bid_i"의 자료형 타입은 string인데 0으로 변경(na.fill메서드를 하려고 하니 정상적으로 변환되지 않았던 것이다. 

(처리 도중 딱히 에러메세지가 없었다...)

val result_df_q_new = result_df_q.na.fill("0", Seq("bid_i"))

0을 string형(큰따옴표)를 씌워서 명령어를 주니 정상적으로 변경되는 것을 확인할 수 있었다.

데이터프레임(dataframe) 값을 na.fill을 통해 변경할 때는 자료형타입을 잘 확인하도록 하자!

Dataframe count중 scala.MatchError 발생 ( show 명령어는 정상적으로 먹는데??? )

간만에 스파크 작업을 진행하다가 다음과 같은 에러문구를 만났다.

Text 파일을 sc.textFile("path") 로 읽어와 해당 RDD를 Dataframe으로 변환하고 dataframe.count시 발생하였다.

문제의 원인은  실제 텍스트파일내의 내용을 RDD로 읽어와 split으로 나눈 후 Dataframe으로 변형하는 과정에서 

컬럼개수가 일치하지 않는 데이터가 있었다.

text파일의 포맷은 다음과 같았다.

이러한 RDD를 다음과 같이 Dataframe으로 만들려고 하였다.

전혀 문제가 없을 것 같지만 이렇게 변경 후 count명령을 내려보면 MatchError가 발생하는 것이다.

알고보니 itemName에 상품명중 탭 구분자("\t")가 들어가 있었던 상품명이 있었던 것이다.

그래서 실제로는 4개의 컬럼으로 나누어져서 Dataframe이 만들어져야 하지만 상품명에 탭구분자가 있는 경우

Array length가 4이상이되면서 macth case 조건에 정상적으로 매칭되지 않았던 것이다.

이럴 경우 Dataframe을 show로 보았을 때는 전혀 문제가 없어보일 수 있지만 count로 모든 데이터의 수를 세게 될 떄

특정 데이터에 문제가 있을 경우 matchError가 발생하게 되는 것이다.

해결방법은 map처리에서 flatMap처리로 변경 후 조건에 맞지 않는 데이터는 None처리를 진행하였다.

또 다시 같은 삽질을 하지 않기 위해 포스팅 남겨본다.

스파크 데이터프레임(Dataframe) partitionBy를 사용해 원하는대로 손쉽게 저장하자!

스파크(Spark) 데이터프레임(Dataframe) 혹은 데이터셋(Dataset)을 통해 작업하게 되면 

sql기반의 명령을 통해서 데이터를 손쉽게 활용할 수 있다는 점과 더불어 특정 컬럼 기반으로 

데이터를 저장할 수가 있다.

잠깐 데이터프레임(Dataframe)과 데이터셋(Dataset)에 대해 언급하자면 데이터셋(Dataset)은 데이터프레임과 RDD의 단점들을

보완한 모델로 Spark 1.6이상 버전부터 사용할 수 있다.

이번에 데이터프레임(Dataframe)을 partitionBy를 통해 저장해보았는데 이런 기능이 있다라고만 알았지

막상 써보니 너무 편해서 정리하게 되었다.

다음과 같은 데이터프레임(Dataframe)이 있을 때

partitionBy를 이용하면 특정 컬럼을 기반으로 디렉토리를 나누어 저장할 수 있다.

예를들어 advid를 기준으로 데이터를 나누어 저장하고 싶을 떄

라고 저장해주면 다음과 같이 파일들이 advid를 기준으로 저장되게 된다.

이 얼마나 간편한가!!!!

다들 partitionBy를 통해 데이터를 원하는대로 손쉽게 저장해서 사용하시길!!!

카산드라(Cassandra) ttl은 read의 성능에 영향을 미치는가?

테스트 환경 및 프로세스

1. 알파 카산드라(node3개)에 ttl을 설정한 테이블과 ttl설정을 하지 않은 테이블을 각각 만든다.
2. 각각의 테이블에 2만개씩의 데이터를 넣는다 (id, value1, value2)  - id(1~20000)
3. api로 해당 테이블을 select하는 요청(id는 1~20000 중 랜덤으로 선택)을 한다.(20초동안 2000번)

테스트 결과 1

99th percentile에 주목해주세요(요청의 99% 평균 처리시간을 의미합니다.

	1번	2번	3번
NO TTL 테이블 select 요청
TTL 걸린 테이블 select 요청

99th percentile을 보시면 알겠지만 ttl이 걸린 테이블이 오히려 처리속도가 더 빠른경우도 있습니다....(살짝 멘붕)

아무리 그래도 ttl걸린 테이블이 당연히 latency가 높을줄 알았는데....ㅎㅎ

그래서 각각의 테이블에 10000개의 데이터를 더 놓고 테스트를 진행하였습니다.

여기서 다른점은 ttl이 걸린 테이블에 임의로 10000개의 데이터의 ttl을 1분을 주고 tombstone이 발생하도록 하였습니다.

그리고 나서 다시 api로 select 테스트를 하였습니다.

테스트 결과2 (각각 테이블에 1만개씩 더 밀어넣음 - ttl걸린테이블에는 tombstone이 발생하도록 ttl을 1분으로 설정하고 넣음)

	1	2	3
NO TTL 테이블 select 요청
TTL 걸린 테이블 select 요청

결과를 보시면 1만개씩을 더 밀어넣었지만 기존 TTL이 걸리지 않은 테이블은 첫 번째 테스트 속도와 거의 비슷한 결과를 도출한 반면

1만개의 데이터를 tombstone이 발생하도록 밀어넣은 TTL이 설정된 테이블은 속도가 현저히 떨어진 것을 확인해볼 수 있습니다.

그렇다면 왜 이런 결과가 나온것인가?

카산드라는 read요청이 들어올 경우 첫 번째로 memtable을 뒤지고
그 다음으로 row cache데이터가 있다면 row cache데이터를 확인합니다. (카산드라는 빈번하게 요청되는 row를 위한 캐시기능을 제공합니다)
row cache에도 데이터가 없는 경우 Bloom filter(메모리에 상주) 확인하여 찾는 row의  키가 있는 SSTables를 확인하여
해당 SSTables를 통해서 데이터를 읽어오게 됩니다.

따라서 데이터가 insert되면서 Bloom filter SSTables내의 데이터의 키값들이 해싱되어 업데이트 되는 경우에는
실제로 SSTables(dis I/O)를 사용하지 않고 데이터를 가지고 오게되어 빠르게 처리가 가능합니다.

하지만 Bloom filter의 해싱된 데이터들이 업데이트 되기전 SSTables의 값이 delete나 drop등으로 변경되는 경우에는
Bloom filter와 SSTable사이간 데이터의 정밀도가 떨어지게 됩니다.

이 경우 Bloom filter는 이미 SSTable내의 Tombstone된 데이터를 가르키게 되고 잘못된 포인팅으로 인해
카산드라는 SSTables에서 직접 해당 데이터를 다시 가져오는 작업(Disk I/O 발생)을 진행합니다.
이러한 이유로 ttl이 걸린 테이블 혹은 row에서 select시 성능이 떨어질 수 있다고 생각됩니다.
더 정확한 내용은 다음 문서를 참고해보시면 좋을 것 같습니다.
[Cassandra how is data read?](How is data read?)

TTL을 설정하고도 성능을 떨어트리지 않는 방법이 있는가?

SSTables과 Bloom Filter 정밀도를 높여서 운영하는 방법입니다.
하지만 모든선택에는 tradeoff가 있듯이 정밀도를 높이게 되면 더 많은 메모리를 필요로 하게 됩니다.
설정 방법은 테이블을 만들 때 bloom_filter_fp_chance 옵션을 주는 방법입니다.
추천값은 0.1이지만 최대 1.0까지 설정이 가능합니다.
1.0으로 설정할 경우 SSTables과의 정밀도를 높일 수 있으나 더 많은 메모리를 필요로 하기 때문에
상황에 맞는 설정값으로 운영하시면 될 것 같습니다.