행복한디벨로퍼

오늘은 스파크의 넓은 종속성(narrow dependency)와 좁은 종속성(wide dependency)에 대해서 포스팅 하도록 하겠습니다.

해당 내용은 '하이 퍼포먼스 스파크'와 '빅데이터 분석을 위한 스파크2 프로그래밍'의 내용을 정리한 것입니다.

[ 좁은 종속성(narrow dependency) ]

개념적으로 좁은 종속성은 자식 RDD의 각 파티션이 부모 RDD의 파티션들에 대해 단순하고 한정적인 종속성을 가지는 것이다. 디자인 시점에 종속성을 결정할 수 있고, 부모 파티션의 값과 상관이 없으며, 각각의 부모가 최대 하나의 자식 파티션을 가진다면 이 종속성은 좁다고 할 수 있다. 특히 좁은 트랜스포메이션의 파티션들은 하나의 부모 파티션에만 종속되거나(map 연산) 디자인 시점에 알게 된 부모 파티션들 중 알려진 일부에만 종속된다(coalesce). 그러므로 좁은 트랜스포메이션은 다른 파티션의 정보를 필요로 하지 않고 데이터의 임의의 부분에 대해 실행이 가능하다.

[ 넓은 종속성(wide dependency) ]

넓은 종속성을 가지는 트랜스포메이션은 임의의 데이터만으로 실행할 수는 없으며, 특별한 방법, 예를 들면 키의 값에 따라 파티셔닝된 데이터를 요구한다. (결국 키의 재분포, 즉 셔플이 필요하다는 의미). 일례로 sort같은 경우 같은 범위의 키들이 같은 파티션 안에 존재하도록 레코드들을 파티셔닝해야 한다. 넓은 종속성의 트랜스포메이션은 sort, reduceByKey, groupByKey, join 그리고 rePartition 함수를 호출하는 모든 것을 아우른다.

특별한 경우로는 스파크가 이미 데이터가 어떤 특정한 방법으로 파티션되어 있다는 것을 갈고 있다면 넓은 종속성의 연산이라도 셔플링을 하지 않는다. 대개 셔플링은 비싼 비용을 치러야 하고 셔플 과정에서 새로운 파티션으로 옮겨야 하는 데이터의 비율이 높을수록 이 비용은 더 비싸지게 마련이다. 

정리하자면, 하나의 RDD가 새로운 RDD로 변환될 때 기존 RDD를 부모 RDD, 새로운 RDD를 자식 RDD라고 하겠습니다. 이때 부모 RDD를 구성하는 파티션이 여러 개의 자식 RDD 파티션과 관계를 맺고 있으면 넓은 의존성을 갖고 있다고 말하고, 그 반대의 경우는 좁은 의존성을 갖고 있다고 표현할 수 있겠습니다.

포스팅을 마치도록 하겠습니다.

도움이 되셨다면 광고 한 번 클릭해주시는 센스 감사합니다:)

이번 포스팅은 저번 포스팅(스파크 설정 Part.1)에 이어 spark-submit 실행시 메모리, 익스큐터, 네트워크, 보안,암호화 관련 설정에 대해 정리해보겠습니다. 해당 내용은 '빅데이터 분석을 위한 스파크2 프로그래밍' 책의 내용을 기반으로 정리하였습니다.

[ 메모리 관련 설정 ]

• spark.memory.fraction : 전체 힙 영역에서 익스큐터와 RDD 데이터 저장에 사용될 크기를 비율로 설정합니다. 기본값은 0.6이며 스파크 내부에서 사용하는 메타데이터나 객체 직렬화 및 역질렬화 등에 필요한 예비 메모리 공간을 확보해서 OOM을 방지할 목적으로 이 값을 조정할 수 있습니다.
• spark.memory.storageFraction : 할당된 메모리에서 데이터 저장에 사용할 비율을 지정할 수 있습니다. 기본값은 0.5이며 이 값을 크게 할 경우 익스큐터에서 사용할 메모리 크기를 줄여야 합니다.
• spark.memory.offHeap.enabled : 기본값은 false이며 true로 설정할 경우 off-heap메모리를 사용합니다. 이 값을 true로 설정했다면 spark.memory.offHeap.size에 오프-힙 메모리 크기를 지정해야 합니다.

