• spark.executor.memory는 실행기에 사용할 수 있는 총 메모리 크기를 정의합니다.
• spark.storage.memoryFraction(기본 ~60%)은 지속된 RDD를 저장하는 데 사용할 수 있는 메모리 양을 정의합니다.
• spark.shuffle.memoryFraction(기본 ~20%)은 무작위 재생용으로 예약된 메모리 양을 정의합니다.
• spark.storage.unrollFraction 및 spark.storage.safetyFraction(총 메모리의 ~30%) - 이러한 값은 Spark에서 내부적으로 사용되므로 변경하지 않아야 합니다

LRU 란?

Least recently used (LRU)[edit]

Discards the least recently used items first. This algorithm requires keeping track of what was used when, which is expensive if one wants to make sure the algorithm always discards the least recently used item. General implementations of this technique require keeping "age bits" for cache-lines and track the "Least Recently Used" cache-line based on age-bits. In such an implementation, every time a cache-line is used, the age of all other cache-lines changes. LRU is actually a family of caching algorithms with members including 2Q by Theodore Johnson and Dennis Shasha,^[3] and LRU/K by Pat O'Neil, Betty O'Neil and Gerhard Weikum.^[4]

The access sequence for the below example is A B C D E D F.

In the above example once A B C D gets installed in the blocks with sequence numbers (Increment 1 for each new Access) and when E is accessed, it is a miss and it needs to be installed in one of the blocks. According to the LRU Algorithm, since A has the lowest Rank(A(0)), E will replace A.

RDD Persistence

One of the most important capabilities in Spark is persisting (or caching) a dataset in memory across operations. When you persist an RDD, each node stores any partitions of it that it computes in memory and reuses them in other actions on that dataset (or datasets derived from it). This allows future actions to be much faster (often by more than 10x). Caching is a key tool for iterative algorithms and fast interactive use.

You can mark an RDD to be persisted using the persist() or cache() methods on it. The first time it is computed in an action, it will be kept in memory on the nodes. Spark’s cache is fault-tolerant – if any partition of an RDD is lost, it will automatically be recomputed using the transformations that originally created it.

In addition, each persisted RDD can be stored using a different storage level, allowing you, for example, to persist the dataset on disk, persist it in memory but as serialized Java objects (to save space), replicate it across nodes. These levels are set by passing a StorageLevel object (Scala,Java, Python) to persist(). The cache() method is a shorthand for using the default storage level, which is StorageLevel.MEMORY_ONLY (store deserialized objects in memory). The full set of storage levels is

• Cache 와 Persist 의 차이 : Cache 는 Default 값(MEMORY_AND_DISK)이 정해져 있는 반면 Persist 는 위와 같은 Storage Level 을 설정할 수 있음
• Dataset 과 RDD 의 Default Storage Level 이 다름 : RDD = MEMORY_ONLY, Dataset = MEMORY_AND_DISK

※ local 공간이 충분치 않으면, out of space 가 발생할 수 있음

Which Storage Level to Choose?

Spark’s storage levels are meant to provide different trade-offs between memory usage and CPU efficiency. We recommend going through the following process to select one:

• If your RDDs fit comfortably with the default storage level (MEMORY_ONLY), leave them that way. This is the most CPU-efficient option, allowing operations on the RDDs to run as fast as possible.

• If not, try using MEMORY_ONLY_SER and selecting a fast serialization library to make the objects much more space-efficient, but still reasonably fast to access. (Java and Scala)

• Don’t spill to disk unless the functions that computed your datasets are expensive, or they filter a large amount of the data. Otherwise, recomputing a partition may be as fast as reading it from disk.

• Use the replicated storage levels if you want fast fault recovery (e.g. if using Spark to serve requests from a web application). All the storage levels provide full fault tolerance by recomputing lost data, but the replicated ones let you continue running tasks on the RDD without waiting to recompute a lost partition.

An SQL join clause combines records from two or more tables. This operation is very common in data processing and understanding of what happens under the hood is important. There are several common join types: INNER, LEFT OUTER, RIGHT OUTER, FULL OUTER and CROSS or CARTESIAN.

Join which uses the same table is a self-join. If an operation uses equality operator, it is equi-join, otherwise, it is non-equi-join.

This article covers different join types in Apache Spark as well as examples of slowly changed dimensions (SCD) and joins on non-unique columns.

Sample data

