About Cassandra Compaction

카산드라는 compaction이라는 것이 존재하는데 이는 sstable을 줄여주는 작업이다.

이것이 왜 필요한지 살펴보자.

카산드라는 write를 하면 곧바로 disk에 쓰지 않는다. memtable이라는 메모리 공간에 데이터가 상주하게 된다. (commit log는 disk에 있겠지만..)

1000개의 row를 write 해본다.

memtable size가 커지는 것을 확인할 수 있다. 아직 sstable이 0개이기 때문에 space used가 0이다.

이때 카산드라가 비정상 종료되더라도 commit log가 있기 때문에 복구가 된다고는 하지만 강제적으로 disk로 써보자. 이때 사용할 수 있는 명령이 flush나 drain이다. (memtable->sstable로 데이터 이동)

하지만 이 작업을 강제로 하지 않아도 commit log, metable size가 한계치에 다다르면 자동으로 일어나는 작업이며 cassandra.yaml에서 옵션으로 조절 가능하다.

아래 사진은 ./nodetool drain을 한 직후이다.

sstable이 1개가 되었고 memtable에 있던 데이터가 모두 disk로 쓰여진 모습이다.

카산드라는 데이터가 delete가 일어날 때 실제로 delete를 하지 않는다. memtable과 sstable을 가져가기 때문에 구조상 delete flag를 남길 수 밖에 없고 timestamp를 통해 어떤 데이터가 조회가 되어야하는지 알아낸다.

만약 데이터가 꾸준히 insert/delete가 발생한다면 카산드라는 계속해서 sstable을 만들어낼 것이다. 이는 성능 저하의 원인이 된다.
따라서 sstable을 줄여주는(모아주는) 작업인 compaction이 필요하다!

compaction을 하면 cpu를 많이 사용한다고 하는데 필자의 경우에는 우려할 만한 수준은 아니었고 오히려 대량 insert 배치가 일어날 때 cpu가 90%이상 튀어서 throughput_mb_per_sec을 통해 속도조절을 하며 배치를 돌리는 형편이다.

현재 쓰는 옵션은 아래와 같다. (더 연구해봐야한다.)
.set("spark.cassandra.output.concurrent.writes","1")
.set("spark.cassandra.output.batch.size.rows","1")
.set("spark.cassandra.output.batch.size.bytes","512")
.set("spark.cassandra.output.throughput_mb_per_sec","1")



./nodetool compact


다음은 하루에 한번 강제로 major compaction을 한 모습이다.

drain을 하지 않았기때문에 memtable에 데이터는 남아있지만 sstable count와 size가 줄어든 것을 확인할 수 있다.

아래는 compcation이 발생할 때 cpu 사용율이며 딱히 문제없어보인다. 개인적으로는 cpu 사용율을 높이더라도 compaction 속도를 올리고 싶다.

그래서 setcompactionthroughput을 default 64MB에서 2049MB로 주고 돌려봤으나 별 차이가 없다. 이는 추후에 방법을 찾아봐야겠다.

compaction strategy는 테이블마다 정할 수 있으며 다음과 같은 선택지가 있다.

1. SizeTiredCompationStrategy(STCS)는 쓰기가 많은 테이블
2. LeveledCompactionStrategy(LCS)는 읽기가 많은 테이블
3. DateTieredCompactionStrategy(DTCS)는 시계열 테이블

현재는 1번으로 둔 상태이며 추후에 읽기만 발생하는 테이블은 LCS로 교체해볼 예정이다.

또한 compactionstats 명령을 통해 현재 진행상태를 볼 수 있으며 compactionhistory 명령으로 history를 볼 수 있다.

작업을 하다보니 확실치는 않은데 발견한 것이 있다. 확실치는 않은데 하다보니 발견한 부분은 memtable은 rowcache, keycache가 안타고 sstable에 있는 데이터만 캐시를 타는 것 같다.