최근에 특정 모듈을 분석해서 다시 만들어야 하는 작업을 진행하였다.

해당 모듈이 약간 성능에 민감한 모듈이다 보니 가능한 한 빠르게 동작하도록 하였어야 했다.

그러다가 ArrayList.contains() 메서드로 된 부분이 있어 HashSet.continas으로 수정해 성능이 조금 높아 진 것을 확인했는데

기본적인 개념으로만 HashSet이 내부적으로 HashMap을 구현하고 있어서 빠를 것 같다는 생각을 했었는데

좀 더 자세히 알고 싶어 포스팅 하게 되었다.

[ HashSet.Contains() ]

내부적으로 HashSet은 HashMap Instance를 구현하고 있고 contains() 메서드를 호출하게되면 HashMap.containsKey(object)가 호출된다고 보면 된다. 자세한 설명은 밑을 참고하자.

[ ArrayList.contains() ] 

ArrayList는 해당 값이 list에 있는지 판단하기 위해 내부적으로 indexOf(object) 메서드를 사용한다. indexOf(object) 메서드는 array 전체를 반복해서 돌고 각각의 element와 비교를 진행한다. 자세한 설명은 밑을 참고하자.

설명을 봤듯이 HashSet은 내부적으로 HashMap을 사용하고 ArrayList는 contains 메서드 내부에서 모든 element들을 돌며 비교작업을 진행하기 때문에 당연히 HashSet의 contains메서드가 빠를 수 밖에 없었던 것이다.

[ 테스트 코드 ]

1. 100만개의 String을 생성하여 HashSet, ArrayList를 만든다.

2. 500번 반복문을 돌며 Set과 List의 contains메서드를 사용해 확인한다.

HashSet Test

@Test
public void testHashSetContainsPerformance() {
    Set<String> testSet = new HashSet<>();
    String baseStr = "hellotheworld";
    Random generator = new Random();

    long start = System.currentTimeMillis();
    for (int i = 0; i < 1000000; i++) {
        testSet.add(baseStr + i);
    }
    long end = System.currentTimeMillis();
    System.out.println("Setup Performance : " + (end - start));

    long start2 = System.currentTimeMillis();
    int matchingCount = 0;
    for (int j =0; j < 500; j++) {
        if (testSet.contains(baseStr + generator.nextInt(5000))) {
            matchingCount++;
        }
    }
    long end2 = System.currentTimeMillis();
    System.out.println("HashSet Contains Performance : " + (end2 - start2));
}

ArrayList Test

@Test
public void testArrayListContainsPerformance() {
    List<String> testList = new ArrayList<>();
    String baseStr = "hellotheworld";
    Random generator = new Random();

    long start = System.currentTimeMillis();
    for (int i = 0; i < 1000000; i++) {
        testList.add(baseStr + i);
    }
    long end = System.currentTimeMillis();
    System.out.println("Setup Performance : " + (end - start));

    long start2 = System.currentTimeMillis();
    int matchingCount = 0;
    for (int j =0; j < 500; j++) {
        if (testList.contains(baseStr + generator.nextInt(5000))) {
            matchingCount++;
        }
    }
    long end2 = System.currentTimeMillis();
    System.out.println("ArrayList Contains Performance : " + (end2 - start2));
}

[ 결과 (시간단위 : ns) ]

결과를 보게되면 실제 set, list에 데이터를 쌓을 때는 hashset이 약 2배 정도 오래 걸리는 것을 알 수 있다. 

(아마 내부적으로 해쉬함수를 돌려 넣기 때문일 것으로 추정)

하지만 contains메서드의 hashset은 거의 시간이 안걸린다고 볼 수 있는 반면에 list는 set에 비해서는 꽤나 걸리는 것으로 볼 수 있다. 

따라서 한 번 데이터를 초기화해놓고 get이나 contains메서드를 사용할 일이 많은 시스템이라면 list보다는 set으로 자료구조를 가져가는 편이 좋다고 할 수 있을 것 같다.

