scala에서 uuid 생성하는 방법(사실 자바 UUID 생성하는 방법)

UUID 객체만 얻을려면 다음을 호출한다.

Spark의 Task는 하나의 partition을 가진다. 

SparkContext의 parallelize를 실행해서 hadoop HDFS에 데이터를 저장할 때, 병렬(spark core) 개수만큼 파티션이 생긴다. 전문 용어로 level of parallelism이라 한다 (hadoop에서도 reduce 개수만큼 파티션 개수가 생긴다)

HDFS에 저장할 용량이 크지 않다면 spark core 개수와 상관없이 하나의 파티션 파일로 모아두는 것이 좋을 수 있다.(전문용어로 aggregation 이라 한다)

이를 위해 repartition 또는 coalesce를 사용할 수 있다.

(coalesce는 random shuffle을 하지 않고, repartition은 random shuffle을 하기 때문에 성능 상 차이점이 있다) 

repartition을 적용하니 잘 작동한다. hadoop의 HDFS에 하나의 파일로 모아졌다. 

sc.parallelize(birthdayUsers.map(x => s"${person.id}\t${person.name}")).repartition(1)

* 참고 : 문서에 따르면 aggregation하는 방법은 3가지로 있다.

http://spark.apache.org/docs/latest/programming-guide.html

* 소스 비교 

repartition은 coalesce에서 shuffle만 한 것이다.

/**
 * Return a new RDD that has exactly numPartitions partitions.
 *
 * Can increase or decrease the level of parallelism in this RDD. Internally, this uses
 * a shuffle to redistribute data.
 *
 * If you are decreasing the number of partitions in this RDD, consider using `coalesce`,
 * which can avoid performing a shuffle.
 */
def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T] = withScope {
  coalesce(numPartitions, shuffle = true)
}

coalesce 소스이다. 

/**
 * Return a new RDD that is reduced into `numPartitions` partitions.
 *
 * This results in a narrow dependency, e.g. if you go from 1000 partitions
 * to 100 partitions, there will not be a shuffle, instead each of the 100
 * new partitions will claim 10 of the current partitions.
 *
 * However, if you're doing a drastic coalesce, e.g. to numPartitions = 1,
 * this may result in your computation taking place on fewer nodes than
 * you like (e.g. one node in the case of numPartitions = 1). To avoid this,
 * you can pass shuffle = true. This will add a shuffle step, but means the
 * current upstream partitions will be executed in parallel (per whatever
 * the current partitioning is).
 *
 * Note: With shuffle = true, you can actually coalesce to a larger number
 * of partitions. This is useful if you have a small number of partitions,
 * say 100, potentially with a few partitions being abnormally large. Calling
 * coalesce(1000, shuffle = true) will result in 1000 partitions with the
 * data distributed using a hash partitioner.
 */
def coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord: Ordering[T] = null)
    : RDD[T] = withScope {
  if (shuffle) {
    /** Distributes elements evenly across output partitions, starting from a random partition. */
    val distributePartition = (index: Int, items: Iterator[T]) => {
      var position = (new Random(index)).nextInt(numPartitions)
      items.map { t =>
        // Note that the hash code of the key will just be the key itself. The HashPartitioner
        // will mod it with the number of total partitions.
        position = position + 1
        (position, t)
      }
    } : Iterator[(Int, T)]

    // include a shuffle step so that our upstream tasks are still distributed
    new CoalescedRDD(
      new ShuffledRDD[Int, T, T](mapPartitionsWithIndex(distributePartition),
      new HashPartitioner(numPartitions)),
      numPartitions).values
  } else {
    new CoalescedRDD(this, numPartitions)
  }
}

소스를 살펴본 것은 

repartition은 네트워크 부하는 올라가지만 무난하게 사용할 수 있는데, coalesce는 동작은 하는데, 100kb보다 크다는 warning이 발생하였지만 잘 동작했기 때문이다. 

문서를 보면, numPartitions=1일 때 계산 비용이 높아질 수 있다고 하며 shuffle을 true로 설정하라 한다. 

However, if you're doing a drastic coalesce, e.g. to numPartitions = 1, this may result in your computation taking place on fewer nodes than you like (e.g. one node in the case of numPartitions = 1). To avoid this, you can pass shuffle = true. This will add a shuffle step, but means the current upstream partitions will be executed in parallel (per whatever the current partitioning is).

스칼라에서는 두 리스트를 결합을 다양하게 진행할 수 있다.

