MAB(Multi Armed Bandit) 대안으로서의 Dynamic Ensemble

MAB(Multi Armed Bandit)

A/B test를 보다 더 유연하게 하면서 최적의 return을 찾아가는 방법이다.

A/B test의 경우 A/B 셋을 준비하고 타겟에 따라 A 혹은 B를 리턴을 주면 되기 때문에 구현하기 쉽기는 하지만 분명히 only A,B라는 한계점이 존재한다.

MAB는 이를 조금 더 확장한 개념으로 n개의 셋을 준비해서 보상에 따라 적절한 셋을 return 하는 것이 concept이다.

가령 상품추천을 할 때에 A모델, B모델, C모델, D모델, 혹은 앙상블된 모델을 미리 준비하고 CTR 등의 보상에 따라 적절한 모델을 return 하도록 구현할 수 있다.

모델의 갯수가 충분하고, 모델을 집계하는데 걸리는 시간이 충분하다면 MAB는 좋은 선택지가 될 것이다.

하지만 불가능한 상황이 분명히 존재한다!!

DE(Dynamic Ensemble)

단순히 A,B 두개의 상품추천 모델만이 존재할 수 있다. 이 때는 A와 B를 적절한 비율로 섞어서 사용해야 하는 경우를 가정해볼 수 있다.

과연 A+B 앙상블된 모델 1개만이 존재할 때 MAB를 어떻게 할 수 있을까?

분명히 MAB는 아니다. 하지만 A와 B를 섞는 비율 n:m을 보상을 통해 동적으로 조절한다면 충분히 비슷한 효과를 낼 수 있을 것이다.

물론 말단에서 어떤 모델을 사용할 것인지 고르는 것이 MAB라면 이는 아예 모델을 만들 때 보상을 사용하고 1개의 모델만 생성되기 때문에 MAB라고 부르기엔 무리가 있다.

하지만 적절한 결과를 return 해야하고, 보상에 사용할 수 있는 방법을 똑같이 사용할 수 있기에 비슷하기도 하다!

그래도 용어를 다르게 한다면 아마 적절한 용어로는 Dynamic Ensemble이 될 것이다.

MAB를 하려다가 상황이 안맞을 것 같아서 DE방식으로 방향을 틀어야할 수도 있는데 개인적인 생각이지만 적절한 대안같다.

kafka + logstash saving to hdfs (webhdfs)

상품추천 분석을 위한 로그들을 kafka broker에 꾸준히 쏴주고 있었다. 이제 이를 사용할 때가 되었다.

이 로그들을 하둡에 쌓을 필요가 생겼고 컨슈머를 구현할지, logstash의 webhdfs를 사용할지 고민하다가 logstash의 webhdfs output이 간단해서 이걸 사용하기로 하였다.

필요한 플러그인들이 설치되어있다고 가정하고 conf파일을 작성한다.

크게 input, filter, output 부분을 작성하면 된다. 필요한 부분을 검색해서 쓰긴했는데 나중에 시간날 때 제대로 사용법을 익혀두는게 좋을 것 같다. (꽤 편리한 것 같다.)

input {

     kafka {

            bootstrap_servers => "a:9092,b:9092,c:9092"

            topics => ["topic1","topic2"]

            decorate_events => true

            consumer_threads => 3

            group_id => "dev_recommend"

            }

}

모든 토픽의 파티션은 3개로 되어있기때문에 consumer_threads를 3으로 했다.

데이터를 가공하는 filter부분에서 토픽접근은 [@metadata][kafka][topic]로 하면되고 or조건은 or로 쓰면 되었다. c1은 unix time으로 된 컬럼이고 컬럼끼리의 구분자는 \t으로 되어있다면 아래처럼 작성해주면 되었다.

filter {

    if [@metadata][kafka][topic] == "topic1" or [@metadata][kafka][topic] == "topic2" {                                                                            

        dissect {

            mapping => {"message" => "%{col1} %{col2}   %{col3}   %{col4}   %{col5}   %{col6}   %{col7}   %{col8}   %{col9}   %{col10}  %{col11}"}

        }

        ruby {

                code => "

                      require 'time'

                      require 'date'

                      ts_new = event.get('[ts]').to_i/1000

                      event.set('day_ymd', Time.at(ts_new).localtime.strftime('%y-%m-%d'))

                      event.set('day_h', Time.at(ts_new).localtime.strftime('%H'))

                      event.set('day_ymdh', Time.at(ts_new).localtime.strftime('%y-%m-%d-%H'))

                  "

            }

    }

}