포스팅을 마치도록 하겠습니다. 감사합니다.

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

참고 : https://www.baeldung.com/java-hashset-arraylist-contains-performance

이번 포스팅에서는 spark-submit 실행시 스크립트상에서 설정할 수 있는 방법에 대해 정리하도록 하겠습니다.

해당 내용은 '빅데이터 분석을 위한 스파크2 프로그래밍' 책의 내용을 기반으로 정리하였습니다.

[ 애플리케이션 관련 설정 ]

• spark.app.name : 애플리케이션 이름. SparkConf의 appName으로 설정하는 것과 같은 속성
• spark.driver.cores : 드라이버가 사용할 코어 수. 클러스터 모드에서만 사용 가능하며 기본값은 1입니다.
• spark.driver.maxResultSize : collect() 메서드 등의 호출 결과로 생성된 결과 값의 최대 크기입니다. 최소 1M 이상으로 지정해야 하며, 이 값을 초과할 경우 전체 잡은 실패로 종료됩니다. 기본값은 1g입니다.
• spark.driver.memory : 드라이버가 사용할 메모리 크기이며, 클라이언트 모드에서 사용할 경우 반드시 SparkConf가 아닌 --driver-memory 실행 옵션이나 프로퍼티 파일을 사용해서 지정해야 합니다. 기본값은 1g입니다.
• spark.executor.memory : 익스큐터 하나의 메모리 크기를 지정합니다. 기본값은 1g입니다.
• spark.local.dir : RDD 데이터를 디스크에 저장하거나 셔플 시 매퍼의 결과를 저장하는 디렉터리를 지정합니다. 콤마(,)를 이용해 여러 위치를 지정할 수 있으며, 성능에 큰 영향을 주므로 반드시 빠른 로컬 디스크를 사용해야 합니다. 기본값은 /tmp 입니다.
• spark.master : 클러스터 매니저 정보를 지정합니다.
• spark.submit.deployMode : 디플로이 모드를 지정합니다. client 또는 cluster 모드를 사용할 수 있습니다.

• spark.driver.extraClassPath : 드라이버 클래스패스에 추가할 항목을 지정합니다. 이 속성은 SparkConf가 아닌 --driver-memory 실행 옵션이나 프로퍼티 파일을 사용해서 지정해야 합니다. 유사한 속성으로 spark.driver.extraJavaOptions, spark.driver.extraLibraryPath가 있으며 각각 드라이버 실행 시 필요한 자바 옵션과 라이브러리 정보를 지정하는 용도로 사용됩니다.
• spark.executor.extraClassPath : 익스큐터의 클래스패스에 추가할 항목을 지정합니다. 유사한 속성으로 spark.executor.extraJavaOptions와 spark.executor.extraLibraryPath가 있습니다.
• spark.files, spark.jars : 각 익스큐터의 실행 디렉터리에 위치할 파일들 또는 jar 파일들을 지정하며, 콤마(,)를 이용해 여러 파일을 지정할 수 있습니다.
• spark.submit.pyFiles : PYTHONPATH에 추가될 .zip, .egg, .py 파일을 지정하며, 콤마(,)를 이용해 여러 파일을 지정할 수 있습니다.
• spark.jars.packages : 익스큐터와 드라이버의 클래스패스에 추가될 의존성 jar정보를 메이븐 코디네이트 형식으로 지정 할 수 있습니다.

• spark.reducer.maxSizeInFlight : 셔플 수행 시 각 리듀서가 매퍼의 실행 결과를 읽어갈 때 사용할 버퍼의 크기를 지정합니다. 기본값은 48m입니다.
• spark.reducer.maxReqslnFlight : 리듀서에서 매퍼의 결과를 가져갈 때 동시에 수행 가능한 최대 요청 수를 지정합니다. 기본값은 int.MaxValue입니다.
• spark.shuffle.compress : 맵의 결과를 압축할 것인지에 대한 설정입니다. true로 설정할 경우 spark.io.compress.codec에 지정한 압축 코덱을 사용해 압축합니다.
• spark.shuffle.service.enabled : 외부 셔플 서비스를 사용할 것인지 여부를 지정합니다. 이와 관련된 내용은 이후의 동적 자원 할당 부분에서 다시 확인해 보겠습니다. 기본값은 false이며 true로 설정할 경우 외부 셔플 서비스를 사용하게 됩니다.