a와 b라는 리스트가 아래와 같다고 한다.

:: 예시이다. 

::: 예시이다. 

List.concat 예시이다.

++ 예시이다.

List 생성자를 이용한 예시이다.

zip 예시이다. 

스칼라에서 zip과 unzip을 사용해서 얻을 수 있는 특별한 이득이 있을만한 예시를 소개한다. 

보통 zip과 unzip은 붙였다가 떼였다가 할 때 사용할 수 있다. 

val b = (List(1, 2, 3) zip List(List(0.0), List(0.2), List(0.3))).unzip
println(b)

다음은 결과이다. 

리스트를 붙여서 특정 정보가 있을 때는 보여주지 않으려 할 때, 

즉, 다음 예시에서 List(0.0)이 존재하면 아예 그 값은 따로 저장하고 싶지 않을 때 사용할 수 있다. 

val a = (List(1, 2, 3) zip List(List(0.0), List(0.2), List(0.3)))
                    .filter(!_._2.forall(_ == 0.0)).unzip
println(a)

이전 예시를 실행하면, 첫번째 값이 List(0.0)이라 filter에서 걸렸기 때문에 1 키는 저장하지 않았다. 

(List(2, 3),List(List(0.2), List(0.3)))

() => A 는 매개변수를 받지 않고 A를 돌려주는 함수이다. 이런 형태를 Thunk라라 한다. 함수형 언어에서 Thunk는 immutable을 지원하는 pure한 비동기 함수이다.  ( () => A 또는 => A 라는 형태를 가진다)

scala와 spark의 내부 코드에 종종 보이는 코드이다. 

아래 예시를 보면, ifThunk 함수는 첫 번째 매개변수를 Boolean 타입으로 받고 

        onTrue와 onFalse라는 Thunk 함수를 받는다. 

이럴 때는 호출하는 부분에서 () =>{ }의 형태로 사용해야 한다

def ifThunk[A] (cond:Boolean, onTrue:() => A, onFalse:() => A): A = {
  if (cond) onTrue() else onFalse()
}
val a = 1
ifThunk(a > 0, () => {println("true")}, () => {println("false")})

결과는 true이다.

() => A 표현식을  =>A로 바꿔도 동일한 기능을 사용할 수 있다. 대신 호출하는 부분에서 () =>을 사용하지 않는다. 

def ifThunk2[A] (cond:Boolean, onTrue: => A, onFalse: => A): A = {
  if (cond) onTrue else onFalse
}
val b = 1
ifThunk2(b > 0, {println("true")}, {println("false")})

// not print
ifThunk2(b > 0, () => {println("true")}, () => {println("false")})

결과를 보면, true 하나만 출력한다.

(참고로 아래 코드도 잘 동작한다)

ifThunk2(b > 0, println("true"), println("false"))

() => {}를 사용한 라인에서는 출력이 없다....  

println으로 확인해 보면, <function0>을 리턴한 코드가 된 것이다. 

println(ifThunk2(b > 0, () => {println("true")}, () => {println("false")}))

() => A 대신  => A를 사용하면, 표현식을 알아서 처리한다. 아래는 에러가 발생하지 않는다. 

ifThunk2(b > 0, println("true"), 3)

결과는 true가 출력된다. 

만약, 3이라는 표현식에 결과가 나타나도록 하려면 다음과 같다.

println(ifThunk2(b < 0, {println("true")}, 3))

결과는 3이 출력된다. ifThunk2는 3을 리턴하였다. 유연성 짱이다. 

그리고, 매개변수를 함수로 넘길 때 다른 이슈를 설명한다. 

multiply는 간단히 Thunk를 이용해 곱하는 함수이다. 

def multiply(b:Boolean, i: => Int) = if(b) i * i else 0
println(multiply(true, 2))

결과 값은 4이다. 

실제 호출 순을 살펴본다. 

println(multiply(true, {println("a") ; 2}))

결과는 다음과 같다. 매개변수에 사용하는 함수는 호출될 때마다 발생한다. 2번 호출했으니 2번 출력된다. 

이를 한번에 모았다가 한번에 함수를 호출하려면, lazy val를 사용한다. 

def multiply2(b:Boolean, i: => Int) = { lazy val j = i; if(b) j + j else 0 }
println(multiply2(true, 2))

println(multiply2(true, {println("b") ; 2}))

결과는 다음과 같다. 즉 넘긴 함수는 코드 안에서 여러 번 호출되더라도 한번만 호출된다. 이를 memoization이라 한다.

 (lazy j에서 확인할 수 있다)

