spark application resource manager - MESOS vs YARN

MESOS

메소스의 경우 마스터가 자원을 중계하는데 특이한 점은 mesos agent(slave)에 mesos excotur가 뜨고(자원을 먼저 점유) 그 안에서 다시 spark-excutor가 실행된다. 

즉 순서가 spark-submit을 할 때 클러스터 매니저를 mesos master로 지정을 하면 메소스 프레임워크가 실행되고 그 안에서 spark driver(spark context)가 실행되면서 mesos 마스터로부터 자원을 다시 할당받고 mesos agent(slave)에 다시 spark excutor를 실행하는 방식이다. 

만약에 deploy mode를 client mode로 하면(default) 실행한 서버(배치서버가 되겠지)에서 mesos framework가 수행되고 cluster mode로 하면 mesos agent 중에서 mesos framework가 수행된다.

메소스의 경우 fine-grained와 coarse-grained가 있는데 fine-grained는 각 태스크마다 spark excutor를 한개씩 만들어서 사용하고 후자는 통으로 점유한 후 공용으로 사용하는 방식인데 fine-grained는 spark 2.4.x 버전 문서를 보다보니 deprecated 되었더라! 이유는 fine-grained의 각 task를 시작할 때 overhead가 크고 spark 작업이 종료될 때 core는 반환하지만 memory를 계속 점유하고 있다고 한다. 결과적으로 default가 coarse-grained으로 사용하고 core, memory 옵션을 주고 적절히 자원을 분배(동적분배 포함)하는 것이 배치 입장에서 이득이다.

deploy mode는 default가 client mode인데 이는 해당 process를 kill했을 때 각 excutor의 작업들이 모두 kill이 되기 때문에 추천하고 만약 cluster mode로 하면 debug를 할 때 문제가 된다. 어차피 collect같은 작업이 사용될 때 넉넉한 배치서버의 자원을 활용하기 위해서라도 master와 배치서버를 동일 시 하는게 좋다. 또한 cluster mode로 할 때에는 별도로 mesos-displatcher를 띄우고 거기에 master를 지정해야한다. 어차피 권한이나 환경등 세팅을 master가 기준이라면 client를 사용하는게 맞고 추후에 agent를 추가할 때 이렇게 하는 편이 더 편하다. 문제발생 소지가 거의 없다.

YARN

메소스랑 구조가 조금 다르긴한데 node manager-mesos agent가 대응이고 리소스 매니저가 mesos master랑 대응한다고 보면 이해하기 쉽다. 메소스처럼 노드매니저가 자원을 리소스 매니저에게 보고하는 형태이다. 

리소스 매니저가 적절한 노드 매니저들을 고르고 각 node manager들은 필요한 자원을 담은 컨테이너를 만들고 그 위에 excutor를 실행하는 방식이다. 만약 deploy mode가 cluster mode라면 node manager 중 한 곳에 spark context(main)이 뜬다.

YARN vs MESOS

두 리소스 매니저의 차이라고 한다면 옵션에서의 차이가 있다. 클라이언트 모드라고 가정을 하고 예를 들어 mesos의 경우에는 total excutor memory 를 지정할 수 있고 yarn에서는 불가능하다. 즉 mesos의 경우에는 excutor memory와 total memory를 지정함으로써 excutor의 갯수를 조절할 수 있다. coarse-grained라서 잘 조절해야 한다.

반면에 YARN에만 있는 기능은 queue를 지정할 수 있다는 것이다. 즉 node manager를 각 queue에 할당하여 배치용, 분석용 등으로 클러스터를 나누어 사용할 수 있다. 장점이라면 queue를 나누더라도 YARN UI 한 곳에서 볼 수 있다는 것이 장점이다. 그런데 그렇게 따지면 mesos도 설치 시 구간을 분할해서 띄운다면 똑같이 사용할 수 있기는 하다. 

정리하면 옵션 넣는 것의 차이와 queue 나눌 수 있는지 여부가 차이라고 본다.

사용자 측면에서는 yarn과 mesos 어떻게 사용하느냐에 따라 양쪽 모두 똑같이 사용이 가능하기 때문에 딱히 뭐가 더 좋냐는 것은 무의미한 것 같다. 어차피 쓰는 사람 입장에서는 리소스 매니저보다 자원 배분을 잘하는 것이 더 중요하기 때문이다.