All subsequent explanations on join types in this article make use of the following two tables, taken from Wikipedia article. The rows in these tables serve to illustrate the effect of different types of joins and join-predicates.

Employees table has a nullable column. To express it in terms of statically typed Scala, one needs to use Option type.

Department table does not have nullable columns, type specification could be omitted.

Inner join

Following SQL code

SELECT * FROM employee INNER JOIN department ON employee.DepartmentID = department.DepartmentID;

could be written in Spark as

Beautiful, is not it? Spark automatically removes duplicated “DepartmentID” column, so column names are unique and one does not need to use table prefix to address them.

Left outer join

Left outer join is a very common operation, especially if there are nulls or gaps in a data. Note, that column name should be wrapped into scala Seq if join type is specified.

Other join types

Spark allows using following join types: inner, outer, left_outer, right_outer, leftsemi. The interface is the same as for left outer join in the example above.

For cartesian join column specification should be omitted:

Warning: do not use cartesian join with big tables in production.

Join expression, slowly changing dimensions and non-equi join

Spark allows us to specify join expression instead of a sequence of columns. In general, expression specification is less readable, so why do we need such flexibility? The reason is non-equi join.

One application of it is slowly changing dimensions. Assume there is a table with product prices over time:

There are two products only: steak and fish, price of steak has been changed once. Another table consists of product orders by day:

Our goal is to assign an actual price for every record in the orders table. It is not obvious to do using only equality operators, however, spark join expression allows us to achieve the result in an elegant way:

This technique is very useful, yet not that common. It could save a lot of time for those who write as well as for those who read the code.

Inner join using non primary keys

Last part of this article is about joins on non unique columns and common mistakes related to it. Join (intersection) diagrams in the beginning of this article stuck in our heads. Because of visual comparison of sets intersection we assume, that result table after inner join should be smaller, than any of the source tables. This is correct only for joins on unique columns and wrong if columns in both tables are not unique. Consider following DataFrame with duplicated records and its self-join:

Note, that size of the result DataFrame is bigger than the source size. It could be as big as n², where n is a size of source.

Conclusion

The article covered different join types implementations with Apache Spark, including join expressions and join on non-unique keys.

Apache Spark allows developers to write the code in the way, which is easier to understand. It improves code quality and maintainability.

total input 30만줄 X 66rows = 1980만 rows = 약 5GB

output = ranks => about 1980만 rows  = 약 549MB

*spark 테스트 중 unusuable node 등의 오류가 나오는데, local disk 의 점유가 높아지면 taskmanager (yarn) 가 일시적으로 node 를 kill 하고 다시 복구시킨다.

다만 그것이 반복되다가 결과가 제대로 나오지 않는 경우도 있다.

*위를 해결하기위해서 현재 DISK (100GB) 를 cluster 마다 추가 할당하였고, temp 파일 저장 경로를 그쪽으로 변경하였다.

* hadoop 의 경우 local_temp 경로에 shuffle 되는 결과들이 쏟아진다. 다만 계정에  temp 폴더 쓰기권한 등이 없으면 에러가난다.

* hadoop 실행 도중 중지시 temp 경로에(주로 /tmp) 로그와 셔플 중간 파일들이 쌓여있을 수 있다. 그파일은 주기적으로 정리 필요

 *spark 에서 pagerank 의 경우에는 ittr 반복이 될수록 local 에 점유정도가 어느정도인지 확인필요하고, 해소방안도 필요하다.

* hadoop 에서 conf.set("mapreduce.textoutputformat.separator", "::"); 등 config 셋팅등에 유의 하자 (yarn 또한 마찬가지) Hadoop - map reduce configuration

 *현재 돌고있는 프로세스는 yarn application -list 확인 후  -kill 등으로 죽일 수 있다.

hadoop 실행시간( log 기준)

https://stackoverflow.com/questions/34175386/precise-definitions-of-map-shuffle-merge-and-reduce-times-in-hadoop

Default zone 확인

포트추가

--add-port=<portid>[-<portid>]/<protocol> 옵션을 사용하여 포트 추가

 설정 reload

val query = wordCounts.writeStream

.outputMode("complete") or .outputMode("append") or .outputMode("update") 

case class DeviceData(device: String, deviceType: String, signal: Double, time: DateTime)

val df: DataFrame = ... // streaming DataFrame with IOT device data with schema { device: string, deviceType: string, signal: double, time: string }
val ds: Dataset[DeviceData] = df.as[DeviceData]    // streaming Dataset with IOT device data

