Spark operator: RDD key conversion operation (2) -combineByKey, foldByKey

CombineByKey

ConcatenByKey [C] (createCombiner: (V) => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C): RDD [(K, C)]
ConcatenByKey [C] (createCombiner: (V) => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, numPartitions: Int): RDD [(K, C)]
AblinkByKey [C] (createCombiner: (V) => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializer: Serializer = Null): RDD [(K, C)]
This function is used to convert RDD [K, V] to RDD [K, C], where the V and C types can be the same or different.
One of the parameters:
CreateCombiner : combiner function, used to convert the V type to C type, the input parameters for the RDD [K, V] in the V, the output is C
MergeValue : Merge value function, a C type and a V type value into a C type, the input parameters (C, V), the output is C
MergeCombiners : Merge combiner function, used to merge two C-type values ​​into a C type, input parameters (C, C), output to C
NumPartitions : the number of RDD partitions , the default number of existing partitions
Partitioner : partition function, the default is HashPartitioner
MapSideCombine : whether the need to combine at the Map side, similar to MapReduce combine, the default is true
Look at the following example: 
    

  1.   ("A", 1), ("A", 2), ("B", 1), ("B", 2), ("C", 1) ))
    2. 

  2.  Rdd1: org.apache.spark.rdd.RDD [(String, Int)] = ParallelCollectionRDD [64] at makeRDD at: 21
    3. 

  3.  
    4. 

  4.  Scala> rdd1.combineByKey (
    5. 

  5.       (V: Int) => v + "_",   
    6. 

  6.       (C: String, v: Int) => c + "@" + v,  
    7. 

  7.       | (C1: String, c2: String) => c1 + "$" + c2
    8. 

  8.       |) .collect
    9. 

  9.  Res60: Array [(String, String)] = Array ((A, 2_ $ 1_), (B, 1_ $ 2_), (C, 1_))
    10.
Three of the mapping functions are:
CreateCombiner: (V) => C
(V: Int) => v + "_" // After each V value plus the character _, return to the C type (String)
MergeValue: (C, V) => C
(C: String, v: Int) => c + "@" + v // merge C and V types, add character in the middle, return C (String)
MergeCombiners: (C, C) => C
(C1: String, c2: String) => c1 + "$" + c2 // merge C and C, add $ in the middle, return C (String)
Other parameters are default values.
Finally, convert RDD [String, Int] to RDD [String, String].
See example: 
    

  1.   Rdd1.combineByKey (
    2. 

  2.        (V: Int) => List (v),
    3. 

  3.        (C: List [Int], v: Int) => v :: c,
    4. 

  4.        (C1: List [Int], c2: List [Int]) => c1 ::: c2
    5. 

  5.  ) .collect
    6. 

  6.  (B, List (2, 1)), (C, List (1))) () () () () ()
    7.
Finally convert RDD [String, Int] to RDD [String, List [Int]].

FoldByKey

Def foldByKey (zeroValue: V) (func: (V, V) => V): RDD [(K, V)]
Def foldByKey (zeroValue: V, numPartitions: Int) (func: (V, V) => V): RDD [(K, V)]
Def foldByKey (zeroValue: V, partitioner: Partitioner) (func: (V, V) => V): RDD [(K, V)]
This function is used for RDD [K, V] to wrap V by V, and the parameter zeroValue indicates that the zeroValue is applied to V according to the mapping function, initialize V, and then apply the mapping function to the initialized V.
See the example directly: 
    

  1.   ("A", 0), ("A", 2), ("B", 1), ("B", 2), ("C", 1) ))
    2. 

  2.  Scala> rdd1.foldByKey (0) (_ + _). Collect
    3. 

  3.  Res75: Array [(String, Int)] = Array ((A, 2), (B, 3), (C, 1)) 
    4. 

  4.  (0), ("A", 2), and then add the value of each of the keys in rdd1, note that zeroValue = 0, need to initialize V, the mapping function is + First apply zeroValue to each V to get: ("A", 0 + 0), ("A", 2 + 0), ie:
    5. 

  5.  (A, 0 + 2), that is, (A, 2), and then apply the mapping function to the initialized V,
    6. 

  6.  
    7.