lazy val과 thunk가 lazy evaluation 의 핵심 개념이다. 

*참조 : https://github.com/fpinscala/fpinscala

두 변수와 두 변수를 계산하는 이항 함수(binary function)을 계산하는 함수는 

for 내장으로 변경할 수 있다. 

def map1[A, B, C](a: Option[A], b: Option[B], f: (A, B) => C): Option[C] = {
  a flatMap (aa => b map(bb => f(aa, bb)))
}

// for comprehension for 내장
def map2[A, B, C](a: Option[A], b: Option[B], f: (A, B) => C): Option[C] = {
  for {
    aa <- a
    bb <- b
  } yield f(aa, bb)
}


val a1 = Some(10)
val b1 = Some(20)
def fff1(x: Int, y: Int): Option[Int] = {
  Some(x + y)
}
println(map1(a1, b1, fff1))
println(map2(a1, b1, fff1))

val a2 = Some("xyz")
val b2 = Some("abc")
def fff2(x: String, y: String): Option[String] = {
  Some(x + y)
}
println(map1(a2, b2, fff2))
println(map2(a2, b2, fff2))

결과는 다음과 같다. 

REPL에서 :type을 사용하면 타입을 볼 수 있다. 

trait Trait {
  val m = 1
}

class Clazz extends Trait {
  var m1 = 333333
  var m2 = "aa"
}

object Execise extends App {
  val clazz = classOf[Clazz]
  val name = clazz.getName
  println("name : " + name)

  //println
  val constructors = clazz.getConstructors
  println("constructors : " + constructors.mkString(" "))

  println
  val methods = clazz.getMethods
  println("methods : " + methods.mkString(" "))

  println
  val fields = clazz.getFields
  println("fields : " + fields.mkString(" "))

  println
  val annotations = clazz.getAnnotations
  println("annotations : " + annotations.mkString(" "))

  println
  val superClass = clazz.getSuperclass
  println("superClass : " + superClass)

  println
  val typeParameters = clazz.getTypeParameters
  println("typeParameters : " + typeParameters.mkString(" "))
}

결과는 다음과 같다. 

스칼라의 classOf의 리턴타입은 java.lang.Class이다.(컴파일러에서 제대로 생성된다)

def classOf[T]: Class[T] = null

또한, 결과를 통해 추측할 수 있겠지만, 일부 내용을 파싱하면서 에러난 부분이 발생했다. 아직 scala 클래스 객체를 완벽히 소화한 상태는 아니다. 이는 스칼라에서 쓸 수 있는 TypeTags를 활용하면 의미있는 정보를 얻을 수 있다. 

먼저 스칼라의 ClassTag이다. ClassTag를 사용하면 컴파일러가 암시 ClassTag를 생성한다. TypeTags의 약한 버전이다. 

import scala.reflect.ClassTag

case class Person(id: Int)

def makeArray[T: ClassTag] (elements: T*) = Array[T](elements: _*)

println(makeArray(1,2,3).mkString(" "))

println(makeArray("one", "two", "three").mkString(" "))

println(makeArray(1, "xxx").mkString(" "))

println(makeArray(Person(1), Person(2)).mkString(" "))

결과는 다음과 같다.

다음은 TypeTag를 활용한 예시이다. 강력한 스칼라 reflection api이다. 

class SuperClazz {
  val m = 1
}

class Clazz extends SuperClazz {
  var m1 = 333333
  var m2 = "aa"
}

import scala.reflect.runtime.universe._
def toType[T: TypeTag](t : T) = typeOf[T]

println(toType(1))
println(toType(false))
println(toType(new Clazz))
println(toType(List(1, "aa")))
println(toType(Seq(1, true, 3.14)))

val t = toType(2)
println(t.typeSymbol)
println(t.erasure)
println(t.typeArgs)
println(t.baseClasses)
println(t.companion)

println()
val t1 = toType(new Clazz)
println(t1.members)
println(t1.baseClasses)

결과는 다음과 같다.

bash에서 사용할 수 있는 간단한 스칼라 커맨드 유틸리티를 소개한다. 

scala REPL을 그대로 사용하고 간단하게 사용할 수 있다. 

실행한 코드는 다음과 같다. 

가장 실수한 부분은 sys.exit였다. scala 실행할 때 종료하면 sys.exit하듯이 스크립트에서도 사용하면 된다. 