cache 정보는 ./nodetool info에서 확인할 수 있다.

key cache ALL로 주어도 용량이 크지 않기때문에 선호한다. (추천)
row cache는 ALL로 주면 테이블에 따라서 무지막지하게 용량을 차지할 수 있다. default는 4GB이고 테스트때문에 capacity를 1MB로 줄여놓은 상태이다. (잘 알고 써야한다.)

off heap memory를 사용하기 때문에 조심해야한다.

기록을 남기기 위해서 주저리 주저리 썼는데 더 연구해 볼 것이 많다.

Cassandra nodetool tablestats(cfstats) - Read Count is always 0

카산드라 모니터링 툴을 Prometheus와 Grafana를 연동해서 쓰고 있다. 그런데 특정 테이블들의 write는 잡히는데 read는 Grafana나 Prometheus에서 수치가 0, NaN 으로만 나왔다.

살펴본 결과 Grafana나 Prometheus, JMX Exporter 등의 문제는 아니었기에 카산드라 자체에서 Read Log를 내뿜지 않는 것으로 결론지었다.

nodetool tablestats -H keyspace.tablename을 날려본 결과이다.

Read Count: 0
Read Latency: NaN ms
Local read count: 0
Local read latency: NaN ms

원인은 select 쿼리에서 allow filltering을 사용한 쿼리들이 잡히지 않는 것이었고 이들은 tablestat에서 잡히지 않았다. 결국 테이블 스키마를 바꿔서(partition key 수정) allow filltering을 사용하면 안되었고 제거하였다.

그 결과 Read Count가 잘 잡힘을 확인할 수 있었다.

MSSQL - When do select into query, insert fails, but the table is created. why!?

다음은 select into 쿼리이고 #tmp라는 테이블에 'A'를 넣기 위함이다.
하지만 bigint로 convert가 되지 않아서 데이터는 들어가지 않는데 #tmp 테이블은 생성되었다.

테스트해보자.

SELECT CONVERT(BIGINT, 'A') col1 INTO #tmp 를 수행하면 분명 convert error가 발생한다. 그리고 바로 이어서 #tmp를 조회해보면 테이블이 생성되었음을 확인할 수 있다.

그럼 혹시 그럼 혹시 convertable variable와 unconvertable variable를 union all로 묶으면 한개라도 들어갈까?

위와 동일한 결과를 얻을 수 있었고 임시 테이블만 생성이 되는 것을 확인할 수 있다.

- 확인 결과(추가)
auto commit mode에서는 select ... into 쿼리는 2개의 트랜잭션으로 이루어져있는데 첫번째는 테이블을 생성하고 두번째는 데이터를 로드한다. 따라서 후속 삽입이 실패하더라도 테이블이 생성된 것은 commit이 된다.

sql server 과거 버전에서는 2개의 작업이 단일 트랜잭션이었고 그 결과 시스템 테이블의 locking&blocking이 유명했다고 한다.

Cannot create multiple column index in Cassandra and allow filtering option

카산드라는 복합키 인덱스 생성이 불가능하다.

왜 그런지 알아보기 전에 카산드라의 데이터 저장구조에 대해서 보자.

다음과 같은 테이블을 생성했다.

CREATE TABLE test.test_tbl (

    col1 text,

    col2 text,

    col3 text,

    col4 text,

    PRIMARY KEY (col1, col2)

);

pk를 col1, col2로 지정했다. pk는 multiple column이 가능하다.

카산드라의 경우 partition key, clustering key가 존재하는데 pk를 지정할 때 partition key, clustering key 순으로 지정한다.

partition key는 hash값에 따라 각 노드에 분산시키고 이를 괄호로 묶어주어 분산시킬 수 있다. 이어서 나오는 키는 clustering key로 노드에서의 데이터 정렬이다. 

카산드라는 where절에 partition key를 지정해서 노드를 찾아가야한다. design essue이다.