Look again 
    

  1.   Scala> rdd1.foldByKey (2) (_ + _). Collect
    2. 

  2.  Res76: Array [(String, Int)] = Array ((A, 6), (B, 7), (C, 3))
    3. 

  3.  // first apply zeroValue = 2 to each V to get: ("A", 0 + 2), ("A", 2 + 2), ie: "A", 2), ("A" 4), and then apply the mapping letter // to the initialized V, and finally get: (A, 2 + 4), ie: (A, 6)
    4. 

  4.  
    5.
Look at multiplication: 
    

  1.   Scala> rdd1.foldByKey (0) (_ * _). Collect
    2. 

  2.  Res77: Array [(String, Int)] = Array ((A, 0), (B, 0), (C, 0))
    3. 

  3.  // first apply zeroValue = 0 to each V, note that the mapping function is multiplied by: ("A", 0 * 0), ("A", 2 * 0)
    4. 

  4.  (A, 0 * 0), that is, (A, 0, 0, 0, 0, 0, 0), and then apply the initialized V to the following: 0)
    5. 

  5.  // other K is the same, and finally got V = 0
    6. 

  6.  
    7. 

  7.  Scala> rdd1.foldByKey (1) (_ * _). Collect
    8. 

  8.  Res78: Array [(String, Int)] = Array ((A, 0), (B, 2), (C, 1))
    9. 

  9.  // When the mapping function is multiplied, it is necessary to set zeroValue to 1 to get the result we want.
    10. 

  10.  
    11. 

  11.  
    12.
When using the foldByKey operator, pay special attention to the mapping function and the value of zeroValue.

Commentaires

Enregistrer un commentaire

Posts les plus consultés de ce blog

Controlling Parallelism in Spark by controlling the input partitions by controlling the input partitions

Image
It is useful to be able to control the degree of parallelism when using Spark. Spark provides a very convenient method to increase the degree of parallelism which should be adequate in practice. This blog entry of the "Inside Spark" series describes the knobs Spark uses to control the degree of parallelism. Controlling the number of partitions in the " local " mode Spark is designed ground up to enable unit testing. To control the degree of parallelism in the local mode simply utilize the function JavaRDD<String> org.apache.spark.api.java.JavaSparkContext.parallelize(List<String> list, int numSlices) The following snippet of code demonstrates this usage- public class SparkControlPartitionSizeLocally { public static void main ( String [] args ) { JavaSparkContext sc = new JavaSparkContext ( "local" , "localpartitionsizecontrol" ); String input = "four score and sev
Lire la suite

Spark performance optimization: shuffle tuning

Image
Article directory 1 shuffle tuning 1.1 Summary of tuning 1.2 Overview of ShuffleManager Development 1.3 HashShuffleManager operating principle 1.3.1 Unoptured HashShuffleManager 1.3.2 Optimized HashShuffleManager 1.4 SortShuffleManager operating principle 1.4.1 General operating mechanism 1.4.2 bypass running mechanism 1.5 shuffle related parameters tuning 1.5.1 spark.shuffle.file.buffer 1.5.2 spark.reducer.maxSizeInFlight 1.5.3 spark.shuffle.io.maxRetries 1.5.4 spark.shuffle.io.retryWait 1.5.5 spark.shuffle.memoryFraction 1.5.6 spark.shuffle.manager 1.5.7 spark.shuffle.sort.bypassMergeThreshold 1.5.8 spark.shuffle.consolidateFiles 2 write in the last words Shuffle Summary of tuning Most of the performance of Spark operations is mainly consumed in the shuffle link, because the link contains a large number of disk IO, serialization, network data transmission and othe
Lire la suite

Spark optimization

Image
One operation and maintenance 1. Master hang up, standby restart is also invalid Master defaults to 512M of memory, when the task in the cluster is particularly high, it will hang, because the master will read each task event log log to generate spark ui, the memory will naturally OOM, you can run the log See that the master of the start through the HA will naturally fail for this reason. solve Increase the Master's memory spark-env.sh , set in the master node spark-env.sh : export SPARK_DAEMON_MEMORY 10g # 根据你的实际情况 Reduce the job information stored in the Master memory spark.ui.retainedJobs 500 # 默认都是1000 spark.ui.retainedStages 500 Hang up or suspend Sometimes we will see the web node in the web ui disappear or in the dead state, the task of running the node will report a variety of lost worker errors, causing the same reasons and the above, worker memory to save a lot of ui The information leads to gc when the heartbeat is lost
Lire la suite