REPL은 좀 제약이 많아서 될 수 있으면 코드베이스가 더 나은 것 같긴 하지만, 필요하다면 스크립트도 괜찮은 방법일 수 있다. 

 scala에서 프로세스를 실행하려면 scala.sys.process._을 import하고, 문자열에 .! (또는 !)을 실행한다. (마치 regular expression처럼..) 

import scala.sys.process._
val result = "ls -al /Users/samuel.kim/dev".!
println(result)

결과는 다음과 같다. 

주의할 점은 !는 ProcessBuilder 클래스의 메소드로서 리턴 값은 Int이다. 즉, 결과를 출력하고 리턴 값으로 0만 받는 형태이다. 

def ! : Int

전체 결과를 String으로 얻으려면, !!을 사용한다. 

import scala.sys.process._
val result = "ls -al /Users/samuel.kim/dev".!!
println(result)

결과는 다음과 같다.

!!의 리턴 값은 String이다. 

def !! : String

그리고, List로 저장할 수 있다. 

import scala.sys.process._
val result = List("ls -al /Users/samuel.kim/dev".!!)
result.foreach(println)
println(result.size)

결과는 다음과 같다. 

파이프라인을 사용하면, 에러가 발생한다.

import scala.sys.process._
val result = List("ls -al /Users/samuel.kim/dev | wc -l".!!)
result.foreach(println)

다음처럼 커맨드 단위로 잘개 쪼개고 #을 파이프에 붙여야 한다. 

import scala.sys.process._
val result = List(("ls -al /Users/samuel.kim/dev" #| "wc -l ").!!)
result.foreach(println)

결과는 다음과 같다. 

이번에는 >>이라는 표준 출력(stdout) append를 사용한다. #|처럼 #>>을 사용하며, #>> 다음에는 new File()을 생성한다. 

import java.io.File
import scala.sys.process._
val result = ("ls -al /Users/samuel.kim/dev" #| "wc -l " 
               #>> new File(s"/Users/samuel.kim/dev/a.txt")).!
println(result)
val result2 = ("cat /Users/samuel.kim/dev/a.txt".!!)
println(result2)

결과는 다음과 같다.

스칼라에서는 표준 입력/출력을 처리할 수 있다.

자세한 내용은 scala.sys.process.ProcessBuilder api를 참조한다.

이를 바탕으로 함수를 만들 수 있다. 

def get(filter: String) = {
  val command = s"ls -al /Users/samuel.kim/dev" #| s"grep $filter "
  println(command)
  command
}

val result = get("xx").!!
println(result)

결과는 다음과 같다.

scala는 쉘의 리턴값이 정상이 아니면 Exception이 발생한다. Try를 써주면 Exception 대신 Failure로 처리해 준다.

def get(filter: String) = {
  val command = s"ls -al /Users/samuel.kim/dev" #| s"grep $filter "
  println(command)
  command
}

val result = Try(get("bbb").!!)
println(result)

결과는 다음과 같다.

scala에서 클래스를 생성하는 방법이다.

class Person {
}
new Person()

class Person(id: Int)
new Person(1)

클래스는 this라는 보조 생성자로 객체를 생성할 수 있다. 

그리고, Option 타입을 사용하면 풍부한 생성이 가능하다. 

하지만, case class를 사용하여 new 객체를 하지 않도록 할 때 일부 코드에서는 컴파일 에러가 발생한다. 

case class Person(id: Int, name: String, department: Option[String]) {
  def this(id: Int) = this(id, "no-name")

  def this(id: Int, name: String) = this(id, name, Some("no-name"))
}

new Person(1, "matt")
new Person(2)
new Person(3, "samuel", Some("develop"))

//Person(1) // compile error
//Person(2, "jack") // compile error
Person(3, "samuel", Some("develop"))

보조 생성자의 한계가 보이기 때문에, 이때 동반자 객체(companion object)를 활용한다.

이전에 컴파일 에러가 발생했던 부분이 사라지고 더욱 Rich하게 사용할 수 있다. 

case class Person(id: Int, name: String, department: Option[String]) {
  def this(id: Int) = this(id, "no-name")

  def this(id: Int, name: String) = this(id, name, Some("no-name"))
}

object Person {
  def apply(id: Int) = new Person(id)

  def apply(id: Int, name: String) = new Person(id)
}

new Person(1, "matt")
new Person(2)
new Person(3, "samuel", Some("develop"))

Person(1)
Person(2, "jack")
Person(3, "samuel", Some("develop"))