마지막으로 ouput 부분은 다음처럼 작성하면 된다.

output {

    if [@metadata][kafka][topic] == "topic1" {

        webhdfs {

            host => "hadoop url"

            port => 50070

            path => "/data/tsv/recommend-logs/%{day_ymd}/%{day_h}/topic1-log-%{day_ymdh}.log"

            user => "usr"

            codec => line { format => "%{message}"}

            }

        #stdout { codec => rubydebug }

    }

    else if [@metadata][kafka][topic] == "topic2" {

        webhdfs {

            host => "hadoop url"

            port => 50070

            path => "/data/tsv/recommend-logs/%{day_ymd}/%{day_h}/topic2-log-%{day_ymdh}.log"

            user => "usr"

            codec => line { format => "%{message}"}

            }

        #stdout { codec => rubydebug }

    }

}

이제 실행해서 화면에 찍어보면 다음과 같은 형식으로 로그가 찍힐것이다.

output file은 계속해서 append를 해야하기때문에 text 포맷이다. 이후 실행하면 yy-mm-dd/시간/파일 경로에 데이터가 잘 쌓이는 것을 확인할 수 있다. 

나중에 제대로 사용법(문법)을 익혀둬야겠다.

일단 작업을 돌려놓고 며칠간 사이즈 모니터링을 해봐야겠다.

nohup bin/logstash --path.data recom_data -f recom.conf > /dev/null 2>&1 &

Kafka + Spark Streaming saving to Cassandra

추천 시스템 서빙에서 메인 DB는 Cassandra를 사용하고 있는데 대부분이 data를 bulk insert 후 select만 해서 return 하는 구조이다. 하지만 최근 실시간 (방문)로그를 활용해야할 필요가 생겼고 이를 Cassandra에 적재하기로 하였다. 

만약 cassandra에 쌓는다면 real-time 환경에서 transaction에 문제가 없는지, 하루 쌓이는 양은 얼마 정도인지, compaction 전략은 어떤걸 써야할지, 피크 상황에서 서버가 버틸지 등의 고민을 하게 되었다.

이러한 시도가 성공할지, 실패할지는 먼저 구현을 해봐야 알 수 있기 때문에 일단 먼저 구현부터 하기로 했다. 만약 데이터양이 많은 것이 문제가 된다면 사이즈를 줄이는 방향으로 타협을 보는 것으로 문제를 해결해도 되는 상황이다.

구현 방법은 두 가지로 생각했다.

1. kafka -> logstash -> cassandra

2. kafka -> spark streaming -> cassandra

일단 1번은 logstash plugin이 존재하기는 하는데 4~5년전 자료라서 pass.

2번의 경우 어느 정도의 성능이 나올지 몰라서 궁금하기도 했고 2번으로 결정했다.

정리하면 real time으로 Kafka broker -> spark streaming(consum) -> cassandra 로 데이터를 흘러가는 것이 목표이다.

기존에 운영중인 kafka 데이터를 사용하기로 했다. kafka의 데이터는 map형식으로 들어오고 있다. 

{"a_key":"a_value","b_key":"b_value","c_key":"c_value","d_key":"d_value"}를 담을 class를 하나 만들었다.

case class TEST ( COL1:String, COL2:String, COL3:String, COL4:String )

이후 카프카와 연동을 해줬다.

  val sparkConf = new SparkConf()
    .setAppName("kapark2cassandra")
    .set("spark.cassandra.auth.username", "cassandra")
    .set("spark.cassandra.auth.password", "cassandra")
    .set("spark.cassandra.connection.host", "cassandra cluster")

  val ssc = new StreamingContext(sparkConf, Seconds(3))

  val kafkaParams = Map[String, Object](
    "bootstrap.servers" -> "zk cluster",
    "key.deserializer" -> classOf[StringDeserializer],
    "value.deserializer" -> classOf[StringDeserializer],
    "group.id" -> "kapark2cassandra",
    "auto.offset.reset" -> "earliest",
    "enable.auto.commit" -> (false: java.lang.Boolean)
  )
  val topics = Array("kapark2cassandra")
  val stream = KafkaUtils.createDirectStream[String, String](
    ssc,
    PreferConsistent,
    Subscribe[String, String](topics, kafkaParams)
  )

  stream.foreachRDD { rdd =>
    val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
    stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
  }

