LRU캐시란

캐시 evict 규칙을 가장 오래동안 사용하지 않는 순으로 삭제하는 것이다.

LRU캐시 구현

LinkedList와 LinkedHashMap으로 쉽게 구현이 가능하다. LinkedList보다 LinkedHashMap으로 구현한게 성능은 더 좋다.

LinkedList 구현

import java.util.LinkedList;

public class LRUList<T> {

  private final LinkedList<T> cache = new LinkedList<>();
  private int cacheSize;

  public LRUList(int cacheSize) {
    if (cacheSize <= 0) {
      throw new IllegalArgumentException("The maximum of the size should be a positive integer.");
    }
    this.cacheSize = cacheSize;
  }

  public T getData(T data) {
    // cache hit
    if(cache.remove(data)){
      cache.addFirst(data);
      // cache miss
    } else {
      int currentSize = cache.size();

      if(currentSize == cacheSize){
        cache.pollLast();
      }
      cache.addFirst(data);
    }
    return cache.getFirst();
  }

}

LinkedHashMap 구현

import java.util.LinkedHashMap;
import java.util.Map;

public class LRUHashMap<K, V> extends LinkedHashMap<K, V> {

  private int size;

  private LRUHashMap(int size) {
    super(size, 0.75f, true);
    this.size = size;
  }

  protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
    return size() > size;
  }

  public static <K, V> LRUHashMap<K, V> newInstance(int size) {
    return new LRUHashMap<K, V>(size);
  }
}