// Select the devices which have signal more than 10
df.select("device").where("signal > 10")      // using untyped APIs 
ds.filter(_.signal > 10).map(_.device)         // using typed APIs

// Running count of the number of updates for each device type
df.groupBy("deviceType").count()                          // using untyped API

// Running average signal for each device type
import org.apache.spark.sql.expressions.scalalang.typed
ds.groupByKey(_.deviceType).agg(typed.avg(_.signal))    // using typed API

1. deduplication
   
   

   / Without watermark using guid column streamingDf.dropDuplicates("guid")   // With watermark using guid and eventTime columns streamingDf   .withWatermark("eventTime", "10 seconds")   .dropDuplicates("guid", "eventTime")

   watermark - watermark 까지 (참고로 이전 date 는 중복제거가 안 될 수 있음)
   without - sate 되는대까지

2. Arbitary State
   Since Spark 2.2, this can be done using the operation mapGroupsWithState and the more powerful operation flatMapGroupsWithState. Both operations allow you to apply user-defined code on grouped Datasets to update user-defined state. For more concrete details, take a look at the API documentation (Scala/Java) and the examples (Scala/Java).
   위의 Operation 은 유저가원하는(임의의) state 를 만든다. 다만 그 State 가 어떻게 활용되는지는 별이야기가없다.

     3. Unsupported!

Unsupported Operations

There are a few DataFrame/Dataset operations that are not supported with streaming DataFrames/Datasets. Some of them are as follows.

• Multiple streaming aggregations (i.e. a chain of aggregations on a streaming DF) are not yet supported on streaming Datasets.

• Limit and take first N rows are not supported on streaming Datasets.

• Distinct operations on streaming Datasets are not supported.

• Sorting operations are supported on streaming Datasets only after an aggregation and in Complete Output Mode.

• Few types of outer joins on streaming Datasets are not supported. See the support matrix in the Join Operations section for more details.

In addition, there are some Dataset methods that will not work on streaming Datasets. They are actions that will immediately run queries and return results, which does not make sense on a streaming Dataset. Rather, those functionalities can be done by explicitly starting a streaming query (see the next section regarding that).

• count() - Cannot return a single count from a streaming Dataset. Instead, use ds.groupBy().count() which returns a streaming Dataset containing a running count.

• foreach() - Instead use ds.writeStream.foreach(...) (see next section).

• show() - Instead use the console sink (see next section).

Some sinks are not fault-tolerant because they do not guarantee persistence of the output and are meant for debugging purposes only. See the earlier section on fault-tolerance semantics. Here are the details of all the sinks in Spark.

https://stackoverflow.com/questions/28700624/spark-submit-scheduling-in-cron

crontab -e 를 통해서 script 를 정기적으로 돌릴때,

#!/bin/sh

spark submit >> 동작안함

#!/bin/bash 

/경로/spark-submit >> 동작함

아래와 같이 bash 로 수정하고 실행시키면 동작한다.

bash is the most common shell used as a default shell for users of Linux systems. It is a spiritual descendent of other shells used throughout Unix history. Its name, bash is an abbreviation of Bourne-Again Shell, an homage to the Bourne shell it was designed to replace, though it also incorporates features from the C Shell and the Korn Shell.

/bin/sh is an executable representing the system shell. Actually, it is usually implemented as a symbolic link pointing to the executable for whichever shell is the system shell. The system shell is kind of the default shell that system scripts should use. In Linux distributions, for a long time this was usually a symbolic link to bash, so much so that it has become somewhat of a convention to always link /bin/sh to bash or a bash-compatible shell. However, in the last couple of years Debian (and Ubuntu) decided to switch the system shell from bash to dash - a similar shell - breaking with a long tradition in Linux (well, GNU) of using bash for /bin/sh. Dash is seen as a lighter, and much faster, shell which can be beneficial to boot speed (and other things that require a lot of shell scripts, like package installation scripts).

If your script requires features only supported by bash, use #!/bin/bash.

But if at all possible, it would be good to make sure your script is POSIX-compatible, and use #!/bin/sh, which should always, quite reliably, point to the preferred POSIX-compatible system shell in any installation.

spark-shell command 에서는 write  할때 문제가 되지 않는데,

spark-submit 에서 정의된 함수를 사용해서 write 하는 과정에서 not serializable 에러가 나오는 경우가 있다.

예를들면,