다음은 데이터 가공 및 카산드라에 전송이다.

ttl은 1주일(604800)으로 셋팅했다. 

stream.foreachRDD { rdd =>
    val spark = SparkSessionSingleton.getInstance(rdd.sparkContext.getConf)
    import spark.implicits._
    val ods = rdd
      .map(record => record.value.toString.drop(1).dropRight(1).split(",(?=([^\"]*\"[^\"]*\")*[^\"]*$)", -1)
      .map(_.split(":(?=([^\\\"]*\\\"[^\\\"]*\\\")*[^\\\"]*$)")).map(arr => (arr(0) -> arr(1))).toMap)
      .map(rec =>
          TEST(
            rec.get("\"COL1\"").mkString.replace("\"", "").trim
            , rec.get("\"COL2\"").mkString.replace("\"", "").trim
            , rec.get("\"COL3\"").mkString.replace("\"", "").trim
            , rec.get("\"COL4\"").mkString.replace("\"", "").trim
          )
      ).toDF()

    ods.createOrReplaceTempView("kapark2cassandra")

    var result = spark.sql(
      """
          select  from_unixtime(col1/1000,"yyyy-MM-dd HH:mm:ss") as col1
                 , col2
                 , col3
                 , col4
          from kapark2cassandra
          where col2 = some
          """)
    result.write.format("org.apache.spark.sql.cassandra")
      .options(Map("table" -> "test" , "keyspace" -> "real_time", "ttl" -> "604800"))
      .mode(org.apache.spark.sql.SaveMode.Append).save()
}

ssc.start()
ssc.awaitTermination()

저기서 다른 데이터 소스와 join, group by해서 사용할 수 있으면 그렇게 하려고 하는데 스트리밍이 밀리지 않는 선에서 처리가 가능한 양이어야 할 것이다.

object SparkSessionSingleton {
  @transient  private var instance: SparkSession = _
  def getInstance(sparkConf: SparkConf): SparkSession = {
    if (instance == null) {
      instance = SparkSession
        .builder
        .config(sparkConf)
        .getOrCreate()
    }
    instance
  }
}

마지막으로 val spark = SparkSessionSingleton.getInstance(rdd.sparkContext.getConf) 부분을 만들기 위한 ojbect 생성해주고 

def main(args: Array[String]) {
  new kapark2cassandra().run(args)
}

수행했더니 데이터가 카산드라에 잘 들어간다.

통계를 보도록 하자.

대충 1큐에 5천건~8천건 정도이다.

3초로 했을 때 조금씩 튀는 부분이 있는데 아무래도 3초는 무리인것같다.

cassandra 지표를 봐도 딱히 튀는 부분도 없고 결국 final result는 약 100 rows씩 들어가기 때문에 부하 테스트를 하기에는 작은 양이었다.

결론

real time data를 cassandra에 넣으려면 kafka + spark streaming으로 처리하는 것도 하나의 옵션이다.

Cassandra can not partition switch

Oracle, MSSQL 등 RDBMS에서는 PARTITION SWITCH를 통해 순간적인 데이터 바꿔끼우기가 가능하다. 이는 배치를 개발할 때 매우 유용하게 사용할 수 있다. 

현재 유효 파티션을 저장한 테이블을 두고 이 테이블 조회를 한번 거치고 본 테이블을 조회하는 방식으로 운영할 수 있다. 가령 PARTITION이 10, 20으로 나뉘어져 있다면 현재 유효 파티션 번호는 10을 기록하고 있는다. 이후 다음 배치에서는 파티션 번호 20으로 데이터를 적재한 후 PARTITION SWITCH를 하고 유효 파티션을 20으로 UPDATE하는 방식이다. 

(ex: ALTER TABLE… EXCHANGE PARTITION)

하지만 카산드라에서는 이러한 기능이 없다. 따라서 동일한 스키마 테이블 2벌을 생성하고 핑퐁하는 방식으로 구현해야한다. (모든 테이블이 2벌씩 나오므로 비효율적일 수 있는데 현재로서는 사용하기 쉽기도 하고 대체할 방법이 딱히 없어보인다.)

먼저 본 테이블을 조회하기전에 조회해야할 유효테이블마스터 테이블이다.

