2020 Data Conference Speaker로 참여한 후기 (주제 : Spark+Cassandra 기반 Big Data를 활용한 추천 시스템 서빙 파이프라인 최적화)

연초에 항상 1~3년간의 장기 계획, 1달씩 단기 계획을 세우면서 어떤 부분을 새로 혹은 보강하기 위해 레벨업 할지 목표를 정하는데 계획에 없던 컨퍼런스에 연사로 참여하게 되었습니다. 컴퓨터월드/IT Daily에서 주최하고 양재 엘타워에서 진행하기 때문에 큰 컨퍼런스라서 부담이 되었는데 함께 일하는 빅데이터 파트분들과 Azul Systems에서 많이 도와주셨습니다.

(링크:http://www.itdaily.kr/conference/event7.html)

데이터 활용 AI&빅데이터, 보안 트랙 부문에서 발표를 진행했고 아무래도 실무를 하는 입장에서 개발적인 부분을 많이 넣고 싶었고, 카산드라를 많이 사용하기를 바라면서 홍보하고 싶고, 그리고 Zing JVM을 적용하면서 성능상 이점을 본 것들을 소개하고 싶었습니다.

주제는 Spark+Cassandra 기반 Big Data를 활용한 추천 시스템 서빙 파이프라인 최적화로 정했고 발표 내용은 어느 정도 카산드라를 사용하는 입장에서는 꼭 고려해야할 부분이라서 어렵지 않은 부분이지만 분산 환경에서 프레임워크들을 사용하다 보면 누구나 겪을 수 있는 상황이라는 측면에서 제가 경험했던 상황으로 풀어보려고 했습니다. 어떻게 트러블 슈팅을 하고 최적화를 했는지, 코드를 짜더라도 Network&Disk I/O 같은 인프라 환경을 고려해야한다는 것을 주로 설명하고자 했습니다.

그리고 국내에선 Cassandra+Zing 조합으로 레퍼런스가 없어서 이 부분이 가지는 이점을 최대한 설명하고자 했는데 국내에서는 일단 Cassandra를 사용하는 분들이 늘어나기를 기대합니다.

(카산드라 한국 사용자모임 그룹 홍보 : https://www.facebook.com/groups/cassandra.kr)

발표자료 : https://www.slideshare.net/SuseongPark/sparkcassandra-data-pipeline-optimazation-at-recommend-system-for-recommend-system?fbclid=IwAR2-jMhdh-Tf_WbBMFOGOG7h1WbFHTctbVaGqH44CupmbQjngMUj0_ppdzA

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에 정리해서 올려놓도록 하겠다.

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가 잘 잡힘을 확인할 수 있었다.

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를 활용하기 때문에 성능상의 이점이 있는데 이를 포기하는 쿼리가 되는 것이다.

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 왔다갔다 했다.

Cassandra + Prometheus + Grafana Monitoring System unsing by jmx exporter(node exporter)

카산드라(cassandra) 모니터링을 하기 위한 시스템을 마련하기 위한 포스팅이다. InfluxDB를 사용하는 케이스가 많이 있지만 이번 포스팅에서는 InfluxDB 없이 모니터링 시스템을 구축해본다.

큰 틀은 cassandra의 jmx를 활성시키고 해당 로그를 Prometheus에서 수집해서 Grafana로 화면에 그려주는 방식이다.

사용 버전
- cassandra 3.11.4
- jmx_prometheus_javaagent-0.3.0.jar
- prometheus 2.14.0
- grafana 5.0.1

그럼 카산드라에서 발생시키는 로그를 수집서버에서 수집해보자.
그 로그들은 각 클라이언트에서 수집서버로 Push(Polling)하는 방식과 수집서버에서 카산드라 로그를 Pulling하는 방식 두 가지로 나뉠 수가 있다. 

일반적인 모니터링 툴들은 각 클라이언트에서 수집서버로 로그를 전송하는 형태를 띄고 있다면 프로메테우스는 각 로그들을 각 클라이언트에서 exporter를 통해 가져오는 방식을 차용하고 있어서 프로메테우스가 장애가 나더라도 서비스 어플리케이션에 문제가 되지 않는다. 왜냐하면 각 클라이언트 들은 메트릭 정보를 수집해놓고 수집해가기를 기다리기만 하기 때문에 프로메테우스의 장애와는 아무런 상관이 없다.

아래는 프로메테우스의 아키텍쳐이다.

간단하게 살펴보면 프로메테우스 서버가 있고 여기서는 PushgaeWay나 Jobs/exporters에서 메트릭정보를 Pulling 하게 된다. 즉 어플리케이션단에서 exporter를 띄워놓거나 Pushgateway로 메트릭 정보를 보내면 프로메테우스 서버에서는 저 메트릭 정보를 가져오는 형태이다.
이 포스팅에서는 node exporter(서버 로그)와 jmx exporter(카산드라 로그)를 사용할 것이다.

그리고 Prometheus server에서 설정해놓은 정보에 따라 AlterManager에 시그널을 주면 이메일같은 채널로 알림기능을 설정할 수 있다.

마지막으로 Prometheus web UI나, Grafana와 같은 Visualization 툴을 통해 프로메테우스에서 수집한 메트릭 정보를 그려줄 수 있다. 이를 위해 PromQL이라는 쿼리를 사용하게 되는데 그라파나에서 제공하는 대시보드를 보면 sum이나 count가 대부분이라서 크게 어렵지 않는 것 같다. 그러니 누가 카산드라 메트릭 정보를 잘 표현할 수 있는 쿼리를 아주 잘 만들어주면 기쁜마음으로 사용하겠다.

카산드라를 설치하고 jmx까지 활성화가 되있다고 가정하자.
jmx 활성화 하는 방법은 어렵지 않다.
필자의 경우 중간에 제대로 동작을 안했었는데 이유는 conf/cassandra-env.sh 파일에서 LOCAL_JMX가 yes로 세팅되어 있었기 때문이다.
따라서 다음과 같이 수정하였다.

if [ "x$LOCAL_JMX" = "x" ]; then
    LOCAL_JMX=no
fi

그럼 이 데이터를 Prometheus server가 가져갈 수 있도록 jmx exporter와 node exporter를 설치하도록 한다.

이 포스팅에서는 두 가지 모두 설치해보고 비교해보자.

먼저 node exporter 세팅이다.

프로메테우스 공식 홈페이지에서 tarball을 받아서 각 노드에 옮겨서 설치한다.

주소는 다음과 같다.
https://prometheus.io/docs/guides/node-exporter/

node exporter는 단순히 압축을 풀어서 실행하기만 하면 9100 포트로 수집된다.
따라서 prometheus server에서 각 노드 9100 port로 들어가서 수집해오면 된다.

UI로 들어가서 확인해보자.

다음은 prometheus server에서 node exporter를 수집해가면 된다.
prometheus.yml 파일에 수집타겟을 설정해주자.

프로메테우스 UI에 가서 targets를 확인해보면 잘 떠있는 것을 확인할 수 있고 실제로 수집 되는 데이터를 그래프로 확인도 가능하다.

테스트로 그래프를 그려본 결과 그려지긴 하지만 뭔가 다이나믹하지 않고 상당히 아쉬운 느낌이 든다.

따라서 위의 프로메테우스 결과를 다시 grafana로 가져와서 grafana를 통해 그려보도록 한다.


이 역시 공식 사이트에 들어가서 다운을 받아서 설치하도록 하자.
주소는 다음과 같다. https://grafana.com/grafana/download


이 역시 압축을 풀고 서버를 띄우기만 하면되서 간단하다.
서버가 올라오면 3000번 포트로 접속해보자.


누군가 만들어서 배포해놓은 대시보드를 사용하자.
https://grafana.com/grafana/dashboards/


아래는 대시보드 11074번을 import 하였다.

그 결과 다음과 같은 아름다운 화면이 완성되었다.

하지만 뭔가 아쉬운 점이 있다면 node exporter는 하드웨어 모니터링 느낌이 강하다.
우리는 카산드라 노드의 정보가 알고싶기 때문이다.

이를 위해서 jmx를 써야한다.
같은 방식으로 jmx exporter 세팅 후 프로메테우스+그라파나로 연동하자.

cassandra metric을 보기 위한 jmx exporter 세팅이다.

다운로드 링크

https://repo1.maven.org/maven2/io/prometheus/jmx/jmx_prometheus_javaagent/0.3.0/

카산드라는 jmx 세팅을 위해 카산드라를 노드별로 한대씩 내렸다가 올려야하기 때문에 조금 귀찮을 수 있어서 한번에 성공하도록 한다.

jmx exporter가 사용할 cassandra.yml 파일은 아래 주소를 참고 했다.
혹시 바뀔수도 있으니 현재 시점 샘플을 포스팅 마지막에 구글 드라이브 링크로 첨부하였다.
https://raw.githubusercontent.com/prometheus/jmx_exporter/master/example_configs/cassandra.yml


저 두 파일을 카산드라 폴더에 넣고 conf/cassandra-env.sh에 JVM_OPTS에서 물고 올라갈 수 있도록 다음을 추가하자.

JVM_OPTS="$JVM_OPTS -javaagent:/sw/cassandra/jmx_prometheus_javaagent-0.3.0.jar=7070:/sw/cassandra/cassandra.yml"


또는 다음처럼 명령어로 추가를 해주자. 모든 수집대상 노드에서 똑같이 반복한다.
echo 'JVM_OPTS="$JVM_OPTS -javaagent:'$PWD/jmx_prometheus_javaagent-0.12.0.jar=7070:$PWD/cassandra.yml'"' >> conf/cassandra-env.sh

위처럼 conf/cassandra-env.sh 마지막 줄에 jmx 에이전트 세팅을 해주고 카산드라를 재기동 한다.

역시 프로메테우스에서도 target ip를 추가해주자.
(이후 dashboard를 하나도 수정하지 않기 위해 job_name을 cassandra_로 시작하도록 하였다. instance명을 반드시 cassandra_로 맞춰주도록 하자.)

위에서 JVM_OPTS 를 설정하면서 7070 포트로 설정했으니 이번에는 7070 포트로 수집을 하면 된다.

프로메테우스 target에서도 up 상태를 확인할 수 있다.

node exporter와는 다르게 cassandra metric이 수집되고 있다.

마지막으로 grafana에서 적당한 대시보드를 골라서 그려보도록 하자.

dashboard number 5408을 활용했다. (https://grafana.com/grafana/dashboards/5408)

datasource가 없다면 생성해주고 5408 json을 import하면 다음과 같은 화면이 그려진다.

이상으로 cassandra + jmx exporter(node exporter) + prometheus + grafana를 활용한 cassandra monitoring system 구축을 마친다.

관련 파일은 아래 구글드라이브에서 첨부하였다.
https://drive.google.com/open?id=16UtW5A175w1tVknXvHFTOQUU9W2d9bNS

내용물
1. cassandra.yml
2. cassandra_rev3.json
3. node_exporter-0.18.1.linux-amd64.tar.gz
4. jmx_prometheus_javaagent-0.3.0.jar