Spark operator: RDD basic conversion operation (4) -union, intersection, subtract

Union

Def union (other: RDD [T]): RDD [T]
The function is relatively simple, that is, the two RDD to merge, not to heavy .

    

  1.   Scala> var rdd1 = sc.makeRDD (1 to 2,1)
    2. 

  2.  Rdd1: org.apache.spark.rdd.RDD [Int] = ParallelCollectionRDD [45] at makeRDD at: 21
    3. 

  3.  
    4. 

  4.  Scala> rdd1.collect
    5. 

  5.  Res42: Array [Int] = Array (1, 2)
    6. 

  6.  
    7. 

  7.  Scala> var rdd2 = sc.makeRDD (2 to 3,1)
    8. 

  8.  Rdd2: org.apache.spark.rdd.RDD [Int] = ParallelCollectionRDD [46] at makeRDD at: 21
    9. 

  9.  
    10. 

  10.  Scala> rdd2.collect
    11. 

  11.  Res43: Array [Int] = Array (2, 3)
    12. 

  12.  
    13. 

  13.  Scala> rdd1.union (rdd2) .collect
    14. 

  14.  Res44: Array [Int] = Array (1, 2, 2, 3)
    15. 

  15.  
    16.

Intersection

Def intersection (other: RDD [T]): RDD [T]
Def node (other: RDD [T], numPartitions: Int): RDD [T]
Def intersection (other: RDD [T], partitioner: Partitioner) (implicit ord: Ordering [T] = null): RDD [T]
The function returns the intersection of two RDDs and goes to the weight .
Parameters numPartitions Specifies the number of partitions for the RDD returned.
The parameter partitioner is used to specify the partition function 
    

  1.   Scala> var rdd1 = sc.makeRDD (1 to 2,1)
    2. 

  2.  Rdd1: org.apache.spark.rdd.RDD [Int] = ParallelCollectionRDD [45] at makeRDD at: 21
    3. 

  3.  
    4. 

  4.  Scala> rdd1.collect
    5. 

  5.  Res42: Array [Int] = Array (1, 2)
    6. 

  6.  
    7. 

  7.  Scala> var rdd2 = sc.makeRDD (2 to 3,1)
    8. 

  8.  Rdd2: org.apache.spark.rdd.RDD [Int] = ParallelCollectionRDD [46] at makeRDD at: 21
    9. 

  9.  
    10. 

  10.  Scala> rdd2.collect
    11. 

  11.  Res43: Array [Int] = Array (2, 3)
    12. 

  12.  
    13. 

  13.  Scala> rdd1.intersection (rdd2) .collect
    14. 

  14.  Res45: Array [Int] = Array (2)
    15. 

  15.  
    16. 

  16.  Scala> var rdd3 = rdd1.intersection (rdd2)
    17. 

  17.  Rdd3: org.apache.spark.rdd.RDD [Int] = MapPartitionsRDD [59] at intersection at: 25
    18. 

  18.  
    19. 

  19.  Scala> rdd3.partitions.size
    20. 

  20.  Res46: Int = 1
    21. 

  21.  
    22. 

  22.  Scala> var rdd3 = rdd1.intersection (rdd2,2)
    23. 

  23.  Rdd3: org.apache.spark.rdd.RDD [Int] = MapPartitionsRDD [65] at intersection at: 25
    24. 

  24.  
    25. 

  25.  Scala> rdd3.partitions.size
    26. 

  26.  Res47: Int = 2
    27. 

  27.  
    28. 

  28.  
    29.

Subtract

Def subtract (other: RDD [T]): RDD [T]
Def subtract (other: RDD [T], numPartitions: Int): RDD [T]
Def subtract (other: RDD [T], partitioner: Partitioner) (implicit ord: Ordering [T] = null): RDD [T]
The function is similar to intersection, but returns an element that appears in RDD and does not appear in otherRDD.
Parameter meaning same intersection 
    

  1.   Scala> var rdd1 = sc.makeRDD (Seq (1,2,2,3))
    2. 

  2.  Rdd1: org.apache.spark.rdd.RDD [Int] = ParallelCollectionRDD [66] at makeRDD at: 21
    3. 

  3.  
    4. 

  4.  Scala> rdd1.collect
    5. 

  5.  Res48: Array [Int] = Array (1, 2, 2, 3)
    6. 

  6.  
    7. 

  7.  Scala> var rdd2 = sc.makeRDD (3 to 4)
    8. 

  8.  Rdd2: org.apache.spark.rdd.RDD [Int] = ParallelCollectionRDD [67] at makeRDD at: 21
    9. 

  9.  
    10. 

  10.  Scala> rdd2.collect
    11. 

  11.  Res49: Array [Int] = Array (3, 4)
    12. 

  12.  
    13. 

  13.  Scala> rdd1.subtract (rdd2) .collect
    14. 

  14.  Res50: Array [Int] = Array (1, 2, 2)
    15.

Commentaires

Enregistrer un commentaire

Posts les plus consultés de ce blog

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

Use Apache Spark to write data to ElasticSearch

ElasticSearch is a Lucene-based search server. It provides a full-text search engine with distributed multiuser capabilities, based on the RESTful web interface. Elasticsearch was developed in Java and released as an open source release under the Apache license terms, which is the current popular enterprise search engine. Designed for cloud computing, to achieve real-time search, stable, reliable, fast, easy to install. This article does not intend to introduce ElasticSearch concept, installation and deployment of knowledge , but directly on how to use Apache Spark data written to ElasticSearch . This article uses the class library is elasticsearch-hadoop , which started from the 2.1 version of the built-in support for Apache Spark function, in the use of elasticsearch-hadoop before we need to elasticsearch-hadoop : < dependency >      < groupId >org.elasticsearch</ groupId >      < artifactId >elasticsearch-hadoop</ artifac
Lire la suite