그리고 본 테이블인 table_A, table_B, table_C을 각각 두벌씩(_10, _20) 생성하자.

table_A_10, table_A_20, table_B_10, table_B_20, table_C_10, table_C_20

table_A 테이블을 조회하기 위해 eff_tbl_mst 테이블을 조회해서 유효코드값 10을 알아내고 이후 table_A_10을 조회하는 방식이다. 이렇게 하면 데이터 적재 배치후에 eff_tbl_mst 테이블에서 div 컬럼 값을 10<->20으로 update해주면 된다. 이후 유효하지 않은 테이블은 truncate.

현재 많이 사용하는 방법이고 쉬우면서도 운영 중에 데이터 전체를 갈아끼우기 편해서 좋다.

하지만 단점도 존재한다. 매번 동일한 테이블이 생성되고 db입장에서는 table을 조회하기 위해서 2번의 쿼리를 수행해야한다는 단점이 있다.

Cassandra SSTable에 의한 Disk I/O 영향

최근 Spark에서 Cassandra에 데이터를 적재할 때 SSTable Write 형식으로 전환했다.

변환 후 Spark excutor에서 카산드라 노드에 데이터를 전송할 때 네트워크 트래픽 문제가 발생하여 이를 대역폭의 35%만 사용하도록 동적 세팅을 해놓았었다.

참조 : Spark to Cassandra SSTable로 bulk insert할 때 throughput 조절하기

이후 대역폭을 1G에서 10G로 인프라를 업그레이드했지만 이번엔 다른 문제가 발생했다.

문제

전송하고자 하는 데이터 size는 50G, SSTable 갯수는 약 3천개이다.

(실시간으로 read, write가 이루어지는)운영중인 카산드라에 SSTable을 복사하고 해당 테이블을 바라보게 하였다.

insert 직후라서 key cache가 없고 sstable은 약 3천개이며 동시에 compaction도 진행이 되었다.

결과적으로 트래픽이 몰리면서 read request가 꾸준히 올라가면서 disk I/O에 부하를 주는 문제가 되었다. 장애가 난 것이다.

당시의 그래프이다.

sstable이 점차 줄어들고 있는 것은 compaction 때문이다.

다음 그래프를 보면 약 19시부터 disk i/o가 full 찬 것을 확인할 수 있으며 22시에 1/3토막이 난 것은 was단에서 read request를 강제적으로 1/3으로 줄였기 때문이다.

그랬더니 disk i/o가 정상적으로 돌아왔으며 두개의 그래프로 추정해보건데 compaction보다도 read에서 더 큰 부하를 많이 준 것으로 판단된다. (물론 read와 compaction 모두 경합이 일어나서 복합적인 문제가 발생한 것이기도 하다.)

정리하면 2가지의 원인은 다음과 같다.

1. 트래픽이 점차 몰리면서 카산드라 read 발생수가 많아졌고 이때 데이터를 조회하기 위해 대량의 sstable을 읽어야하기 때문에 disk i/o 부하를 줌.

2. 데이터 적재 후 약 대량의 sstable이 생성되었고 카산드라 자체적으로 compaction이 돌면서 disk i/o에 부하를 줌.

문제를 해결하기 위해 SSTable 갯수를 줄이는 것을 먼저 시도하였고 withBufferSizeInMB로 조절이 가능했다. 하지만 이 옵션은 oom이 발생할 수 있어 꾸준히 모니터링을 해야하지만 default 128에서 현재 256으로 올린 상태에서는 안정적인 disk io를 유지했다.

점차 테스트해보면서 이를 더 올릴 생각이다.

추후에 더 문제가 발생하면 table을 나누던지, compaction strategy를 바꿔봐야하겠다. 시계열 데이터가 아니지만 LCS가 도움이 된다는 글이 있기에 시도해봐야겠다.

아무튼 정리하면

운영중인 카산드라에 bulk load를 할 때 disk I/O도 고려해야한다는 것이다.

Spark to Cassandra SSTable로 bulk insert할 때 throughput 조절하기

최근 Spark에서 Cassandra로 데이터를 부을 때 SSTable을 활용해서 Bulk Insert를 하였다.

분산환경에서 여러 테스트를 거치며 노하우를 공유차 포스팅을 남긴다.

Spark executor들에서 병렬로 Cassandra에 부을 때 주의할 점은 초당 처리율이다.