where col1 = '1'은 잘 된다는 의미이다. 하지만 where col2 = '1'이나 col3 = '1' 등 partition key를 지정을 하지 않으면 다음과 같은 에러가 뜬다.

InvalidRequest: Error from server: code=2200 [Invalid query] message="Cannot execute this query as it might involve data filtering and thus may have unpredictable performance. If you want to execute this query despite the performance unpredictability, use ALLOW FILTERING"

그럼 혹시 인덱스를 만들면 될까?

CREATE INDEX idx_test_tbl_col2 ON upsell.test_tbl ( col2 );

이 경우에는 select * from test_tbl where col2 = '1'; 이 잘 동작한다.

하지만 select * from test_tbl where col2 = '1' and col3 = '1'; 동일한 에러가 뜬다.

그렇다면 col3에도 동일하게 인덱스를 만들어주면 where col2 = '1' and col3 = '1'가 될까 테스트해보자.

InvalidRequest: Error from server: code=2200 [Invalid query] message="Cannot execute this query as it might involve data filtering and thus may have unpredictable performance. If you want to execute this query despite the performance unpredictability, use ALLOW FILTERING"

역시 동작하지 않는다.

datastax 공식문서에 보면 다음 내용이 있다.

이를 피하기 위해서는 ALLOW FILTERING 옵션을 사용하라고 한다.

select * from test_tbl where col2 = '1' and col3 = '1' ALLOW FILTERING;

하지만 allow filtering은 클러스터의 모든 노드에 query를 날리기 때문에 성능제약이 있다. 카산드라의 경우 hash key를 활용하기 때문에 성능상의 이점이 있는데 이를 포기하는 쿼리가 되는 것이다.

Artificial Intelligence - Gradient descent

대부분 딥러닝을 입문할 때 가장 처음 접하는 개념은 경사하강법, Gradient descent다.
오늘은 그 Gradient Descent에 대해서 포스팅을 해보려고 한다.

ML/DL의 기초가 되는 것이기 때문에 한번은 봐둘만 하다.


단일 선형 회귀로서 공부한 시간에 따라 성적이 매겨진다고 가정한다.

이는 공부시간(x), 성적(y)로 두고 1차 방정식 y=ax+b로 표현할 수 있다. 이는 하나의 직선이 그려지고 잘만 그려진다면 공부시간에 따른 성적을 예측할 수 있을 것이다.

그럼 어떻게 방정식의 a와 b를 구할 수 있을까? 결국 반복시키며(학습시키며) 찾아내는 과정이 이번 포스팅의 핵심이다.


직선 하나가 정해져있고 x와 y를 정확하게 안다면 기울기 a를 구하는 공식은 y의 증가량/x의 증가량으로 구할 수 있다. 하지만 우리는 직선이 어떻게 그려져야하는지 모르고 데이터(x, y)만 주어졌기 때문에 대략적인 선을 긋고 이를 조금씩 움직이면서 적당한 방정식을 만들어내가는 것이 목표이다.

평균제곱오차라는 개념을 활용할 것이며 이 오차를 줄이는 방향으로 움직일 것이다.

오차(error)는 (실제 y - 예측 y)이고 이를 제곱하여 평균을 내자. 제곱하는 이유는 데이터의 성격에 따라 값이 +- 왔다갔다 할 수 있어서 제곱을 함으로써 값은 모두 양수가 되고 또한 오차가 클 수록 패널티를 부여한다고 생각할 수 있다.

아무튼 그 오차제곱의 합을 평균내는 것을 평균제곱오차(Mean squared error - mse)라고 하며 그 mse는 다음과 같다.

mse = sum((실제 y - 예측 y)^2) / 데이터의 갯수 n 가 된다.
mse = 1/n * sum((y-ax-b)^2)이다.


이제 mse를 줄이는 방향으로 움직여보자.

