RDD Operate Partition

Pipe

데이터를 처리할 때 외부 프로세스 활용에 쓰인다. 운영체제에 종속이므로 아래 예제는 윈도우에서 하면 오류가 발생한다.

test("Pipe") {
  val rdd = sc.parallelize(List("1,2,3", "4,5,6", "7,8,9"))
  val result = rdd.pipe("cut -f 1,3 -d ,")
  println(result.collect.mkString(", "))
}

Coalesce, Repartition

파티션 개수 조정.

Coalesce은 줄이는 것만 처리, Repartition는 늘리는 것도 가능

Coalesce은 셔플(X). Repartition는 셔플(O)

RepartitionAndSortWithinPartitions

Repartition을 하는데 Key에 연산을 먹여서 그 연산 결과를 기준으로 분리하게 만든다. 이건 예제를 보면 빠르게 이해할 수 있다.

정렬까지 된다.

Ex)

test("RepartitionAndSortWithinPartitions") {
  val r = scala.util.Random
  val data = for (i <- 1 to 10) yield (r.nextInt(100), "-")
  val rdd1 = sc.parallelize(data)
  val rdd2 = rdd1.repartitionAndSortWithinPartitions(new HashPartitioner(3))
  rdd2.foreachPartition(it => {
    println("==========");
    it.foreach(v => println( s"{$v }, mod result = ${v._1%3} ") )
  })
}

Result)

==========
{(33,-) }, mod result = 0
==========
{(10,-) }, mod result = 1
{(13,-) }, mod result = 1
{(37,-) }, mod result = 1
{(52,-) }, mod result = 1
{(52,-) }, mod result = 1
{(82,-) }, mod result = 1
==========
{(38,-) }, mod result = 2
{(41,-) }, mod result = 2
{(77,-) }, mod result = 2

PartitionBy

RepartitionAndSortWithinPartitions와 비슷한데, Key를 Sort 할지 말지의 여부 인듯

Ex)

test("RepartitionAndSortWithinPartitions2") {
  val r = scala.util.Random
  val data = for (i <- 1 to 10) yield (r.nextInt(100), "-")
  val rdd1 = sc.parallelize(data)
  val rdd2 = rdd1.partitionBy(new HashPartitioner(3))
  rdd2.foreachPartition(it => {
    println("==========");
    it.foreach(v => println( s"{$v }, mod result = ${v._1%3} ") )
  })
}

Result) sort 가 안되었다...

==========
{(99,-) }, mod result = 0
{(75,-) }, mod result = 0
{(0,-) }, mod result = 0
{(24,-) }, mod result = 0
{(12,-) }, mod result = 0
==========
{(52,-) }, mod result = 1
{(25,-) }, mod result = 1
{(4,-) }, mod result = 1
{(16,-) }, mod result = 1
==========
{(2,-) }, mod result = 2