만약 두개를 병행해서 쓴다는 것은 굳이..? 인것같고 만약 여러 팀과 나눠서 사용한다면 yarn, 단일 팀에서 모든 배치를 관할한다면 mesos를 추천하고 싶다. 

6) 빅데이터 플랫폼 아키텍처에 대하여.. 다른 팀과 협업 시 구성하면 좋은 프레임워크(hive, hue)

무슨 내용을 쓸까하다가 보안적인 부분에 대해서 포스팅을 안했기 때문에 이번에는 이 부분에 대해서 다뤄보려고 한다. 최근 깃랩에 사이드프로젝트를 만들어보니 토큰 발급이 필수로 바뀌어서 문득 다음 포스팅 주제도 보안적인 부분을 다루면 좋겠다고 생각했다.

선택사항이고 개인적으로 보안적인 부분은 최전방에서 최대한 막고 내부 동료들끼리는 개발 효율을 위해 최대한 풀어줘야한다는 생각이지만 이는 업에 따라 법적인 문제(생각보다 고려할 부분이 많다)도 있기 때문에 잘 고려해서 해야한다. 단지 이 포스팅은 한가지 시나리오일 뿐이다.

상황은 하둡 데이터를 접근해야한다. 당연히 관리자는 모두 접근이 가능하겠지만 만약 다른 부서 사람들이 접근을 해야하는 경우는 어떻게 해야할까? 데이터를 직접 뽑아줄 수도 있지만 분석가는 데이터를 불러와서 모델링을 해야하기 때문에 결국 open을 해야하는 이슈가 있다.

하둡 자체에 kerberos 인증이 있기는 하지만 우리가 로그인할 때의 수준이지 데이터 접근에 대한 개념이 아니다.

그럼 어떻게 할까.

1차적으로 하둡 페이지들에 대한 개인의 접근은 모두 막는다. 즉 모든 방화벽을 열어주지 않는다. 굳이 열어준다면 리소스 매니저만 열어준다.

그럼 사람들이 데이터는 어떻게 접근하는가?

hue만 열어준다. (hue는 절대 스케쥴러 용도가 아닌 단순 파티션이나 데이터 확인용!!) 

그럼 단지 쿼리만 날릴 수 있겠지만 불필요한 접근이 없게 될 것이다.

그리고 개발이나 배치는 hive를 통해서만 접근하도록 한다. 즉 개발서버에서 hive만 접근가능하게 열어주고 hive 쿼리만 날릴 수 있게 한다. 어차피 데이터는 쿼리로 접근할테니.. 중요한 것은 hdfs에 direct로 접근하지 못하게 하는 것이다.

체감되는 트랜드 순서가 MR -> Tez -> Spark인 것 같은데 hive on spark를 구성하면 좋겠지만 MR이든 Tez든 Spark든 모두 Spark로 처리할 필요는 없고 데이터의 사이즈에 따라서 MR이나 Tez를 써야하는 경우도 있다.

갑자기 다른데로 샌것같은데 아무튼 이번 포스팅은 보안(?)이라기보다는 어쨌든 아무나 접근 못하게 하는 방법(?)을 제안해보았다.

결론은 개인 컴퓨터에선 hue만 열어줘도 된다.

3) 빅데이터 플랫폼 아키텍처에 대하여.. 데이터를 처리를 위한 Spark

지난 포스팅에서 하둡에 대해서 알아보았다. 최초 하둡을 세팅하고부터는 사실상 전체 리부팅을 할 일이 거의 없고 데이터를 열심히 사용하고 관리를 하게 된다. 이렇게 열심히 모은 데이터를 이제 처리를 해야한다.

빅데이터가 주목받기 시작한 것은 여러가지 이유가 있겠지만 In-memory 방식이 가능하게 되면서 성능이 비약적으로 발전하고 더 많은 아웃풋을 낼 수 있었기 때문이라고 생각한다. 

지금은 Spark가 보편화되고 확고하게 자리 잡았지만 그 전에는 MapReduce 방식으로 데이터를 처리했다. Hive등 다른 프레임워크를 이용하지 않고 자바로 순수 맵리듀스 방식으로 배치를 짠다고 생각해보자. Mapper와 Reducer를 구현해야 한다는 것 자체가 배치의 규모에 따라 상당히 복잡해질 수 있다. 물론 disk I/O기반이라서 아무리 데이터가 커도 터질 염려를 거의 하지 않고 꾸역꾸역 돌아가는 배치를 만들 수는 있는데 너무 비효율적이다.