a와 b를 변경할 것이기 때문에 각각 어느 방향으로 얼마만큼 이동시킬 것인가 하는 문제가 있다. 위의 수식은 a의 입장에서 보면 a에 대한 2차 방정식이다.(b도 마찬가지이다.)

a와 mes간 관계를 그래프로 그려보면 다음과 같다.

다시 처음으로 돌아와서 error를 최소화 하기 위해서는 그림에서처럼 그래프의 접선의 기울기가 0인 부분, 즉 볼록한 아래지점이 a값이면 좋을 것 같다.

그럼 a의 시작점을 임의로 지정했을 때 어떻게 움직이면 될까? 접선의 기울기가 음수이면 + 방향으로, 양수이면 - 방향으로 움직이면 된다.

그럼 얼마만큼 움직여야하나.
예를 돕기위해 그래프의 가로를 시간, 세로를 위치라고 보면 속도는 위치의 변화량 / 시간의 변화량이 된다. 순간 속도는 위치를 시간으로 미분한 것이다. 그럼 해당 그래프의 한 점을 미분하면(시간이 0으로 수렴하면) 해당 지점에서의 순간 속도가 된다. 우리는 그 순간 속도에 learning rate(LR)라는 미리 정의한 상수를 곱한 만큼의 거리만큼 이동해주도록 하자. LR은 사람이 정하는 것이기 때문에 적당한 값으로 learning rate를 정의해야 한다.

mse = 1/n * sum((y-ax-b)^2)를 a에 대해서 편미분하면 -2/n*sum((y-ax-b)*x)이다.
그리고 y-ax-b는 error이기 때문에 치환하면 -2/n*sum(error*x)로 쓸 수 있다.
즉 a_diff = -2/n*sum(error*x)이다.

기울기가 음수이면 + 방향으로, 양수이면 - 방향으로 움직이기 위해서 편미분한 결과(a_diff)를 LR만큼 곱해서 빼주면 마침내 a = a - LR*a_diff가 완성되고 이를 통해 a를 업데이트를 해주면 된다.
이를 계속해서 반복해가면서 a를 계속 업데이트해준다면 결국 a가 적당한 지점에 도달할 수 있을 것이다. b역시 똑같이 편미분을 해서 b_diff를 구하자.

b_diff = -2/n*sum((y-ax-b)) = -2/n*sum(error)

정리하면 a와 b는 다음처럼 업데이트를 할 수 있다.
a = a - LR*a_diff, b = b-LR*b_diff 두 식을 얻을 수 있다.

그럼 코드로 보도록 하자.

import numpy as np

import pandas as pd

data_set = [[1,80],[2,85],[4,89],[8,93]]

x = [i[0] for i in data_set]

y = [i[1] for i in data_set]

origin_x = np.array(x)

origin_y = np.array(y)

a=1 #init a

b=50 #init b

LR = 0.005 #learning rate

for i in range(0,2001):

  predict_y = a * origin_x + b #predicted y

  err = origin_y - predict_y #error

  a_diff = -1*(2/len(x))*sum(origin_x*err)

  b_diff = -1*(2/len(x))*sum(err)

  

  a = a-LR*a_diff

  b = b-LR*b_diff

  if( i % 100 == 0):

    print("i=%.f, a=%.04f, b=%.04f" %(i, a, b))

결과

대략 2000번 시도하니 a는 1.7, b는 80로 수렴하고 있다. 이로써 y=ax+b 수식을 구할 수 있었고 이를 통해 선형회귀(linear regression)를 간단하게 구현해보았다.
마찬가지로 y=ax1+bx2+c의 경우처럼 a, b, c를 구하는 방법도 똑같이 구하면 된다.

하둡 - 오류가 발생한 블록 처리하기 (Status : Corrupt)

간혹 장비 노후라던지의 문제로 인해 하둡 블록이 문제가 되는 경우가 있다.
블록에 오류가 발생할 경우 Corrupt라는 메시지가 뜨게 되는데 데이터의 원본이 존재하거나 다른 방법으로 복구가 가능하다면 문제가 되지 않는다.