가령 같은 scala class 안에 function 을 정의하고 바로 사용 하게되면,

org.apache.spark.SparkException: Task not serializable when write  에러가 나는데, 

object getgap {
  def getgap(s1:String , s2 : String) : Long =   {
    if(s1.length<8 || s2.length<8)   0
    val format = new java.text.SimpleDateFormat("HH:mm:ss.SSS")
    val v1 = format.parse(s1)
    val v2 = format.parse(s2)
    (v2.getTime() - v1.getTime())  / 100
  }
}

  import org.apache.spark.sql.SparkSession
  val spark = SparkSession.builder().master("yarn").appName("message_summerization").getOrCreate()
  import spark.implicits._
  //val inputpath = "/user/sangil55/m_input"
  val rd = spark.read.json(inputpath)


  val df_timegap = df_semi2.map(row=>(row.getString(0), row.getString(1), row.getString(2), getgap(row.getString(1),row.getString(2)) ) )
    .withColumnRenamed("_1", "jobid").withColumnRenamed("_2", "starttime").withColumnRenamed("_3", "endtime").withColumnRenamed("_4", "gap")
    .sort("starttime")

이 getgap 함수를 object 로 빼주게되면 static class 로 인식하여
잘돌아간다.

spark 환경에서 같은클래스에서 선언된 함수는 따로 클래스화되지 않고, 임시저장되는데 이걸 바로 사용하게되면
mapper 에서 임시저장된 함수를 제대로 분산시키지 못하는듯하다.

이와 비슷한 에러가 자주나오니,
spark-submit 환경에서 function 으로 동작하는 놈들은 웬만하면 따로 뺴주는게 좋다.

rdd- groupbyKey 와 달리 spark dataset 은 groupby 로 리듀스를 할 수 있는데, 

이때 .agg 형식으로 aggregation 을 한다.

다만  .agg 안에 들어가는 function 은 단한 선언으로 안되고, 

udf (user defined aggregationfuncion) 을 만들어서 구현해주어야 하는데, 동작이 뭔가 까다롭다

일단 shell 이나 main 에서는

아래와 같이 실행 해주면되고

val td = rd.select($"jobid".alias("jobid"), $"time".alias("starttime"), $"time".alias("endtime"), $"result".alias("report_result"), $"result".alias("error"))
//val td = rd.select($"jobid".alias("jobid"),$"time".alias("starttime"))

import CustomAgg._
val minagg = new minAggregation()
val maxagg = new maxAggregation()
val erragg = new errorAggregation()

UDF 는 아래와 같이 만들어준다.

(max 값찾는 agg / min 값찾는 agg , error 찾는 agg 등이다.)

update 에서는 파티션내의 병합이 

merge 에서는 파티션간의 병합이 이루어지는 것 같다.

ackage CustomAgg

import org.apache.spark.sql.Row
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types._
import org.apache.spark.unsafe.types.UTF8String

import scala.collection.mutable.ArrayBuffer