• spark.eventLog.enabled : 스파크 이벤트 관련 로깅을 수행할 것인지를 설정합니다. 기본 값은 false이며 true로 설정할 경우 spark.eventLog.dir에 로깅을 수행할 경로를 지정해야 합니다. 이벤트 로깅을 활성화할 경우 종료된 애플리케이션에 대한 상세 실행 히스토리 정보를 스파크 UI에서 확인할 수 있습니다. 
• spark.ui.port : 스파크 UI 포트를 지정합니다. 기본값은 4040입니다.
• spark.ui.killEnabled : 스파크 UI를 통해 잡을 중지(kill)시킬 수 있도록 할 것인지 설정합니다. 기본값은 true입니다.
• spark.ui.retainedJob : 종료된 잡에 대한 정보를 몇 개까지 유지할 것인지 설정합니다. 유사한 옵션으로 spark.ui.retainedStages, spark.ui.retainedTasks, spark.ui.retainedExecutors, spark.ui.retainedDrivers, spark.ui.retainedBatches 등이 있습니다.

• spark.broadcast.compress : 브로드캐스트 변수의 값을 압축할 것인지 설정합니다. 기본값은 true입니다.
• spark.io.compression.codec : 브로드캐스트 변수나 셔플을 위한 중간 결과물 등 스파크 내부에서 사용하는 데이터를 압축할 때 사용할 압축 코덱을 지정합니다. l4z, lzf, snappy를 사용할 수 있으며 기본값은 lz4입니다.
• spark.kyro.classesToRegister : Kyro 직렬화를 위해 등록할 커스텀 클래스 정보를 지정합니다. 만약 클래스 등록 방식을 좀 더 커스텀하게 진행하고자 한다면 spark.kyro.registrator를 사용할 수 있습니다.
• spark.serializer : 스파크에서 사용할 객체 직렬화 방식을 설정합니다. org.apache.spark.Serializer의 하위 클래스를 지정할 수 있으며, 현재 스파크에서는 JavaSerializer와 KyroSerializer라는 두 클래스를 제공하고 있습니다. 

다음 메모리 관련 설정, 익스큐터 관련 설정, 네트워크 관련 설정, 보안 관련 설정, 암호화 관련 설정은 다음 포스팅에서 하도록 하겠습니다.

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

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

실제로 실무에서 스파크로 작업된 결과를 hdfs에 남기기전에 coalesce명령어를 써서 저장되는 파일의 개수를 지정해주곤 했다.

업무에서 사용하긴 했지만 실제 repartition연산과 어떤점이 다른지 모르고 사용했었는데 책을 보며 알게되어 기록.

핵심은 셔플을 하느냐 안하느냐!!!

coalesce와 repartition

오우.....이런 중요한 차이점이 있었다니....그렇다면 coalesce를 사용하면 셔플을 발생시키지 않기때문에 파티션마다 데이터의 사이즈가 다를꺼고 hdfs write했을때 repartition으로 개수를 조정한것과는 다르게 사이즈가 뒤죽박죽이겠네?!!! (나중에 시간되면 테스트해보자)

[ 업데이트 내용 ] 

댓글에서 관련내용에 대해 적어주신분이 있어 확인할겸 관련 내용 업데이트 합니다.

실제 repartition내부는 coalesce메소드를 호출하는 형태로 되어있습니다.

coalesce내부 소스코드도 올려봅니다.

소스코드의 주석을 보면 'This results in a narrow dependency' 좁은 의존성을 초래한다고 적혀 있는데 관련해서는 따로 포스팅하도록 하겠습니다.

그리고 위에서는 coalesce는 파티션 수를 줄이는 것만 가능하다고 적어놨지만 'true'옵션을 주면 늘리는 것 또한 가능하네요.