LinkedHashMap에서 accessOrder를 true로 주면 LRU처럼 동장한다. 만약 ‘false`이면 insert되는 순서로 저장된다.

성능이 중요한 멀티쓰레딩 환경에서 LRU 캐시를 사용할 때

import java.io.Serializable;
import java.util.LinkedHashMap;
import java.util.Set;

public class LRUMap<K, V> implements Serializable {

  private static final long serialVersionUID = 99123123123L;

	// 성능 개선을 위해  accessOrder를   변경
  private final LinkedHashMap<K, V> cache = new LinkedHashMap<>(0, 0.75f, false); 

  private int cacheSize;

  public LRUMap(int cacheSize) {
    if (cacheSize <= 0) {
      throw new IllegalArgumentException("The maximum of the size should be a positive integer.");
    }
    this.cacheSize = cacheSize;
  }

  // get과 put이 동시에 발생하므로 access에 대한 put을 별도로 진행하지 않음
  public V get(K key) {
    synchronized (cache) {
      return cache.get(key);
    }
  }

  public void put(K key, V v) {
    this.put(key, v, true);
  }

  public void put(K key, V v, boolean check) {

    if (check) {

      if (key == null || v == null) {
        throw new IllegalArgumentException("The key or the value is null.");
      }

      synchronized (cache) {

        if (cache.containsKey(key)) {
          cache.remove(key);
        } else {
          if (cache.size() >= cacheSize) {
            cache.remove(cache.keySet().iterator().next());
          }
        }

        // 테스트를 통해 삭제 후 put을 수행하는 것이 유리함을 확인함
        cache.put(key, v);
      }

    } else {

      cache.put(key, v);
    }
  }

  public void remove(K k) {
    synchronized (cache) {
      cache.remove(k);
    }
  }

  public boolean contains(K k) {
    synchronized (cache) {
      return cache.containsKey(k);
    }
  }

  public int size() {
    synchronized (cache) {
      return cache.size();
    }
  }

  public Set<K> keySet() {
    return cache.keySet();
  }

  @Override
  public String toString() {
    return "LRUMap{" +
      "cache=" + cache +
      ", cacheSize=" + cacheSize +
      '}';
  }
}

“이 포스팅은 쿠팡 파트너스 활동의 일환으로, 이에 따른 일정액의 수수료를 제공받습니다.”

Parquet rewriter만들기 작업

회사에서 parquet로 적재된 데이터를 특정row의 특정컬럼 데이터를 Null로 update하는 작업을 수행하던 중 발견한 이슈이다.

impala로 insert overwirte하면, 불필요한 file I/O가 발생하여 직접 parquet rewriter를 만들어 rowGroup단위로 필요한 부분을 update하는 엔진을 만들었다. 여기서 최적화를 한번더하면, rowGroup보다 더 작은 단위인 page단위로 update를 하는 처리를 할 수 있는데, rowGroup단위로 해도 충분히 성능이 나와 stop했음 해당 엔진을 만들 때 참고한 소스)

Java를 이용해서 Hadoop Parquet read/write를 하기 위해서 많이 사용되는 것이 avro-parquet라이브러리 일 것이다. 하지만 이 라이브러리에서는 INT96 type (Decimal, timestamp)를 지원하지 않는다. (parquet-format은 INT96타입을 지원하지 않음 참고)

Decimal타입을 사용하지 않으면, 굳이 해당 내용을 생각할 필요가 없다.

    PrimitiveType asPrimitive = parquetType.asPrimitiveType();
    PrimitiveTypeName parquetPrimitiveTypeName = asPrimitive.getPrimitiveTypeName();
    final OriginalType annotation = parquetType.getOriginalType();
    Schema schema = (Schema)parquetPrimitiveTypeName.convert(new PrimitiveTypeNameConverter<Schema, RuntimeException>() {
    public Schema convertBOOLEAN(PrimitiveTypeName primitiveTypeName) {
        return Schema.create(Type.BOOLEAN);
    }

    public Schema convertINT32(PrimitiveTypeName primitiveTypeName) {
        return Schema.create(Type.INT);
    }

    public Schema convertINT64(PrimitiveTypeName primitiveTypeName) {
        return Schema.create(Type.LONG);
    }

    public Schema convertINT96(PrimitiveTypeName primitiveTypeName) {
        throw new IllegalArgumentException("INT96 not yet implemented.");
    }

    public Schema convertFLOAT(PrimitiveTypeName primitiveTypeName) {
        return Schema.create(Type.FLOAT);
    }

    public Schema convertDOUBLE(PrimitiveTypeName primitiveTypeName) {
        return Schema.create(Type.DOUBLE);
    }

    public Schema convertFIXED_LEN_BYTE_ARRAY(PrimitiveTypeName primitiveTypeName) {
        int size = parquetType.asPrimitiveType().getTypeLength();
        return Schema.createFixed(parquetType.getName(), (String)null, (String)null, size);
    }

    public Schema convertBINARY(PrimitiveTypeName primitiveTypeName) {
        return annotation != OriginalType.UTF8 && annotation != OriginalType.ENUM ? Schema.create(Type.BYTES) : Schema.create(Type.STRING);
    }

위 코드는 avro-parquet 내부소스인데, INT96 type에 대해서 컨버터를 만들어 주지 않아.

java.lang.IllegalArgumentException: INT96 not yet implemented

Exception을 표시한다.

Parquet-writer를 만들때 이슈를 정리하면 다음과 같다.

배경지식

• Parquet에서 INT96을 지원하지 않는다.
• INT96은 우리가 일반적으로 아는 Decimal, timestamp type이 저장되는 데이터저장공간이다.
• Kudu에서는 decimal type이 지원함

이슈

• kudu talbe을 impala을 이용해 parquet table만들 때, decimal type을 INT96으로 저장하고 읽을때 메타스토어를 refer하여 정상적으로 읽음.(HUE, Impala사용자는 문제없음)
• spark에서는 Parquet read시 INT96을 무조건 timestamp로 치환해 읽음(Spark사용자들은 decimal type값을 정상적으로 읽을 수 없음)
• parquet-rewriter작업만들때 INT96타입을 처리하는 별도 로직이 필요하고, 구성원들의 합의가 필요함.

결정이 필요한 내용

• parquet-rewriter작업은 INT96타입을 어떻게 처리하면 좋을까요?

Reference

parquet공식

• https://issues.apache.org/jira/browse/PARQUET-323

parquet소스

• https://github.com/apache/parquet-mr/blob/master/parquet-avro/src/main/java/org/apache/parquet/avro/AvroSchemaConverter.java#L279

Spark소스

• https://github.com/apache/spark/blob/a988aaf3fa38e89a05ab644ce79ddb14af56859d/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaConverter.scala#L154

Sets which Parquet timestamp type to use when Spark writes data to Parquet files.
INT96 is a non-standard but commonly used timestamp type in Parquet.
TIMESTAMP_MICROS is a standard timestamp type in Parquet, which stores number of microseconds from the Unix epoch.
TIMESTAMP_MILLIS is also standard, but with millisecond precision, which means Spark has to truncate the microsecond portion of its timestamp value.

converter관련 소스

• https://stackoverflow.com/questions/54657496/how-to-write-timestamp-logical-type-int96-to-parquet-using-parquetwriter

올해 목표 조기달성!

2019년 1월 올해는 꼭 해야할 일을 정리했다. 그 중 OpenSource활동하기라는 막연한 계획이 있었다.

2월에 Cassandra에 BulkLoad하는 기능이 필요해 이것 저것 찾던 중에 요구사항에 만족할 만한 것들이 없었다. 그래서 한번 내가 만들어 볼까라는 생각을 했고, 그렇게 회사 프로젝트에 CassandraBulkLoad 모듈을 만들어 적용했다. 하지만 원시 소스형태였고, 사용하기 불편한 형태의 코드였다.

작업한 코드는 BulkLoad하는 동안 Cassandra Read성능이 크게 떨어지지 않고, BulkLoad를 하는 방법이 단순해서 이거 Spark Lib으로 제작해보자라는 생각만 가지고 시간을 보냈다.

8월에 회사를 퇴사하게되면서, 자유시간이 많아졌다. 2월부터 생각만 했던 작업이 생각났고 자연스레 코드를 다시보았다. 그리고 Spark Lib형태로 제작하기 시작했다. 그렇게 탄생한게 Spark2CassandraBulkLoad이다.

해당 현재 해당 Repo는 Star가 5개이고, 대부분 프로젝트 내용을 알고 있는 회사사람들이다.

Lib을 제작하면서, 처음으로 JCenter에 배포도 해보고 spark-package.org에 등록도 하였다.

퇴사를 하면서 intellij License가 만료되어 걱정이였는데, 해당 repo로 JetBrains에 opensource lincense를 신청해 1년치 All Products License 받았다.

3년 회사생활를 잘 마무리한 것 같아 뿌듯하다. 다음 주부터 새로운 집에서 새로운 회사로 출근한다.

2019년 남은 4개월동안 새로운 곳에서 의미있는 일을 한번 해보자!

Java8 Date함수 정리

제발쫌 기억하자!!

현재 그리고 앞으로의 프로젝트에서 java8이하 버전을 사용할 가능성이 거의 없다고 생각한다. 그래서 java8 java.time.*을 정리하고자 한다.

(데이터를 날짜별로 분류하거나 로드할때 날짜 연산이 많이 사용되는데, 할때마다 날짜관련 메소드가 햇갈려서 이번에 정리하고자 한다.)

클래스

1. 날짜 (Temporal)

• Instant : machine time에 유용한 1970년 1월 1일부터 시간을 세는 클래스 (millisecond 뿐만 아니라 nanosecond까지 센다)
• LocalDate : [년,월,일]과 같은 날짜만 표현하는 클래스 (시간은 포함하지 않는다)
• LocalDateTime : [년,월,일,시,분,초]를 표현하는 클래스
• LocalTime : [시,분,초]와 같이 시간만 표현하는 클래스

2. 기간 (TemporalAmount)

• Period : 두 날짜 사이의 [년,월,일]로 표현되는 기간 (시간을 다루지 않다 보니 LocalDate를 사용한다)

• Duration : 두 시간 사이의 [일,시,분,초]로 표현되는 기간 (Instant 클래스를 사용하고, seconds와 nanoseconds로 측정 되지만 [일,시,분,초]로 변환해 주는 메쏘드를 제공)

기타

• ChronoUnit : 한가지의 단위를 표현하기 위한 클래스 (년,월,일,시,분,초 등)
• DayOfWeek : 요일

기억하기!

날짜 가져오기

  LocalDate.now(); // 오늘
  LocalDateTime.now(); // 지금
  LocalDate.of(2019, 7, 22); // 2019년7월22일
  LocalDateTime.of(2019, 7, 22, 23, 23, 50); // 2019년7월22일23시23분50초
  Year.of(2017).atMonth(7).atDay(22).atTime(10, 30); // 2019년7월22일 10시30분00초

기간 가져오기

  Period.ofYears(2); // 2년간(P2Y)
  Period.ofMonths(5); // 5개월간(P5M)
  Period.ofWeeks(3); // 3주간(P21D)
  Period.ofDays(20); // 20일간(P20D)

  Duration.ofDays(2); // 48시간(PT48H)
  Duration.ofHours(8); // 8시간(PT8H)
  Duration.ofMinutes(10); // 10분간(PT10M)
  Duration.ofSeconds(30); // 30초간(PT30S)

날짜 + 기간 = 날짜

  LocalTime.of(9, 0, 0).plus(Duration.ofMinutes(10)); // (9시 + 10분간) = 9시10분
  LocalDate.of(2019, 7, 22).plus(Period.ofDays(1)); // (2019년7월22일 + 1일간) = 2019년7월23일
  LocalDateTime.of(2019, 7, 1, 23, 47, 5).minus(Period.ofWeeks(3)); // (2019년7월1일 23시47분05초 - 3주간) = 2019년6월10일 23시47분05초
  LocalDate.now().plusDays(1); // (오늘 + 1일) = 내일
  LocalTime.now().minusHours(3); // (지금 - 3시간) = 3시간 전

날짜 - 날짜 = 기간

  Period.between(LocalDate.of(1950, 6, 25), LocalDate.of(1953, 7, 27)); // (1953년7월27일 - 1950년6월25일) = 3년1개월2일간(P3Y1M2D)
  Period.between(LocalDate.of(1950, 6, 25), LocalDate.of(1953, 7, 27)).getDays(); // 3년1개월2일간 => 2일간
  LocalDate.of(1950, 6, 25).until(LocalDate.of(1953, 7, 27), ChronoUnit.DAYS); // 3년1개월2일간 => 1128일간
  ChronoUnit.DAYS.between(LocalDate.of(1950, 6, 25), LocalDate.of(1953, 7, 27)); // 3년1개월2일간 => 1128일간

  Duration.between(LocalTime.of(10, 50), LocalTime.of(19, 0)); // (19시00분00초 - 10시50분00초) = 8시간10분간(PT8H10M)
  Duration.between(LocalDateTime.of(2015, 1, 1, 0, 0), LocalDateTime.of(2016, 1, 1, 0, 0)).toDays(); // 365일간
  ChronoUnit.YEARS.between(LocalDate.of(2015, 5, 5), LocalDate.of(2017, 2, 1)); // 1년간

날짜 변환하기

- LocalDate -> String

LocalDate.of(2020, 12, 12).format(DateTimeFormatter.BASIC_ISO_DATE); // 20201212

- LocalDateTime -> String

LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")); // 2015-04-18 00:42:24

- LocalDateTime -> java.util.Date

Date.from(LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant()); // Sat Apr 18 01:00:30 KST 2015

- LocalDate -> java.sql.Date

Date.valueOf(LocalDate.of(2015, 5, 5)); // 2015-05-05

- LocalDateTime -> java.sql.Timestamp

Timestamp.valueOf(LocalDateTime.now()); // 2015-04-18 01:06:55.323

- String -> LocalDate

LocalDate.parse("2002-05-09"); // 2002-05-09
LocalDate.parse("20081004", DateTimeFormatter.BASIC_ISO_DATE); // 2008-10-04

- String -> LocalDateTime

LocalDateTime.parse("2007-12-03T10:15:30"); // 2007-12-03T10:15:30
LocalDateTime.parse("2010-11-25 12:30:00", DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")); // 2010-11-25T12:30

- java.util.Date -> LocalDateTime

LocalDateTime.ofInstant(new Date().toInstant(), ZoneId.systemDefault()); // 2015-04-18T01:16:46.755

- java.sql.Date -> LocalDate

new Date(System.currentTimeMillis()).toLocalDate(); // 2015-04-18

- java.sql.Timestamp -> LocalDateTime

new Timestamp(System.currentTimeMillis()).toLocalDateTime(); // 2015-04-18T01:20:07.364

- LocalDateTime -> LocalDate

LocalDate.from(LocalDateTime.now()); // 2015-04-18

- LocalDate -> LocalDateTime

LocalDate.now().atTime(2, 30); // 2015-04-18T02:30

- 요일로 날짜 가져오기

  LocalDate.now().with(TemporalAdjusters.next(DayOfWeek.SATURDAY)); // 다음 토요일
  LocalDate.of(2016, 5, 1).with(TemporalAdjusters.dayOfWeekInMonth(3, DayOfWeek.SUNDAY)); // 2016년5월 세번째 일요일
  LocalDate.of(2015, 7, 1).with(TemporalAdjusters.firstInMonth(DayOfWeek.MONDAY)); // 2015년7월 첫번째 월요일

- 언어별 출력

  DayOfWeek.MONDAY.getDisplayName(TextStyle.FULL, Locale.ENGLISH); // Monday
  DayOfWeek.MONDAY.getDisplayName(TextStyle.NARROW, Locale.ENGLISH); // M
  DayOfWeek.MONDAY.getDisplayName(TextStyle.SHORT, Locale.ENGLISH); // Mon

  DayOfWeek.MONDAY.getDisplayName(TextStyle.FULL, Locale.KOREAN); // 월요일
  DayOfWeek.MONDAY.getDisplayName(TextStyle.NARROW, Locale.KOREAN); // 월
  DayOfWeek.MONDAY.getDisplayName(TextStyle.SHORT, Locale.KOREAN); // 월

  Month.FEBRUARY.getDisplayName(TextStyle.FULL, Locale.US); // February
  Month.FEBRUARY.getDisplayName(TextStyle.FULL, Locale.KOREA); // 2월

Reference

• https://jekalmin.tistory.com/entry/%EC%9E%90%EB%B0%94-18-%EB%82%A0%EC%A7%9C-%EC%A0%95%EB%A6%AC

NoSQL 아키텍처를 이해기 위한 용어정리

개요

NoSQL, 분산스토리지를 사용하면서 많이 들었던 용어를 정리할 필요가 있다고 느껴 정리하고자 한다.

1. Bloom Filter

블룸필터(Bloom Filter)는 1970년도에 Burton H. Bloom이 고안한 것으로 공간 효율적인 probabilistic data structure이며 구성요소가 집합의 구성원인지 점검하는데 사용된다. False positive(없다고 예측하고 실제로 없는경우)들이 가능하며, false negative(없다고 예측하고 실제로 있는경우)들은 불가능하다. 요소들은 집합에 추가될 수 있으나 제거는 되지 않는다. 그리고 블룸 필터는 메모리를 좀 더 사용함으로써 자주 호출되는 비싼 함수들의 성능을 크게 향상시키는 방법 중에 하나이다.

(즉 집합에 원소가 없으면 진짜 없다는 것을 보장해준다.)

구성 요소들과 작동 방식

구성 요소

• 해시(hash) 함수 : 해시 알고리즘 구현체.
• Bloom Filter Key : 해시 알고리즘에 의해 반환된 KEY.
• Bloom Filter Index : Bloom Filter Key를 임의의 크기의 비트들의 블록으로 나눈 하나 하나.
• 블룸 필터는 두가지를 제공 : 하나는 추가하는 add(), 하나는 존재하는지에 대한 isExist().
• fFalse positive(없다고 예측하고 실제로 없는경우), false negative(없다고 예측하고 실제로 있는경우)
• 작동 방식 예를 들어, m(bloom filter의 bit 사이즈) = 10, k(hashing functions의 수) = 2로하자. 2개의 해시 함수를 h1, h2 하고 Bloom Filter 배열은 아래와 같다.

    +-+-+-+-+-+-+-+-+-+-+
    |0|0|0|0|0|0|0|0|0|0|
    +-+-+-+-+-+-+-+-+-+-+
     0 1 2 3 4 5 6 7 8 9

데이터 A를 삽입한다고 가정하면, 데이터 A를 두 해시 함수 h1, h2를 거쳐 해시 값을 계산한다. 만일 3=h1(A), 7=h2(A) 였다고 하자. 이 해시 값 3, 7에 해당하는 곳에 1을 셋팅한다. 이제 배열은 아래와 같아진다.

    +-+-+-+-+-+-+-+-+-+-+
    |0|0|0|1|0|0|0|1|0|0|
    +-+-+-+-+-+-+-+-+-+-+
     0 1 2 3 4 5 6 7 8 9

그 다음 데이터 A를 검색한다면, 해시 함수를 거치면 3=h1(A), 7=h2(A)이다. 3, 7에 대응하는 배열을 보면 양쪽 모두 1로되어 있어서 리턴은 true로 던저질 것이다.

    +-+-+-+-+-+-+-+-+-+-+
    |0|0|1|1|0|0|0|1|0|0|
    +-+-+-+-+-+-+-+-+-+-+
     0 1 2 3 4 5 6 7 8 9
           *       *

그 다음 데이터 C를 검사한다면, 해시 함수를 거치면 3=h1(C), 6=h2(C)이다. 그런데 3은 1이나 6이 0이므로 false가 리턴되어 없는 것으로 간주하게 된다.

    +-+-+-+-+-+-+-+-+-+-+
    |0|0|1|1|0|0|0|1|0|0|
    +-+-+-+-+-+-+-+-+-+-+
     0 1 2 3 4 5 6 7 8 9
           *     *

이런 형태로 동작하는 방식이다. Bloom Filter는 기존 hash table의 key-value의 쌍으로 자정하지 않고 hash table상에 키가 존재하는지 안하는지 true, false 정보의 조각(Bit의 Set)이 저장된다고 보면 된다. 즉, 각 Value를 저장하는 대신에 Bloom Filter는 Key가 존재하는 지점(해시 함수에 의해)을 가르키는 Bits의 배열이라고 봐도 된다. 찾는 비트의 조합이 모두 1일 경우 제대로 찾을 확률이 높아진다는 것이다. Space와 Time을 적절하게 배합하고 고른 분포도를 가지며 Collision을 대처한다면 좋은 이득을 얻을 수 있다. 많은 비트를 할당할수록 성능은 좋을 수 있으나 많은 메모리가 필요하고 해싱 함수를 늘리게 되면 연산이 많아지게 되어 성능은 느리나 메모리를 덜 차지하게 되는 trade-off 관계가 존재한다. Optimizing하는게 관건이다.

2. Vector Clocks

https://medium.com/@balrajasubbiah/lamport-clocks-and-vector-clocks-b713db1890d7

3. Gossip Protocol

Decentralization, Partition tolerance를 지원하기 위해 Cassandra는 Gossip Protocol을 사용한다. Gossip Protocol은 각 노드가 클러스터의 다른 노드의 상태정보를 추적하는 것이고 이는 매초 발생한다.

• Peer to peer communication protocol.
• Nodes are arranged in ring format.
• Data is replicated to multiple nodes.
• Nodes periodically exchange info. they have.
• Nodes also exchange their own info.
• Each message has its associated version.
• No master-slave concept, and hence no single point of failure.

Gossip protocol 동작 방식

1. 초당 1 회, gossiper는 클러스터에서 임의의 노드를 선택하고 그 노드와 함께 gossip세션을 초기화합니다.
2. gossip initiator는 선택한 친구에게 GossipDigestSynMessage를 보냅니다.
3. 친구가 메시지를 받으면 GossipDigestAckMessage를 반환합니다.
4. initiator가 친구로부터 ack 메시지를 받으면 친구에게 GossipDigestAck2Message를 전송하여 gossip 라운드를 완료합니다
5. 만약 친구로 부터 ack를 받지 못하면 해당 노드를 convicts로 마크합니다.

4. Two Phase Commit

Two Phase Commit(a.k.a. 2PC)은 distributed system에서 atomic commit을 보장하는 프로토콜이다.

2PC에서 노드는 한 개의 coordinator와 여러 개의 cohort로 나누어진다. Coordinator는 commit 할 transaction을 만드는 노드고, cohort들은 coordinator가 보낸 transaction을 commit 하거나 revert 한다. 2PC는 이때 fail 하지 않은 모든 cohort가 같은 상태를 유지하도록 하는 것이다. 즉, fail 하지 않은 모든 노드는 다 같이 commit 하거나 revert 한다.

이때 coordinator를 어떻게 선정할지는 2PC의 영역이 아니다. 고정된 coordinator를 계속 사용할 수도 있고, 차례대로 돌아가면서 coordinator가 될 수도 있고, 별도의 leader election을 사용하여 coordinator를 선정할 수도 있다. 2PC는 어떻게든 coordinator가 선정된 뒤의 일이다.

2PC는 이름 그대로 2가지 phase로 나누어져 있다. 첫 번째 phase는 voting phase라고 부르고, 두 번째 phase는 commit phase라고 불린다. 각 phase의 시작은 coordinator가 보내는 메시지로 시작한다.

Voting phase에서 coordinator는 commit 하고 싶은 transaction을 commit 할지 말지 투표하는 요청을 cohort에게 보낸다. VOTE 메시지를 받은 cohort들은 이 transaction을 바로 commit 하지 않는다. 해당 transaction을 커밋할 수 있으면 YES 메시지를, 없으면 NO 메시지를 coordinator에게 보낸다.

Voting phase에서 coordinator가 quorum 이상의 YES 메시지나 NO 메시지를 모으면 commit phase를 시작한다. 이때 일부 cohort에 문제가 생겨서 더 진행되지 않는 것을 방지하기 위해서 일정 시간 응답을 주지 않는 cohort는 NO 메시지를 보냈다고 가정한다.

이때 quorum을 얼마로 잡는가에 따라서 시스템의 consistency model과 resilience가 결정된다. 예를 들어 N개의 coordinator가 있는 시스템에서 N개의 YES 메시지를 모아야 한다면, 하나의 failure도 용납하지 않는 resilient 하지 않지만, strong consistency를 보장하는 시스템이 된다. Quorum이 얼마가 되어야 하는지는 정해지지 않았다. 하지만 non-partition 상황에서 consistency를 보장하기 위해서는 최소 N/2 이상의 YES 메시지를 모아야 한다.

Coordinator가 quorum 이상의 YES 메시지를 받았으면 cohort들에게 COMMIT 메시지를 보내고, quorum 이상의 NO 메시지를 받았으면 cohort 들에게 ROLLBACK 메시지를 보낸다. cohort는 COMMIT 메시지를 받았으면 voting phase에서 받았던 transaction을 커밋하고, ROLLBACK 메시지를 받았으면 그 transaction을 버린다. COMMIT이든 ROLLBACK이든 메시지를 처리하고 나면 cohort는 coordinator에게 처리했다는 메시지를 보낸다. Coordinator가 cohort들에게 처리가 끝났다는 메시지를 받으면 commit phase가 끝난다.

위의 과정을 거쳐 2PC가 진행된다. 앞서 말했듯이 2PC는 괜찮은 resilience를 보이면서, 성능도 나쁘지 않기 때문에 많이 사용된다. 특히 atomic commit을 지원하는 프로토콜 중에서는 가장 적은 메시지 수로 commit 될 수 있는 프로토콜이다.

하지만 2PC에는 심각한 문제가 하나 있다. 2PC는 VOTE 메시지를 보낸 coordinator가 죽어서 COMMIT이나 ROLLBACK 메시지를 보내지 못하면 YES 메시지를 보낸 cohort가 안전하게 상태를 회복할 방법이 없다. 이는 YES 메시지를 보낸 cohort의 상태가 undefined이기 때문이다.

5. CAP

“분산 시스템에서는 3개 속성을 모두 가지는 것이 불가능하다!” 이다. 각 속성은 아래 3가지이다.

Consistency (일관성)

• CAP이론에서 말하는 Consistency는 ACID의 ‘C’가 아니다! ACID의 ‘C’는 “데이터는 항상 일관성 있는 상태를 유지해야 하고 데이터의 조작 후에도 무결성을 해치지 말아야 한다”는 속성이다.
• CAP의 ‘C’는 “Single request/response operation sequence”의 속성을 나타낸다. 그 말은 쓰기 동작이 완료된 후 발생하는 읽기 동작은 마지막으로 쓰여진 데이터를 리턴해야 한다는 것을 의미한다.
• 모든 노드가 같은 시간에 같은 데이터를 보여줘야 한다. (저장된 데이터까지 모두 같을 필요는 없음)

Availability (가용성)

• “특정 노드가 장애가 나도 서비스가 가능해야 한다”라는 의미를 가진다.
• 데이터 저장소에 대한 모든 동작(read, write 등)은 항상 성공적으로 리턴되어야 한다.
• 명확해 보이는 단어이기는 하지만 분산 시스템에서의 특징을 말하는 것이기 때문에 “서비스가 가능하다”와 “성공적으로 리턴”이라는 표현이 애매하다. 얼마동안 기다리는 것 까지를 성공적이라고 할 수 있느냐에 대한 문제가 남아있다. “20시간정도 기다렸더니 리턴이 왔어! Availability가 있는 시스템이야!”라고 할 수 없기 때문이다.

Partitions Tolerance (분리 내구성)

• Tolerance to network Partitions인데 보통은 Partition-tolerance라고도 한다.
• 노드간에 통신 문제가 생겨서 메시지를 주고받지 못하는 상황이라도 동작해야 한다.
• Availablity와의 차이점은 Availability는 특정 노드가 “장애”가 발생한 상황에 대한 것이고 Tolerance to network Partitions는 노드의 상태는 정상이지만 네트워크 등의 문제로 서로간의 연결이 끊어진 상황에 대한 것이다

6. Eventual Consistency

서버가 여러대인 분산시스템에서 데이터를 조회했을 때 특정 서버는 변경된 데이터가 조회되고 일부는 변경되지 않은 상태로 조회될 수 있다. 그 때 데이터의 일관성을 위해서 모든 서버에 결과값을 질의하고 N개 이상이 같은 값을 반환할 때 사용자에게 해당 값을 보여주는 것을 의미한다.

ex > Cassandra에서 Consistency read level을 설정하는 것

7. Column Storage

정의

데이터의 저장을 칼럼단위로 처리하는 데이터베이스를 말한다.

장점

칼럼 단위의 값은 데이터가 유사할 가능성이 높다. 이로 인해 높은 압축율을 얻을 수 있다.

MIN, MAX, SUM, COUNT 와 같은 연산에서 높은 성능을 얻을 수 있다.

Spark SQL jdbc사용시 주의할 사항

개요

Spark로 작업한 결과를 Mysql로 적재하는 일이 있었다.

자주 이용하는 방식은 Mysql Connection관리와 key 중복이 발생할때 update를 하기 위해서 아래 두가지 방식을 많이 사용했다.

• collect()하여 드라이버노드에서 insert 하는 방식
• RDD로 변형하여foreachPartition를 돌면서 insert 하는 방식

하지만 데이터가 매우 커지면 위 방식들은 OOM이나, IOException을 발생 시켰다.

그래서 Dataset의 jdbc write 방식을 이용하게 되었다. 이를 이용하면서 몇가지 문제점을 기록하고자 한다.

jdbc write 사용방법

val prop = new Properties()
prop.setProperty("driver", "com.mysql.jdbc.Driver")
prop.setProperty("user", "mysql.user")
prop.setProperty("password", "mysql.password")

resultDataset.write.mode(SaveMode.Append)
	.jdbc("Mysql.url", "tableName", prop)

이렇게만 하면 각 executor에서 insert문을 실행한다.

주의점

SaveMode에 따라서 Mysql table을 다루는 방식이 다른다.

• SaveMode.Overwrite
  • 관련 table을 drop - create를 한뒤 insert를 진행한다.
  • 2.1버전 이상부터는 truncate옵션을 줘 drop table를 방지할 수 있다.(default는 false, 사용법 : .option("truncate", true))
• SaveMode.Append
  • insert가 진행되는데, key겹쳐면 insert를 진행함 하지만 mysql에서 에러를 던짐
• SaveMode.Ignore
  • insert가 진행되는데, key겹쳐면 isnert를 진행안함 mysql까지 안감
• SaveMode.ErrorIfExists
  • default option
  • insert가 진행되는데, key겹쳐면 isnert를 진행안함 mysql까지 안감, spark에서 에러를 던짐

사용하는 점에서는 문제가 없지만 해결할 방법이 없는 경우가 있다.

INSERT INTO table (column_list)
VALUES (value_list)
ON DUPLICATE KEY UPDATE
   c1 = v1, 
   c2 = v2,
   ...;

위 Query처럼 key겹칠때 update를 하고 싶을 때가 있다. 하지만 SparkSql에서는 해당 기능을 지원하지 않는다.

그래서 다른 방법을 사용해야한다.(파일에 쓰고 insert 등)

또하나 주의할 점은 SaveMode.Overwrite에서 cretae 될때 테이블 정보가 없기 때문에 Dataset에 있는 colum정보로 table이 생성된다.

그렇기 때문에 SaveMode.Overwrite를 사용할때는 truncate옵션을 true로 사용하는 것이 좀 더 안전하다.

Reference

• Tips for using JDBC in Apache Spark SQL
• jdbc Spark PR