그리고 혜성처럼 Spark가 등장했고 분산처리를 인메모리 방식으로 처리할 수 있었다. 현재는 3.1.2 버전까지 릴리즈 되었다.

먼저 스파크는 단일머신에서 Standalone mode로 띄워서 사용하거나, 여러 머신을 묶어서 Cluster Mode로 사용가능하다. 분석 클러스터용 서버를 여러대를 구비해놓고 사용하게 된다. 그리고 당연히 데이터와 가까이 있는 것이 좋을 것이다. 

Python이나 Scala로 배치를 구현하게 되는데 하둡에서 데이터를 읽고, 쿼리를 사용하여 Group by 등의 연산을 하고 어딘가에 하둡에 데이터를 쓰는 방식이 주로 이용된다. 이러한 작업은 대부분 비슷하며 ./bin/spark-submit으로 작업을 넘기는데 이 때 어느정도 리소스를 사용할건지, driver와 excuter의 갯수, memory, core 등을 지정할 수 있다. 

초창기 버전1까지는 데이터를 다룰 때 RDD를 사용했고 2버전부터 Dataframe, Dataset을 지원하게 되는데 이는 분산 데이터 구조이며 보통 dataframe이나 dataset을 사용하게 되는데 rdd로 사용해야할 경우에도 dataframe이나 dataset으로 변환시켜서 사용하도록 하자. RDD는 쭉 나열된 데이터라면 dataframe부터는 구조화된 테이블형식이라서 SQL문으로도 처리가 가능하다는 것이 장점이니까..!! 그리고 Spark SQL의 성능이 비약적으로 발전했기 때문에 당연히 사용해야한다.

분산 데이터 구조이다보니 데이터가 흩어져있다. foreach문을 돌릴 때 쪼개져있는 데이터들이 excuter에서 병렬로 돈다는 것이다. 정말 편리한 기능이고 쓰면서 고려해야할 부분은 다른 당연히 데이터가 적절히 분배를 시켜야 좋은 효율을 낼 수 있다. 

스파크로 이런 배치성 작업말고도 streaming 처리도 가능하다.(어쨌든 mini batch이긴 하지만..) 실시간으로 들어오는 곳(kafka 등)에 빨때를 꽂고 주기를 설정하여 배치 작업을 진행하게 되는데 아무래도 이런 부분은 kafka에서 장애가 났을 때 데이터 중복처리 등 offset 관리를 어떻게 할지를 더 고려하게 될 것이다.

그리고 Spark에는 ML 라이브러리도 존재한다. 일반적으로 로직이 단일 머신에서 통으로 돌아가기 때문에 분산 처리를 위한 ML lib에는 많은 것을 지원하고 있지는 않다. 하지만 사용하기 편하고 쉽게 사용가능하도록 가이드도 제공되고 있다.

그리고 GraphX도 존재하는데 사용안해봤다.

Spark의 특성중 고려할 부분은 lazy evaluation이다. 배치를 쭉 짜더라도 순서대로 즉시 도는 것이 아니다. 상태를 계속해서 저장하고 transformation과 action으로 작업이 이루어진다. action이 이루어질 때 비로소 작업이 수행되며 그 전까지는 어떻게 수행되어야 최적으로 수행될지 옵티마이저가 계산을 하면서 최적의 루트를 찾는다. 이는 큰 장점이다.

하지만 transformation이 너무 길어지고 복잡해질 경우 스파크가 이해를 못하고 에러를 뱉는 경우도 존재하는데 이 때 의도적으로 action을 한번씩 취해주면 해결이 된다.(중간 결과를 하둡에 쓰고 다시 읽어서 시작한다던가 하는 방식으로...)

사실 스파크 자체는 사용하는 것은 어렵지 않다. 아파치 재단에서 나온 수 많은 프레임워크들과 호환도 잘 되고 잘만 조합해서 사용하면 좋은 생태계를 구성할 수 있다. 데이터를 쓰거나, 읽을 때 병렬로 수행된다는 것은 성능을 높일 수도 있지만 그 만큼 순간적인 부하가 일어나는 부분을 고려해야하기도 하지만 스루풋을 조절하는 방법도 제공한다. Free하면서 고성능을 낼 수 있기 때문에 하둡+스파크 조합이 확고하게 자리를 잡을 수 있었을 것이다.