object CustomAggregation
  {

    class maxAggregation extends UserDefinedAggregateFunction {
    // Input Data Type Schema
    //def inputSchema: StructType = StructType(Array(StructField("col5", StringType)))

    // Intermediate Schema
    //def bufferSchema = StructType(Array(    StructField("col5_collapsed", StringType)))

      def inputSchema = new StructType().add("x", StringType)
      def bufferSchema = new StructType().add("buff", ArrayType(StringType))
    // Returned Data Type .
    def dataType: DataType = StringType

    // Self-explaining
    def deterministic = true

    // This function is called whenever key changes
    def initialize(buffer: MutableAggregationBuffer) = {
      buffer.update(0, ArrayBuffer.empty[String])// initialize array
    }

    // Iterate over each entry of a group
    def update(buffer: MutableAggregationBuffer, input: Row) = {
      if (!input.isNullAt(0)) {
        println(input.getString(0))
        println(buffer.getSeq[String](0).mkString)
        if(buffer.getSeq[String](0).mkString.isEmpty)
          buffer.update(0, Seq(input.getString(0)))
        else if(buffer.getSeq[String](0).mkString.compareTo(input.getString(0)) > 0)
          buffer.update(0, buffer.getSeq[String](0))
        else
          buffer.update(0, Seq(input.getString(0)))
        println(buffer.getSeq[String](0).mkString)
      }
    }


    // Merge two partial aggregates
    def merge(buffer1: MutableAggregationBuffer, buffer2: Row) = {
      if(buffer1.getSeq[String](0).mkString.compareTo(buffer2.getSeq[String](0).mkString) > 0)
        buffer1.update(0, buffer1.getSeq[String](0))
      else
        buffer1.update(0, buffer2.getSeq[String](0))
    }

      def evaluate(buffer: Row) = UTF8String.fromString(
        buffer.getSeq[String](0).mkString(","))
  }

    class minAggregation extends UserDefinedAggregateFunction {
      // Input Data Type Schema
      //def inputSchema: StructType = StructType(Array(StructField("col5", StringType)))

      // Intermediate Schema
      //def bufferSchema = StructType(Array(    StructField("col5_collapsed", StringType)))

      def inputSchema = new StructType().add("x", StringType)
      def bufferSchema = new StructType().add("buff",StringType)
      // Returned Data Type .
      def dataType: DataType = StringType

      // Self-explaining
      def deterministic = true

      // This function is called whenever key changes
      def initialize(buffer: MutableAggregationBuffer) = {
        buffer.update(0, "")// initialize array
      }

      // Iterate over each entry of a group
      def update(buffer: MutableAggregationBuffer, input: Row) = {
        if (!input.isNullAt(0)) {

          if(buffer.getString(0).isEmpty)
            buffer.update(0, input.getString(0))
          else if(buffer.getString(0).compareTo(input.getString(0)) < 0)
            buffer.update(0, buffer.getString(0))
          else
            buffer.update(0, input.getString(0))

          println( "updated :" + buffer.getString(0))
        }
      }


      // Merge two partial aggregates
      def merge(buffer1: MutableAggregationBuffer, buffer2: Row) = {

        if(buffer1.getString(0).compareTo(buffer2.getString(0))< 0 && !buffer1.getString(0).isEmpty)
          buffer1.update(0, buffer1.getString(0))
        else
          buffer1.update(0, buffer2.getString(0))
      }

      def evaluate(buffer: Row) = UTF8String.fromString(
        buffer.getString(0))
    }


    class errorAggregation extends UserDefinedAggregateFunction {
      // Input Data Type Schema
      //def inputSchema: StructType = StructType(Array(StructField("col5", StringType)))

      // Intermediate Schema
      //def bufferSchema = StructType(Array(    StructField("col5_collapsed", StringType)))

      def inputSchema = new StructType().add("x", StringType)
      def bufferSchema = new StructType().add("buff", ArrayType(StringType))
      // Returned Data Type .
      def dataType: DataType = StringType

      // Self-explaining
      def deterministic = true

      // This function is called whenever key changes
      def initialize(buffer: MutableAggregationBuffer) = {
        buffer.update(0, Seq(""))// initialize array
      }

      def checkerror (str : String) : Boolean = {

        if(str.compareTo("Info") == 0) false
        if(str.compareTo("0") == 0) false
        if(str.compareTo("Success") == 0) false
        if(str.compareTo("") == 0) false
        true
      }

      // Iterate over each entry of a group
      def update(buffer: MutableAggregationBuffer, input: Row) = {
        if (!input.isNullAt(0)) {
          if(checkerror(input.getString(0)))
          {
            input.getString(0)
            buffer.update(0, Seq(input.getString(0)))
          }
        }
      }


      // Merge two partial aggregates
      def merge(buffer1: MutableAggregationBuffer, buffer2: Row) = {
        println(buffer1.getSeq[String](0).toString())
        println(buffer2.getSeq[String](0).toString())
        buffer1.update(0, buffer1.getSeq[String](0) ++ buffer2.getSeq[String](0))
      }

      def evaluate(buffer: Row) = UTF8String.fromString(
        buffer.getSeq[String](0).mkString(","))
    }


    class GeometricMean extends UserDefinedAggregateFunction {
      // This is the input fields for your aggregate function.
      override def inputSchema: org.apache.spark.sql.types.StructType =
        StructType(StructField("value", StringType) :: Nil)

      // This is the internal fields you keep for computing your aggregate.
      override def bufferSchema: StructType = StructType(
        StructField("count", LongType) ::
          StructField("product", StringType) :: Nil
      )

      // This is the output type of your aggregatation function.
      override def dataType: DataType = StringType

      override def deterministic: Boolean = true

      // This is the initial value for your buffer schema.
      override def initialize(buffer: MutableAggregationBuffer): Unit = {
        buffer(0) = ""
        buffer(1) = " "
      }

      // This is how to update your buffer schema given an input.
      override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
        buffer(0) = buffer.getAs[String](0) + 1
        buffer(1) = buffer.getAs[String](1) + input.getAs[String](0)
      }

      // This is how to merge two objects with the bufferSchema type.
      override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
        buffer1(0) = buffer1.getAs[Long](0) + buffer2.getAs[Long](0)
        buffer1(1) = buffer1.getAs[Double](1) * buffer2.getAs[Double](1)
      }

      // This is where you output the final value, given the final value of your bufferSchema.
      override def evaluate(buffer: Row): Any = {
        math.pow(buffer.getDouble(1), 1.toDouble / buffer.getLong(0))
      }
    }