아무래도 실 운영환경에서 Bulk를 쓴다는 것은 데이터가 10~100GB이상의 데이터일 것이고 이 때 주의할 점은 Network Traffic이다.

카산드라 서버 네트워크 카드의 limit을 고려하며, 즉 인프라 환경을 반드시 고려하면서 분산처리를 해야한다.

필자의 경우에는 각 카산드라 노드가 초당 1Gbps를 받을 수 있고 나름대로 base line을 max 35%인 350Mbps만 사용하기로 정했다. 동시에 2개의 배치를 돌리더라도 70%사용이라면, 유사시에 충분히 대비할 수 있기 때문이다. (30%인 300Mbps를 여분으로 남김.)

따라서 목표는 배치당 350Mbps을 limit으로 잡았고 이때 변수는 파티션갯수와 각 파티션에서의 전송속도이다. 변수를 1개로 줄이기 위해 file size가 1G당 파티션은 1개로 고정하였다. file size가 커질 수록 partition이 너무 많이 생길까봐 개인적으로 1~2를 추천한다.

즉 10G파일을 insert할 때에는 파티션이 10개가 되어 10개가 병렬로 insert가 된다.

(파티션갯수는 repartition으로 조절하며 전송속도는 mapreduce.output.bulkoutputformat.streamthrottlembits 옵션으로 정할 수 있다.)

다음은 예시 코드이다.

val fs: FileSystem = FileSystem.get(sc.hadoopConfiguration)
val file_size = fs.getContentSummary(new Path(outputFilePath)).getLength
var partition_num = (file_size/1024/1024/1024) //1기가당 파티션2개
if ( partition_num == 0 ) { partition_num=1 }
var streamthrottlembits = (350 / partition_num) + ""
if ( streamthrottlembits == "0" ) { streamthrottlembits="1" }

file size로 partition 갯수를 정하고 streamthrottlembits를 정했다. 나뉜 파티션으로 foreachPartition을 돌릴 것이다.

try {
 val conf = new Configuration()
 val cassHosts = "ip1,ip2,ip3,ip4,ip5"
 val shuffledCassHosts=Random.shuffle(cassHosts.split(",").toList)
 val selectedCassHost=shuffledCassHosts.head
 conf.set("cassandra.output.thrift.address", selectedCassHost)
 conf.set("mapreduce.output.bulkoutputformat.streamthrottlembits", streamthrottlembits)
 new SSTableLoader(new File(dir), new ExternalClient(conf), new OutputHandler.LogOutput).stream().get()
} finally {
 var deleteFolderList = dir_file.listFiles();
 for ( i <- 0 to deleteFolderList.length -1 ) {
  deleteFolderList(i).delete();
 }
 dir_file.delete();
}

이후 sstable을 만들고 전송을 할 때 위에서 구한 streamthrottlembits값으로 속도를 조절하였다.

또한 코드상으로는 카산드라 클러스터 ip들 중에서 랜덤으로 1개를 골라서 insert를 하였는데  이는 인프라마다 다르겠지만 어쨌든 한쪽 노드로 traffic이 몰리지 않게하기 위함이다.

사실 더 좋은 구성은 카산드라를 1개의 스위치에 모두 물리지 않고 2개 이상의 스위치로 분리해서 구성하였다면 더욱 안전한 아키텍처가 될 것이다.

network traffic이 원하는데로 max 35%만 사용하고 있다.

cpu 상태도 안정적이다.

Spark to Cassandra - memtable에서 sstable write로 전환기 (cassandra bulk load)

스파크를 활용해서 카산드라에 매일 100억이상의 row, 250G 이상의 데이터를 붓고 있었다.

붓는 방식은 memory write, 즉 memtable에 insert를 하는 방식으로 사용하고 있었다. 하지만 이러한 방법은 문제가 되었다.

상황

초기에는 데이터 양이 많지 않았고, 순간적인 부하만 견디면 됐기 때문에 이 정도면 새벽시간대에 충분히 커버가 가능하다고 생각했다. 하지만 데이터셋이 많아지고 갈수록 커지는 데이터 사이즈에 따라 새벽시간대에 cpu가 100%까지 치는 일이 많아졌다.

이는 카산드라 입장에서 read, write에 영향을 주고 결국 pending threads와 db connection timeout 수치를 기하급수적으로 늘어 장애가 발생할 위험이 존재했다.