Spark Read Parquet - Timestamp and Timezone Confusion

Spark 배치에서 Parquet File을 데이터를 읽을 때 timestamp가 +0900시간이 되는 문제가 발생했다. timezone 문제인데 처음에는 win server to linux 마이그레이션 문제인 줄 알았는데 그게 아니었다. (txt로 저장하면 이런 문제가 없다.)

Spark 버전은 2.4.7 이다.

서버에서 timedatectl을 쳐보면 다음과 같다.

Time Zone이 Asia/Seoul (KST, +0900)으로 되어있다.

따라서 Spark에서 Parquet을 읽을 때 KST로 변환해서 읽는 것 같다. 

그럼 간단하게 생각하면.. UTC로 읽으면 해결될 문제이고 분산환경이기 driver, excuter 모두 해결되어야 할 문제 같은데 이 부분이 조금 찝찝하다.

어쨌든 방법 중에 한가지는 Spark Submit을 할 때 아래 옵션을 준다.

--conf spark.sql.session.timeZone='UTC' \

--conf spark.driver.extraJavaOptions='-Duser.timezone=UTC' \

--conf spark.executor.extraJavaOptions='-Duser.timezone=UTC' \

그럼 해결된다.

또 다른 방법은 코드 상에서 spark session을 생성할 때 옵션을 주고 해결이 가능하다.

.config("spark.sql.session.timeZone", "UTC") 

.config("spark.driver.extraJavaOptions", "-Duser.timezone=UTC") 

.config("spark.executor.extraJavaOptions", "-Duser.timezone=UTC")

아래는 공식문서에서 발췌한 내용이다.

Spark internally stores timestamps as UTC values, and timestamp data that is brought in without a specified time zone is converted as local time to UTC with microsecond resolution. When timestamp data is exported or displayed in Spark, the session time zone is used to localize the timestamp values. The session time zone is set with the configuration 'spark.sql.session.timeZone' and will default to the JVM system local time zone if not set.

분산환경이기도 하고 deploy mode에 따라 옵션들이 어떻게 동작하는지는 테스트해볼 필요가 있겠다.

--20210321 추가

송희진님께서 모든 스파크 배치에 일괄적용을 위해 spark_submit에 아래처럼 추가

exec "${SPARK_HOME}"/bin/spark-class org.apache.spark.deploy.SparkSubmit 

--conf spark.sql.session.timeZone='UTC' \

--conf spark.driver.extraJavaOptions='-Duser.timezone=UTC' \

--conf spark.executor.extraJavaOptions='-Duser.timezone=UTC' \

"$@"

HA hadoop+Spark 조합으로 namespace 사용할 때 발생할 수 있는 에러(UnknownHostException)

현재 HA hadoop + Spark + mesos + azkaban 조합으로 배치를 구성하고 있는데 서버에 직접 들어가서 shell script를 수행하면 정상적으로 동작을 한다.

scala source에서는 hadoop url을 namespace로 접근한다.

그러나 아즈카반에서 수행을 하면 에러를 뱉는다.

Exception in thread "main" java.lang.IllegalArgumentException: java.net.UnknownHostException: hadoop-cluster-bdnode

....

at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)

Caused by: java.net.UnknownHostException: hadoop-cluster-bdnode

... 40 more

....

java.lang.RuntimeException: azkaban.jobExecutor.utils.process.ProcessFailureException: Process exited with code 1

...

Caused by: azkaban.jobExecutor.utils.process.ProcessFailureException: Process exited with code 1

at azkaban.jobExecutor.utils.process.AzkabanProcess.run(AzkabanProcess.java:125)

at azkaban.jobExecutor.ProcessJob.run(ProcessJob.java:304)

shell script에서는 동작할 때에는 hadoop conf(hdfs-site.xml, core-site.xml) 정보를 알고 있다. bashrc에 hadoop_conf_dir을 해놨기 때문이다. 하지만 스케쥴러 azkaban에서 수행하려고 하면 해당 구성을 다시 shell script에 export 해주거나 spark-submit file안에서 알려줘야 한다. 배치를 수행하는 command shell을 모두 바꿀 수가 없어서 그냥 후자로 적용했다.

export HADOOP_HOME=/data01/sw/hadoop

export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop

이제 잘 동작한다.

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으로 처리하는 것도 하나의 옵션이다.