https://www.cloudera.com/documentation/enterprise/5-8-x/topics/admin_spark_tuning.html

spark 에는 여러가지 종류의 transformation / action 등이 있다. 종류에 따라서 동작이다른데, 

-map, filter : narrow transformation 

-coalesce : still narrow 

-groupbyKey ReducebyKey : wide transformation >> Shuffle 이 많이 일어나고, stage 가 생기게 된다.

이처럼 stage 와 shuffle 이 늘어날 수록 incur high disk & network I/O 등이 일어나 expensive 한 작업이 일어나게된다.

또는 numpatition 등에 의해 stage boundary 를 trigger 할수도 있다.

스파크를 튜닝 하는 방법은 여러가지가 있을 수 있다.

1. chooosing Optimal transformatons
   a. map + reducebyKey   <<< aggregateByKey
   b. join <<< cogroup  (if already grouped by key)
   후자가 더빠르다.
2. Shuffle 자체를 줄이는 방법
   reduce by key 등의 transformation 은 lazy evaluation 이 일어나는데,  부모 Partition 갯수가 같게 세팅되면, 한번에 일어날 수 있음
   그렇기 떄문에 ReducebyKey( XXX, numpartion = X) 에 들어가는 X 가 중요하다.
   
   
   
   
   
   
3. 보통은 shuffle 을 줄이는 것의 performence가 늘어난다. 그러나 다음의 경우에는  그렇지 않다.
   -inputfile 이 크고 Partition 이 늘어야 하는 경우 (shuffle 도 늘어남)
   -Reduce 개수가 가 한번에 몰린경우 Tree / Tree Aggregate 등을 사용하면 shuffle 이 늘어나지만, perfomance 가 늘어날 수 있다.
   
   
4. secondary sort 
   -RepartitionAndSortWithinPartitions >> repartition + sort
   전자가 shuffle machinery 를 이용하여 후자가 사용하는 sort 보다 빠르다.   이 sorting 은 join 에서도 쓰인다.
   
   
5. Memory tunning 
   -a. Yarn 을 튜닝하자
    특히 Yarn.nodemanager.resouce.memory-mb  / Yarn.nodemanager.resouce.cpu-vcores 를 이용하여 리소스매니저의 리소스를 확보해야한다.
   -b. Executors-cores 과 executor-memory 를 실행시 옵션으로 셋팅  단, heap size 는 max(excutormemory * 0.07 ,384)
   -c. AM 은  client deploy mode 와 cluster deploy mode 가 있는데, 전자는  1GB 1core 를 client 에서 할당하고
      gnwksms --Driver 옵션을 이용하여 따로 설정에 유의
   -d. number of cores for excutor 는 5이하가 되는 것이 좋다.
   -e.tiny excutor → perf 저하
   
   
6. tuning num of partitions
    보통 Partition 의 개수의 따르나 아래동작은 다르다.
   coalesce → fewer 
   union → sum of its parent partitions number
   cartesian → product of its parents
   
   특히 Join Cogroup / GrouBuKey 등의 동작은 Memory 사용률이 높고 exceed 가 발생하면, DISK I/O 와 GC 등이 일어나 속도가 감소된다. (파티션별 용량 설정이 중요)
   Repartitions 으로 partition 개수를 늘릴 수 있다
   heuristic 한 방법으로 OPTIMAL Partition # = 전체 코어수 *3 정도 가 좋다는데 정확하지않고, spark 에서도 1.5배씩 늘려가며 찾으라는 말을 한다.

         보통은 아주 많은 경우가 적은경우보다는 좋다고 한다.

7. Reduce data size 이건 당연한 소리

8. 출력 format 을 Sequence 파일로 해라