Spark Streaming Job(Kafka-Consumer)의 성능을 높일 수 있다.

* Kafka topic의 파티션 개수 * 설정 파일의 spark.streaming.kafka.maxRatePerPartition 이다.

* 파티션 개수가 10개 * spark.streaming.kafka.maxRatePerPartition이 5 개이면 => 50 event /duration 성능이 나온다.

그러나 maxRatePerPartition이 너무 크면 OOM이 발생하거나 스트리밍 시스템(sink)에 영향을 줄 수 있다.

스트리밍이 제대로 동작되는지 확인하려면 2가지를 확인할 수 있다.

1) 52 completed batches, 653 records (동작안되면 0 completed batches 라고 뜬다)

2) Processing Time과 Total Delay 처리 Avg 시간을 확인한다. -로 나온다면 처리하는 것이 아니다. 

아무리 수정해도 sbt가 잘 동작하지 않으면, sbt의 로컬 디렉토리인  ~/.sbt 을 다시 지우고 시작하자!!

sbt 에서 컴파일 속도가 나지 않는다고 계속 아래 커맨드를 사용하라고 로그가 나와서 

[warn] Getting the hostname Alvins-MacBook-Pro.local was slow (5003.850955 ms).
This is likely because the computer's hostname is not set.
You can set the hostname with the command:
  scutil --set HostName $(scutil --get LocalHostName).

아래 커맨드를 사용하니 잘 동작한다. 

참고할 내용

실제 코드 

https://github.com/sbt/sbt/pull/3766/files

질문 & 답

https://apple.stackexchange.com/questions/175320/why-is-my-hostname-resolution-taking-so-long

\

scala circe 라이브러리를 사용하다 아래와 같은 에러를 만났다.

대충 코드는 이렇다. Message case class를 json으로 변경하는 것이다.

object MessageType extends Enumeration {
  type MessageType = Value
  val DEL  = Value
  val INSERT = Value
}

case class Message(
    val version: String,
    val pipelineType: MessageType.Value,
    val headers: Map[String, Object],
    val createdAt: String
)

scala enumeration은 아래와 같이 처리해 주었는데.. 역시 동일한 에러이다. 

implicit val decoder: Decoder[MessageType.Value] = 
Decoder.enumDecoder(MessageType)
implicit val encoder: Encoder[MessageType.Value] = 
Encoder.enumEncoder(MessageType)

혹시나 circe는  Map의 value type을 중요하게 본다. 역시 에러다

case class Message(

val version: String,
val pipelineType: MessageType.Value,
val headers: Map[String, Any],
val createdAt: String
)

아래와 같이 value type을 String으로 변경하니 동작한다

case class Message(
val version: String,
val pipelineType: MessageType.Value,
val headers: Map[String, String],
val createdAt: String
)

경험해보니 최대한 간결한 패턴의 case class를 써야 circe가 잘 동작한다. 

spark app을 graceful하게 종료하는 방법이 정리된 링크이다.

https://www.linkedin.com/pulse/how-shutdown-spark-streaming-job-gracefully-lan-jiang/

https://github.com/lanjiang/streamingstopgraceful/blob/master/src/main/scala/com/cloudera/ps/GracefulShutdownExample.scala

spark / sbt 1.3.0-RC1를 사용 중이다. 

sbt test를 실행해

테스트 코드를 모두 완료하고 종료할 때 Java의 ShutdownHookManager에서 Can not access 또는 Can not load 예외가 발생할 수 있다. 

이는 jvm shutdown하면서 shutdownhook에서 정리할 내용보다 먼저 자원이 정리되면서 발생하는 문제이다. 

sbt 실행시 아래와 같은 설정을 추가한다. 더 이상 해당 에러는 발생하지 않는다. 

-Dsbt.classloader.close=false

스칼라에서 enum 타입에 맞는 문자열 값을 찾아 enum을 리턴하는 메소드가 필요하다.

이럴 때는 사용할 만한 메소드로 filter와 find가 적당하다.

filter는 List 배열을 리턴하기에 마땅치 않고 find가 적당한 것 같다.

public enum TestEnum {
   A("a"),
   B("b");

   private String value;

   private TestEnum(String value) {
      this.value = value;
   }

   public String getValue() {
      return value;
   }
}

object Test {

  def main(args: Array[String]): Unit = {

    val wrongEnumType : String = "xxx"

    val wrong = TestEnum.values().find(e => 
                      wrongEnumType.equals(e.getValue))

    if (wrong.isDefined) {
      println(wrong)
    } else {
      println("wrong enum type")
    }

  }
}

https://github.com/sbt/sbt/releases/tag/v1.3.0

sbt 1.3.0에 병렬 라이브러리 다운로드 cousier를 포함하게 되었다.

super shell도 있고..

스칼라의 Cats 라이브러리를 이해하는데 도움이 될만한 자료.

1. 엄익훈님의 Scala Night 발표 자료 

https://www.slideshare.net/ikhoon1/real-world-cats-93894867  

http://tech.kakao.com/2017/09/02/parallel-programming-and-applicative-in-scala/

2. 소프트웨어 미 자료

https://blog.softwaremill.com/9-tips-about-using-cats-in-scala-you-might-want-to-know-e1bafd365f88

3. medium 자료

https://medium.com/@abu_nadhr/scala-cats-library-for-dummies-part-1-8ec47af7a144

https://medium.com/@abu_nadhr/scala-cats-library-for-dummies-part-2-1f2fb31f804c

https://medium.com/@abu_nadhr/scala-cats-library-for-dummies-part-3-fd3b185088f0

https://medium.com/@abu_nadhr/scala-cats-library-for-dummies-part-4-b096f33cc153

4. scala with cats 자료

https://books.underscore.io/scala-with-cats/scala-with-cats.pdf