하지만 기존 처리하던 partitions의 개수보다 많은 파티션수로 처리할 경우에는 반드시 shuffle옵션을 true로 주셔야합니다(매개변수로 넘겨주면됨)

이상으로 포스팅을 마치도록 하겠습니다.

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

'빅데이터 분석을 위한 스파크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)' 내용을 학습하다가 정리한 내용이다.

스프링부트(SpringBoot)로 카프카에서 데이터를 consume에 처리해야 하는 일이 있어 작업 중 발생한 에러에 대해 간단히 남겨본다...

나중에 같은 실수를 하지 않기 위해...

일단 현재 사용하고 있는 카프카 버전은 0.8.2.0이고 spring-integration-kafka를 사용하였다.

실제 클라우데라에 설치된 카프카 버전

카프카 낮은 버전을 사용하고 있어 요즘 spring-kafka 연동 가이드로는 에러가 발생해 삽질을 좀 하였다...

시간이 된다면 0.8.2.2 consume 모듈을 git에 올려서 링크 걸어보도록 하겠다.

다음과 같이 KafkaConfig를 설정해주고

[ 스프링 설정 및 코드 ]

@EnableIntegration
@Configuration
public class KafkaIntegration {

    private static final String BOOTSTRAP_SERVER = "serverIp:2181";
    private static final String ZOOKEEPER_CONNECT = "serverIp:2181";

    @Getter
    @Component
    public static class KafkaConfig {
        private String topic = "data_log";
        private String brokerAddress = BOOTSTRAP_SERVER;
        private String zookeeperAddress = ZOOKEEPER_CONNECT;

        KafkaConfig(){}

        public KafkaConfig(String t, String b, String zk) {
            this.topic = t;
            this.brokerAddress = b;
            this.zookeeperAddress = zk;
        }
    }
}

Consumer 빈 등록

@Configuration
public class ConsumerConfiguration {

    @Autowired
    private KafkaIntegration.KafkaConfig kafkaConfig;

    @Bean
    public IntegrationFlow consumer() {

        KafkaHighLevelConsumerMessageSourceSpec messageSourceSpec = Kafka.inboundChannelAdapter(
                new ZookeeperConnect(this.kafkaConfig.getZookeeperAddress()))
                .consumerProperties(props -> props.put("auto.offset.reset", "smallest")
                        .put("auto.commit.interval.ms", "100"))
                .addConsumer("ectc_dsp_test", metadata -> metadata.consumerTimeout(100)
                    .topicStreamMap(m -> m.put(this.kafkaConfig.getTopic(), 1))
                    .maxMessages(10).valueDecoder(String::new));

        Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer = 
                e -> e.poller(p -> p.fixedDelay(100));

        return IntegrationFlows
                .from(messageSourceSpec, endpointConfigurer)
                .<Map<String, List<String>>>handle((payload, headers) -> {
                    payload.entrySet().forEach(
                            e -> System.out.println((e.getKey() + '=' + e.getValue())));
                    return null;
                })
                .get();
    }

}

으로 등록 후 어플리케이션을 실행하면 다음과 같은 에러메시지가 발생하였다.

[ 발생한 에러메세지 ]

분명 로그를 봐도 내가 KafkaConfig에서 설정해준 서버 IP와 정상적인 커넥션을 맺었는데 왜이런단 말인가???

[ 해 결 ]

주키퍼(Zookeeper) 내부적으로 클러스터간 통신시 혹은 zookeeper to kafka간 통신시 서버의 IP보다는 호스트명으로 서로를 인지한다는 말?을 들은적이 있어 로컬 host파일에 서버의 호스트와 IP를  등록하고 다시 실행해보았더니 정상적으로 카프카에서 메세지를 consume하는 것을 확인할 수 있었다.

혹시나 다음과 같은 문제가 발생한다면 host파일에 서버의 호스트명과 IP를 등록 후 다시 해보길...

정확한 원인은 나중에 관련한 문서나 Zookeeper를 좀더 깊게 공부하게 되어 발견하게 된다면 추후 또 포스팅해보도록 하겠습니다. 

ref : https://spring.io/blog/2015/04/15/using-apache-kafka-for-integration-and-data-processing-pipelines-with-spring