하지만 어떠한 이유로 든 간에 replication 1로 생성된 파일이 Corrupt가 발생한다면 해당 파일은 더이상 사용이 불가능하기 때문에 지워줘야한다. (과거 Polybase로 HDFS에 내리는 경우 replication 1로 떨궈졌다.)

다음 명령어를 통해 하둡 파일 시스템 상태를 체크한다.
hadoop fsck /

Status : CORRUPT가 확인되고 블록 1개가 문제가 되었다.
(replication이 1이고 테스트 파일이기 때문에 더 이상 필요가 없어서 지우기로 한다.)

hadoop fsck -delete 혹은 -move를 통해 해당 파일을 정리한다.

-move를 하면 /lost+found로 파일이 이동하는데 어차피 필요가 없으니 -delete를 하도록 한다.

hadoop fsck -delete

완료되면 Status가 Healthy로 나와야하는 것 같은데 그게 아닌가.. 뭐 어찌됐든 -delete를 하고 node overview에서 에러가 사라졌고.. 다시 hadoop fsck / 를 날려보았다.

하둡 상태가 정상(HEALTHY)로 돌아왔다.

Tip - Cassandra Stress with ZING (Readynow Option)

이전 포스팅에서 카산드라에 g1gc와 ZING을 테스트 했었는데 사실 정확한 테스트를 하려면 ZING을 제대로 이해하고 테스트를 해야한다.

ZING의 경우에는 compile time이 존재하기 때문에 warm up이 필요하고 일정량 이상의 스트레스를 주어야(시간이 들여야) 퍼포먼스가 어느 정도 궤도에 오르게 된다.

하지만 카산드라 테스트 시 ZING의 신박한 기술인 Readynow Option을 통해 즉시 좋은 퍼포먼스를 확인할 수 있다.

먼저 설명을 하자면 한번 테스트를 하고 log를 떨군다음 이후 테스트에서 해당 log를 읽어서 warm up time을 없애는 것이다! (즉 두번 수행한다.)

출처는 다음과 같다. (azul 공식 문서)

즉 stress test tool을 활용할 때 ZING으로 테스트를 하려면 readynow 옵션을 주어 정확한 테스트가 가능하다.

순서는 cassandra의 stress-stress 스크립트를 수행할 때 -XX:ProfileLogOut을 주면서 output log를 생성할 위치를 알려주고 로그가 잘 떨어지는지 확인한다. 이후 로그가 잘 떨어지는지 확인이 되면 -XX:ProfileLogIn 옵션을 주고 output 로그를 읽도록 스크립트를 변경하면 된다.

테스트 결과 compile time이 없기때문에 즉시 일정수준 이상의 Performence가 나온 것을 확인할 수 있었고 정말 신기한 기능이라는 생각이 들었다.


마지막으로 카산드라를 테스트 할 때에는 메모리와 cpu를 잘 고려해서 테스트를 해야한다. 예를 들어 read performence를 테스트하려고 1m부터 100m까지 늘려가면서 테스트를 할 때 잘 관찰해보면 thread count를 올려가면서 테스트하는 것을 확인할 수 있다. 하지만 서버 사양에 따라 적절한 thread count가 존재하기 때문에 한번은 크게 돌려보면서 어느 수준부터 overload가 발생하는지 확인할 필요가 있다. 필자의 경우 thread가 24개 정도에서 overload가 발생했기때문에 테스트 할 때 일정하게 -rate threads=24 옵션을 주고 테스트를 진행하였다.

번외로 zing trial version으로 테스트를 해본 결과 성능이 좋았고 특히 GC 수행시간이 아주 일정하게 5ms이하로 유지하는 것을 보고 놀라웠다. 참고로 G1GC의 경우 약 10~15ms 왔다갔다 했다.