물론 클러스터 구조에 따른 어느 정도 내구성은 보장했지만 노드가 순차적으로 돌아가며 부하를 받는 것을 보고있자니 걱정이 되었다.

Spark에서 cassandra로 데이터를 넣을 때 throuput을 조정하긴 했지만 이는 근본적인 해결책이 되지 못했다.

Before Graph

write / sec

cpu usage

network traffic

특정 시점에 conection timeout & pending threads

개선 계획

memtable에 direct로 붓지 말고 sstable에 데이터를 붓는 방식으로 변경한다. 이는 disk에 쓰는 방식이다. spark에서 병렬로 write를 한다면 충분한 성능이 나올 것이다. disk I/O와 network 성능에 맞춰서 적당한 throuput을 찾는다.

여러 깃소스, 블로그 레퍼런스를 통해 대부분이 spark to cassandra에서 비슷한 문제를 겪었고 대부분이 동일한 방법으로 해결한 것 같다. 소스를 참고해서 우리에 맞는 방식으로 튜닝을 시도했다.

github Arunkumar&joswlv 등 소스를 참고해서 적용 전에 계속해서 테스트를 했다.

다음은 예제 소스이다. (라이브러리화 한 소스는 github에 올려두었다.)

현재 mesos로 resource를 관리하고 있으며 데이터 size에 따라 partition 갯수와 excutor 갯수를 조절해가며 적절한 구간을 찾아야한다. 참고 자료에서는 hdfs 블록 사이즈인 256M를 활용하는 것 같은데 아마 카산드라의 옵션 중에 sstable 관련해서 변경하지 않는다면 비슷하게 적용할 것 같다.

sstable을 작성할 directory를 랜덤하게 만들고, create와 insert문을 준비, 각각의 partition(excutor)에서 sstable을 parallel create&load, 마지막으로 작업했던 directory를 지우는 방식이다.

"mapreduce.output.bulkoutputformat.streamthrottlembits"를 1000으로 했는데 이는 적용하는 인프라에서의 네트워크 상태를 고려해서 설정해야한다.

1차적으로 모든 소스를 이렇게 바꾼 후의 모습이다.

After Graph

write / sec

cpu usage

network traffic

특정 시점에 conection timeout & pending threads

결과가 아주 다이나믹하고 아름답게 변했다. 

1. memtable에 direct로 write를 할 때에는 write/sec가 60k 수준으로 올라갔지만 이는 cpu가 튀는 원인이 되었다. 하지만 sstable write를 할 경우에는 cpu가 튀지 않았고 오히려 더 빠르게 insert가 되었다. 

2. cpu 사용량은 compaction에도 영향을 주는데 이는 결국 실제 운영상황에서의 response from cassandra에 영향을 준다.

3. cpu가 안정화되었으니 network traffic을 더 dynamic하게 사용할 수 있다. 현재 포스팅에서는 속도를 더 높여 처리율을 올릴 수 있다.

4. cpu에 따른 connection timeout과 pending threads가 안정적으로 변했다.

5. 카산드라의 disk는 ssd로 되어있는데 이는 sstable write를 사용할 때 성능을 극대화할 수 있었다.

6. 포스팅에는 없지만 단순히 spark to cassandra 배치 중에 memtable에 insert하던 배치가 2시간정도 걸렸었는데 sstable insert로 바꾼 뒤 약 15분으로 단축이 되었다. (2h -> 15m)

주의할점

만약 대량의 데이터를 sstable write를 할 경우에는 네트워크 트래픽을 다 잡아먹을 수 있다. 이는 결국 카산드라의 response에도 영향을 준다. 따라서 사용자가 적은 시간대를 찾거나(새벽), 처리율을 조절하며 적절한 구간을 찾아야한다. 

sstable insert는 수 많은 sstable이 생성되고 이는 쌓이면 쌓일수록 compaction에 부담이 된다. 따라서 적절한 compaction 전략을 구상하여 적용하도록 한다.

partition 갯수에 따른 병렬처리에 따라 성능을 좌우한다. 적절한 partition 갯수를 조절하며 병령처리 성능을 올려야한다. 이는 테스트 중이고 추후 성능테스트 포스팅을 할 예정이다.

distribution 환경에서의 insert 성능 테스트 포스팅을 할 것이며, sstable을 통한 bulk insert 소스는 github에 정리해서 올려놓도록 하겠다.