Analysis of Spark Accumulator commonly used in production (Custom Accumulator)

Train of thought & demand

Refer to the implementation idea of IntAccumulatorParam (as mentioned in the above article):

Trait AccumulatorParam [T] extends AccumulableParam [T, T] {def addAccumulator (T1: t, T2: t): t = {/ / addInPlace has many specific implementation classes / / if you want to implement customization, you have to implement this method addInPlace (T1, T2)}}

Customization can also be implemented in this way, which is compatible with our custom accumulator.

Requirements: here to implement a simple case, using a distributed method to achieve random numbers * * custom AccumulatorParam * * Created by lemon on 2018-7-28. * / object UniqueKeyAccumulator extends AccumulatorParam [Map [Int, Int]] {override def addInPlace (R1: Map [Int, Int], R2: Map [Int, Int]): Map [Int, Int] = {/ / + + for adding r1++r2} override def zero (initialValue: Map [Int, Int]): Map [Int, Int] = {var data: Map [Int] Int] = Map () data}} / * use a custom accumulator Realize the random number * * Created by lemon on 2018-7-28. * / object CustomAccumulator {def main (args: Array [String]): Unit = {val sparkConf = new SparkConf () .setAppName ("CustomAccumulator") .setMaster ("local [2]") val sc = new SparkContext (sparkConf) val uniqueKeyAccumulator = sc.accumulable (Map [int]) Int] () (UniqueKeyAccumulator) val distData = sc.parallelize (1 to 10) val mapCount = distData.map (x = > {val randomNum = new Random (). NextInt (20) / construct a Kmurv pair val map: Map [Int, Int] = Map [Int, Int] (randomNum-> randomNum) uniqueKeyAccumulator + = map}) println (mapCount.count ()) / / gets the key value in the value of the accumulator And print uniqueKeyAccumulator.value.keys.foreach (println) sc.stop ()}}

The running result is as follows: # # ideas & requirements

Refer to the implementation idea of IntAccumulatorParam (as mentioned in the above article):

Trait AccumulatorParam [T] extends AccumulableParam [T, T] {def addAccumulator (T1: t, T2: t): t = {/ / addInPlace has many specific implementation classes / / if you want to implement customization, you have to implement this method addInPlace (T1, T2)}}

Customization can also be implemented in this way, which is compatible with our custom accumulator.

Requirements: here to implement a simple case, using a distributed method to achieve random numbers * * custom AccumulatorParam * * Created by lemon on 2018-7-28. * / object UniqueKeyAccumulator extends AccumulatorParam [Map [Int, Int]] {override def addInPlace (R1: Map [Int, Int], R2: Map [Int, Int]): Map [Int, Int] = {/ / + + for adding r1++r2} override def zero (initialValue: Map [Int, Int]): Map [Int, Int] = {var data: Map [Int] Int] = Map () data}} / * use a custom accumulator Realize the random number * * Created by lemon on 2018-7-28. * / object CustomAccumulator {def main (args: Array [String]): Unit = {val sparkConf = new SparkConf () .setAppName ("CustomAccumulator") .setMaster ("local [2]") val sc = new SparkContext (sparkConf) val uniqueKeyAccumulator = sc.accumulable (Map [int]) Int] () (UniqueKeyAccumulator) val distData = sc.parallelize (1 to 10) val mapCount = distData.map (x = > {val randomNum = new Random (). NextInt (20) / construct a Kmurv pair val map: Map [Int, Int] = Map [Int, Int] (randomNum-> randomNum) uniqueKeyAccumulator + = map}) println (mapCount.count ()) / / gets the key value in the value of the accumulator And print uniqueKeyAccumulator.value.keys.foreach (println) sc.stop ()}}

The running result is as follows: