RDD操作：Transformation与Action

àì夳堔傛蜴生んèń 2022-09-06 10:10 185阅读 0赞

Spark对RDD的操作可以整体分为两类: `Transformation`和`Action`

这里的Transformation可以翻译为转换，表示是针对RDD中数据的转换操作，主要会针对已有的RDD创建一个新的RDD: 常见的有map、flatMap、filter等等

Action可以翻译为执行，表示是触发任务执行的操作，主要对RDD进行最后的操作，比如遍历、 reduce、保存到文件等，并且还可以把结果返回给Driver程序

不管是Transformation里面的操作还是Action里面的操作，我们一般会把它们称之为`算子`，例如:map 算子、reduce算子

其中Transformation算子有一个特性:lazy  
lazy特性在这里指的是，如果一个spark任务中只定义了transformation算子，那么即使你执行这个任务，任务中的算子也不会执行。  
也就是说，transformation是不会触发spark任务的执行，它们只是记录了对RDD所做的操作，不会执行。  
只有当transformation之后，接着执行了一个action操作，那么所有的transformation才会执行。 Spark通过lazy这种特性，来进行底层的spark任务执行的优化，避免产生过多中间结果。

Action的特性:执行Action操作才会触发一个Spark 任务的运行，从而触发这个Action之前所有的 Transformation的执行

以的[《WordCount》][WordCount]代码为例:

// 第一步:创建SparkContext
    val conf = new SparkConf() 
    conf.setAppName("WordCountScala") // 设置任务名称
    	.setMaster("local") // local表示在本地执行 
    // 第二步:加载数据
    var path = "~/hello.txt" 
    if(args.length==1){ 
      path = args(0)
    }
    // 这里通过textFile()方法，针对外部文件创建了一个RDD，linesRDD，实际上，程序执行到这只是创建了一个指向文件的引用，并没有执行
    val linesRDD = sc.textFile(path)
    // 第三步:对数据进行切割，把一行数据切分成一个一个的单词 
    // 这里通过flatMap算子对linesRDD进行了转换操作，把每一行数据中的单词切开，获取了一个逻辑上的wordsRDD
    val wordsRDD = linesRDD.flatMap(_.split(" "))
    // 第四步:迭代words,将每个word转化为(word,1)这种形式 
    // 这个操作和前面分析的flatMap的操作是一样的，最终获取了一个逻辑上的pairRDD，此时里面没有任何数据
    val pairRDD = wordsRDD.map((_,1))
    // 第五步:根据key(其实就是word)进行分组聚合统计 
    // 这个操作也是和前面分析的flatMap操作是一样的，最终获取了一个逻辑上的wordCountRDD，
    val wordCountRDD = pairRDD.reduceByKey(_ + _)
    // 第六步:将结果打印到控制台 
    // 这行代码执行了一个action操作，foreach，此时会触发之前所有transformation算子的执行 
    // 注意:只有当任务执行到这一行代码的时候任务才会真正开始执行计算，如果任务中没有这一行代码，前面的所有算子不会执行
    wordCountRDD.foreach(wordCount=>println(wordCount._1+"--"+wordCount._2))
    // 第七步:停止SparkContext 
    sc.stop()

#### 常用Transformation介绍 ####

<table> 
 <thead> 
  <tr> 
   <th>算子</th> 
   <th>介绍</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>map</td> 
   <td>将RDD中的每个元素进行处理，一进一出</td> 
  </tr> 
  <tr> 
   <td>filter</td> 
   <td>对RDD中每个元素进行判断，返回true则保留</td> 
  </tr> 
  <tr> 
   <td>flatMap</td> 
   <td>与map类似，但是每个元素都可以返回一个或多个新元素</td> 
  </tr> 
  <tr> 
   <td>groupByKey</td> 
   <td>根据key进行分组，每个key对应一个Iterable</td> 
  </tr> 
  <tr> 
   <td>reduceByKey</td> 
   <td>对每个相同key对应的value进行reduce操作</td> 
  </tr> 
  <tr> 
   <td>sortByKey</td> 
   <td>对每个相同key对应的value进行排序操作(全局排序)</td> 
  </tr> 
  <tr> 
   <td>join</td> 
   <td>对两个包含&lt;key,value&gt;对的RDD进行join操作</td> 
  </tr> 
  <tr> 
   <td>distinct</td> 
   <td>对RDD中的元素进行全局去重</td> 
  </tr> 
 </tbody> 
</table>

#### Transformation操作开发实战 ####

下面我们来针对常见的Transformation来具体写一些案例

*  map:对集合中每个元素乘以2
 *  filter:过滤出集合中的偶数
 *  flatMap:将行拆分为单词
 *  groupByKey:对每个大区的主播进行分组
 *  reduceByKey:统计每个大区的主播数量
 *  sortByKey:对主播的音浪收入排序
 *  join:打印每个主播的大区信息和音浪收入
 *  distinct:统计当天开播的大区信息

**Scala版本**

object TransformationOpScala { 
    	def main(args: Array[String]): Unit = {  
    		val sc = getSparkContext 
    		// map:对集合中每个元素乘以2 
    		mapOp(sc) 
    		// filter:过滤出集合中的偶数 
    		filterOp(sc) 
    		// flatMap:将行拆分为单词
    		flatMapOp(sc) 
    		// groupByKey:对每个大区的主播进行分组
    		groupByKeyOp(sc) 
    		groupByKeyOp2(sc) 
    		// reduceByKey:统计每个大区的主播数量 
    		reduceByKeyOp(sc) 
    		// sortByKey:对主播的音浪收入排序
    		sortByKeyOp(sc) 
    		// join:打印每个主播的大区信息和音浪收入 
    		joinOp(sc) 
    		// distinct:统计当天开播的大区信息
    		distinctOp(sc)
    		sc.stop() 
    	}
    	
    	def distinctOp(sc: SparkContext): Unit = { 
    		val dataRDD = sc.parallelize(Array((150001,"US"),(150002,"CN"),(150003,"CN"),(150004, "IN")))
    		// 由于是统计开播的大区信息，需要根据大区信息去重，所以只保留大区信息 
    		dataRDD.map(_._2).distinct().foreach(println(_))
    	}
    	
    	def joinOp(sc: SparkContext): Unit = { 
    	    val dataRDD1 = sc.parallelize(Array((150001,"US"),(150002,"CN"),(150003,"CN"),(150004, "IN")))
    	    val dataRDD2 = sc.parallelize(Array((150001,400),(150002,200),(150003,300),(150004, 100)))
    	    
    	    val joinRDD = dataRDD1.join(dataRDD2)
    		joinRDD.foreach(tup=>{  
    			// 用户id
    			val uid = tup._1
    			val area_gold = tup._2
    			// 大区
    			val area = area_gold._1 //音浪收入
    			val gold = area_gold._2 
    			println(uid+"\t"+area+"\t"+gold)
    		}) 
    	}
    
    	def sortByKeyOp(sc: SparkContext): Unit = { 
    		val dataRDD = sc.parallelize(Array((150001,400),(150002,200),(150003,300),(150004, 100)))  
    		// 由于需要对音浪收入进行排序，所以需要把音浪收入作为key，在这里要进行位置互换 
    		dataRDD.map(tup=>(tup._2,tup._1))
    			.sortByKey(false) // 默认是正序，第一个参数为true，想要倒序需要把这个参数设置为false
    			.foreach(println(_))
    		// sortBy的使用:可以动态指定排序的字段，比较灵活 
    		dataRDD.sortBy(_._2,false).foreach(println(_))
    	}
    
    	def reduceByKeyOp(sc: SparkContext): Unit = { 
    		val dataRDD = sc.parallelize(Array((150001,"US"),(150002,"CN"),(150003,"CN"),(150004, "IN"))) 
    		// 由于这个需求只需要使用到大区信息，所以在map操作的时候只保留大区信息即可 
    		// 为了计算大区的数量，所以在大区后面拼上了1，组装成了tuple2这种形式，这样就可以 
    		dataRDD.map(tup=>(tup._2,1)).reduceByKey(_ + _).foreach(println(_))
    	}
    
    
    	def groupByKeyOp(sc: SparkContext): Unit = { 
    		val dataRDD = sc.parallelize(Array((150001,"US"),(150002,"CN"),(150003,"CN"),(150004, "IN")))
    		// 需要使用map对tuple中的数据进行位置互换，因为我们需要把大区作为key进行分组操作 
    		// 此时的key就是tuple中的第一列,其实在这里就可以把这个tuple认为是一个key-value 
    		// 注意:在使用类似于groupByKey这种基于key的算子的时候，需要提前把RDD中的数据组装成tuple2的形式 
    		// 此时map算子之后生成的新的数据格式是这样的:("US",150001) 
    		// 如果tuple中的数据列数超过了2列怎么办?看groupByKeyOp2 
    		dataRDD.map(tup=>(tup._2,tup._1)).groupByKey().foreach(tup=>{ 
    			// 获取大区信息
    			val area = tup._1 
    			print(area+":") 
    			// 获取同一个大区对应的所有用户id val it = tup._2
    			for(uid <- it){ 
    		      print(uid+" ")
    		    }
    			println() 
    		})
    	}
    
    
    	def groupByKeyOp2(sc: SparkContext): Unit = { 
    		val dataRDD = sc.parallelize(Array((150001,"US","male"),(150002,"CN","female"),(150003,"CN","male"),(150004, "IN","male")))
    		// 如果tuple中的数据列数超过了2列怎么办? 
    		// 把需要作为key的那一列作为tuple2的第一列，剩下的可以再使用一个tuple2包装一下 
    		// 此时map算子之后生成的新的数据格式是这样的:("US",(150001,"male")) 
    		// 注意:如果你的数据结构比较复杂，你可以在执行每一个算子之后都调用foreach打印一 
    		dataRDD.map(tup=>(tup._2,(tup._1,tup._3))).groupByKey().foreach(tup=>{ 
    			//获取大区信息
    			val area = tup._1
    			print(area+":") 
    			// 获取同一个大区对应的所有用户id和性别信息 val it = tup._2
    			for((uid,sex) <- it){ 
    		    	print("<"+uid+","+sex+"> ")
    		    }
    			println() 
    		})
    	}
    
    	def flatMapOp(sc: SparkContext): Unit = { 
    	 	val dataRDD = sc.parallelize(Array("good good study","day day up"))
    		dataRDD.flatMap(_.split(" ")).foreach(println(_))
    	}
    	
    	def filterOp(sc: SparkContext): Unit = { 
    		val dataRDD = sc.parallelize(Array(1,2,3,4,5))
    		dataRDD.filter(_ % 2 ==0).foreach(println(_))
    	}
    	
    	def mapOp(sc: SparkContext): Unit ={ 
    		val dataRDD = sc.parallelize(Array(1,2,3,4,5))
    		dataRDD.map(_ * 2).foreach(println(_))
    	}
    	
    	/** * 获取SparkContext */
    	private def getSparkContext = { 
    	    val conf = new SparkConf()
    	    conf.setAppName("TransformationOpScala")
    	      .setMaster("local")
    	    new SparkContext(conf)
    	}
    }

**Java版本**

import org.apache.hadoop.yarn.webapp.hamlet.Hamlet;
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaPairRDD;
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.JavaSparkContext;
    import org.apache.spark.api.java.function.*;
    import scala.Int;
    import scala.Tuple2;
    import scala.Tuple3;
    import java.util.Arrays;
    import java.util.Iterator;
    
    public class TransformationOpJava { 
    
        public static void main(String[] args) { 
        
    		JavaSparkContext sc = getSparkContext(); 
    		// map:对集合中每个元素乘以2
    		mapOp(sc);
    		// filter:过滤出集合中的偶数 
    		filterOp(sc);
    		// flatMap:将行拆分为单词 
    		flatMapOp(sc); 
    		// groupByKey:对每个大区的主播进行分组
    		groupByKeyOp(sc);
    		groupByKeyOp2(sc); 
    		// reduceByKey:统计每个大区的主播数量 
    		reduceByKeyOp(sc); 
    		// sortByKey:对主播的音浪收入排序
    		sortByKeyOp(sc); 
    		// join:打印每个主播的大区信息和音浪收入
    		joinOp(sc);
    		distinctOp(sc);
    		sc.stop(); 
    	}
    	
    	private static void distinctOp(JavaSparkContext sc) { 
            Tuple2<Integer, String> t1 = new Tuple2<>(150001, "US");
            Tuple2<Integer, String> t2 = new Tuple2<>(150002, "CN");
            Tuple2<Integer, String> t3 = new Tuple2<>(150003, "CN");
            Tuple2<Integer, String> t4 = new Tuple2<>(150004, "IN");
            JavaRDD<Tuple2<Integer, String>> dataRDD = sc.parallelize(Arrays.asList(t1, t2, t3, t4));
    
            dataRDD.map(new Function<Tuple2<Integer, String>, String>() { 
                @Override
                public String call(Tuple2<Integer, String> tup) throws Exception { 
                    return tup._2;
                }
            }).distinct().foreach(new VoidFunction<String>() { 
                @Override
                public void call(String area) throws Exception { 
                    System.out.println(area);
                }
            });
        }
    
    	private static void joinOp(JavaSparkContext sc) { 
            Tuple2<Integer, String> t1 = new Tuple2<>(150001, "US");
            Tuple2<Integer, String> t2 = new Tuple2<>(150002, "CN");
            Tuple2<Integer, String> t3 = new Tuple2<>(150003, "CN");
            Tuple2<Integer, String> t4 = new Tuple2<>(150004, "IN");
            Tuple2<Integer, Integer> t5 = new Tuple2<Integer, Integer>(150001, 400);
            Tuple2<Integer, Integer> t6 = new Tuple2<Integer, Integer>(150002, 200);
            Tuple2<Integer, Integer> t7 = new Tuple2<Integer, Integer>(150003, 300);
            Tuple2<Integer, Integer> t8 = new Tuple2<Integer, Integer>(150004, 100);
    
            JavaRDD<Tuple2<Integer, String>> dataRDD1 = sc.parallelize(Arrays.asList(t1, t2, t3, t4));
            JavaRDD<Tuple2<Integer, Integer>> dataRDD2 = sc.parallelize(Arrays.asList(t5, t6, t7, t8));
    
            JavaPairRDD<Integer, String> dataRDD1Pair = dataRDD1.mapToPair(new PairFunction<Tuple2<Integer, String>, Integer, String>() { 
                @Override
                public Tuple2<Integer, String> call(Tuple2<Integer, String> tup)
                        throws Exception { 
                    return new Tuple2<Integer, String>(tup._1, tup._2);
                }
            });
    
            JavaPairRDD<Integer, Integer> dataRDD2Pair = dataRDD2.mapToPair(new PairFunction<Tuple2<Integer, Integer>, Integer, Integer>() { 
                @Override
                public Tuple2<Integer, Integer> call(Tuple2<Integer, Integer> tup)
                        throws Exception { 
                    return new Tuple2<Integer, Integer>(tup._1, tup._2);
                }
            });
    
            dataRDD1Pair.join(dataRDD2Pair).foreach(new VoidFunction<Tuple2<Integer, Tuple2<String, Integer>>>() { 
                @Override
                public void call(Tuple2<Integer, Tuple2<String, Integer>> tup)
                        throws Exception { 
                    System.out.println(tup);
                }
            });
        }
    	
    	private static void sortByKeyOp(JavaSparkContext sc) { 
            Tuple2<Integer, Integer> t1 = new Tuple2<Integer, Integer>(150001, 400);
            Tuple2<Integer, Integer> t2 = new Tuple2<Integer, Integer>(150002, 200);
            Tuple2<Integer, Integer> t3 = new Tuple2<Integer, Integer>(150003, 300);
            Tuple2<Integer, Integer> t4 = new Tuple2<Integer, Integer>(150004, 100);
            JavaRDD<Tuple2<Integer, Integer>> dataRDD = sc.parallelize(Arrays.asList(t1, t2, t3, t4));
            // 使用sortBy
            dataRDD.sortBy(new Function<Tuple2<Integer, Integer>, Integer>() { 
                @Override
                public Integer call(Tuple2<Integer, Integer> tup) throws Exception { 
                    return tup._2;
                }
            },false,1).foreach(new VoidFunction<Tuple2<Integer, Integer>>() { 
                @Override
                public void call(Tuple2<Integer, Integer> tup)
                        throws Exception { 
                    System.out.println(tup);
                }
            });
        }
    
    	private static void reduceByKeyOp(JavaSparkContext sc) { 
            Tuple2<Integer, String> t1 = new Tuple2<>(150001, "US");
            Tuple2<Integer, String> t2 = new Tuple2<>(150002, "CN");
            Tuple2<Integer, String> t3 = new Tuple2<>(150003, "CN");
            Tuple2<Integer, String> t4 = new Tuple2<>(150004, "IN");
    
            JavaRDD<Tuple2<Integer, String>> dataRDD = sc.parallelize(Arrays.asList(t1, t2, t3, t4));
            dataRDD.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() { 
                @Override
                public Tuple2<String, Integer> call(Tuple2<Integer, String> tup)
                        throws Exception { 
                    return new Tuple2<String, Integer>(tup._2,1);
                }
            }).reduceByKey(new Function2<Integer, Integer, Integer>() { 
                @Override
                public Integer call(Integer i1, Integer i2) throws Exception { 
                    return i1 + i2;
                }
            }).foreach(new VoidFunction<Tuple2<String, Integer>>() { 
                @Override
                public void call(Tuple2<String, Integer> tup) throws Exception { 
                    System.out.println(tup);
                }
            });
        }
    
    	private static void groupByKeyOp(JavaSparkContext sc) { 
    	    Tuple2<Integer,String> t1 = new Tuple2<Integer,String>(150001, "US");
    	    Tuple2<Integer,String> t2 = new Tuple2<Integer,String>(150002, "CN");
    	    Tuple2<Integer,String> t3 = new Tuple2<Integer,String>(150003, "CN");
    	    Tuple2<Integer,String> t4 = new Tuple2<Integer,String>(150004, "IN");
    
    		JavaRDD<Tuple2<Integer, String>> dataRDD = sc.parallelize(Arrays.asList(t1, t2, t3, t4));
    
    		dataRDD.mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() { 
    			@Override
    			public Tuple2<String, Integer> call(Tuple2<Integer, String> tup)
                    throws Exception { 
                	return new Tuple2<String, Integer>(tup._2,tup._1);
           		}
        	}).groupByKey().foreach(new VoidFunction<Tuple2<String, Iterable<Integer>>>() { 
    	        @Override
    			public void call(Tuple2<String, Iterable<Integer>> tup) 
    					throws Exception { 
    				//获取大区信息
    				String area = tup._1; 
    				System.out.print(area+":"); 
    				// 获取同一个大区对应的所有用户id 
    				Iterable<Integer> it = tup._2; 
    				for (Integer uid: it) { 
    	                System.out.print(uid+" ");
    	           	}
                	System.out.println();
            	}
    		}); 
    	}
    	
    	private static void groupByKeyOp2(JavaSparkContext sc) { 
    	    Tuple3<Integer,String,String> t1 = new Tuple3<Integer,String,String>(15001, "US", "male");
    	    Tuple3<Integer,String,String> t2 = new Tuple3<Integer,String,String>(150002, "CN", "female");
    	    Tuple3<Integer,String,String> t3 = new Tuple3<Integer,String,String>(150003, "CN", "male");
    	    Tuple3<Integer,String,String> t4 = new Tuple3<Integer,String,String>(150004, "IN", "female");
    	    JavaRDD<Tuple3<Integer,String,String>> dataRDD = sc.parallelize(Arrays.asList(t1, t2, t3, t4));
    	    
    	    dataRDD.mapToPair(new PairFunction<Tuple3<Integer, String, String>, String, Tuple2<Integer,String>>() { 
                @Override
                public Tuple2<String, Tuple2<Integer, String>> call(Tuple3<Integer, String, String> tup)
                        throws Exception { 
                    return new Tuple2<String, Tuple2<Integer, String>>(tup._2(),new Tuple2<Integer, String>(tup._1(),tup._3()));
                }
            }).groupByKey().foreach(new VoidFunction<Tuple2<String, Iterable<Tuple2<Integer, String>>>>() { 
                @Override
                public void call(Tuple2<String, Iterable<Tuple2<Integer, String>>> tup)
                        throws Exception { 
                    //获取大区信息
                    String area = tup._1;
                    System.out.print(area+":");
                    //获取同一个大区对应的所用用户id和性别信息
                    Iterable<Tuple2<Integer, String>> it = tup._2;
                    for(Tuple2<Integer,String> tu : it){ 
                        System.out.print("<"+tu._1+","+tu._2+"> ");
                    }
                    System.out.println();
                }
            });
    	}
    
    	private static void flatMapOp(JavaSparkContext sc) { 
    	    JavaRDD<String> dataRDD = sc.parallelize(Arrays.asList("good good study", "day day up"));
    	    dataRDD.flatMap(new FlatMapFunction<String, String>() { 
    	        @Override
    	        public Iterator<String> call(String line) throws Exception { 
    	            String[] words = line.split(" ");
    	            return Arrays.asList(words).iterator();
    	        }
    	    }).foreach(new VoidFunction<String>() { 
    	        @Override
    	        public void call(String word) throws Exception { 
    	            System.out.println(word);
    			}
    		});
    	}
    	
        private static void filterOp(JavaSparkContext sc) { 
            JavaRDD<Integer> dataRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5));
            dataRDD.filter(new Function<Integer, Boolean>() { 
                @Override
                public Boolean call(Integer i1) throws Exception { 
                    return i1 % 2 == 0;
                }
            }).foreach(new VoidFunction<Integer>() { 
                @Override
                public void call(Integer i1) throws Exception { 
                    System.out.println(i1);
    			}
    		});
    	}
    
        private static void mapOp(JavaSparkContext sc) { 
            JavaRDD<Integer> dataRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5));
            dataRDD.map(new Function<Integer, Integer>() { 
                @Override
                public Integer call(Integer i1) throws Exception { 
                    return i1 * 2;
                }
            }).foreach(new VoidFunction<Integer>() { 
                @Override
                public void call(Integer i1) throws Exception { 
                    System.out.println(i1);
    			}
    		});
    	}
    	
        private static JavaSparkContext getSparkContext() { 
            SparkConf conf = new SparkConf();
            conf.setAppName("TransformationOpJava")
                    .setMaster("local");
            return new JavaSparkContext(conf);
    	}
    }

#### 常用Action算子 ####

<table> 
 <thead> 
  <tr> 
   <th>算子</th> 
   <th>介绍</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>reduce</td> 
   <td>将RDD中的所有元素进行聚合操作</td> 
  </tr> 
  <tr> 
   <td>collect</td> 
   <td>将RDD中所有元素获取到本地客户端(Driver)</td> 
  </tr> 
  <tr> 
   <td>count</td> 
   <td>获取RDD中元素总数</td> 
  </tr> 
  <tr> 
   <td>take(n)</td> 
   <td>获取RDD中前n个元素</td> 
  </tr> 
  <tr> 
   <td>saveAsTextFile</td> 
   <td>将RDD中元素保存到文件中，对每个元素调用toString</td> 
  </tr> 
  <tr> 
   <td>countByKey</td> 
   <td>对每个key对应的值进行count计数</td> 
  </tr> 
  <tr> 
   <td>foreach</td> 
   <td>遍历RDD中的每个元素</td> 
  </tr> 
 </tbody> 
</table>

#### Action操作开发实战 ####

下面针对常见的Action算子来写一些具体案例

*  reduce:聚合计算
 *  collect:获取元素集合
 *  take(n):获取前n个元素
 *  count:获取元素总数
 *  saveAsTextFile:保存文件
 *  countByKey:统计相同的key出现多少次
 *  foreach:迭代遍历元素

**Scala版本**

import org.apache.spark.{ SparkConf, SparkContext}
    
    
    object ActionOpScala { 
    	def main(args: Array[String]): Unit = { 
        	val sc = getSparkContext
    	    // reduce：聚合计算
    	    reduceOp(sc)
    	    // collect：获取元素集合
    	    collectOp(sc)
    	    // take(n)：获取前n个元素
    	    takeOp(sc)
    	    // count：获取元素总数
    	    countOp(sc)
    	    // saveAsTextFile：保存文件
    	    saveAsTextFileOp(sc)
    	    // countByKey：统计相同的key出现多少次
    	    countByKeyOp(sc)
    	    // foreach：迭代遍历元素
    	    foreachOp(sc)
    	
    	    sc.stop()
    	}
    
    	def foreachOp(sc: SparkContext): Unit = { 
    		val dataRDD = sc.parallelize(Array(1,2,3,4,5))
    		//注意：foreach不仅限于执行println操作，这个只是在测试的时候使用的
    		//实际工作中如果需要把计算的结果保存到第三方的存储介质中，就需要使用foreach
    		//在里面实现具体向外部输出数据的代码
    		dataRDD.foreach(println(_))
    	}
    
      	def countByKeyOp(sc: SparkContext): Unit = { 
    	    val daraRDD = sc.parallelize(Array(("A",1001),("B",1002),("A",1003),("C",1004)))
    	    //返回的是一个map类型的数据
    	    val res = daraRDD.countByKey()
    	    for((k,v) <- res){ 
    	    	println(k+","+v)
    	    }
      	}
    
      	def saveAsTextFileOp(sc: SparkContext): Unit = { 
    	    val dataRDD = sc.parallelize(Array(1,2,3,4,5))
    	    //指定HDFS的路径信息即可，需要指定一个不存在的目录
    	    dataRDD.saveAsTextFile("hdfs://bigdata01:9000/out001")
      	}
    
    	def countOp(sc: SparkContext): Unit = { 
    		val dataRDD = sc.parallelize(Array(1,2,3,4,5))
    	    val res = dataRDD.count()
    	    println(res)
        }
    
    	def takeOp(sc: SparkContext): Unit = { 	
    		val dataRDD = sc.parallelize(Array(1,2,3,4,5))
      		//从RDD中获取前2个元素
      		val res = dataRDD.take(2)
      		for(item <- res){ 
        		println(item)
     		}
    	}
    
    	def collectOp(sc: SparkContext): Unit = { 
    	    val dataRDD = sc.parallelize(Array(1,2,3,4,5))
    	    //collect返回的是一个Array数组
    	    //注意：如果RDD中数据量过大，不建议使用collect，因为最终的数据会返回给Driver进程所在的节点
    	    //如果想要获取几条数据，查看一下数据格式，可以使用take(n)
    	    val res = dataRDD.collect()
       		for(item <- res){ 
         			println(item)
       		}
      	}
    
    	def reduceOp(sc: SparkContext): Unit = { 
    	    val dataRDD = sc.parallelize(Array(1,2,3,4,5))
    	    val num = dataRDD.reduce(_ + _)
    	    println(num)
    	}
    
      	def getSparkContext = { 
    	    val conf = new SparkConf()
    	    conf.setAppName("ActionOpScala")
    	      .setMaster("local")
    	    new SparkContext(conf)
      	}
    }

**Java版本**

import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.JavaSparkContext;
    import org.apache.spark.api.java.function.Function2;
    import org.apache.spark.api.java.function.PairFunction;
    import org.apache.spark.api.java.function.VoidFunction;
    import scala.Tuple2;
    
    import java.util.Arrays;
    import java.util.List;
    import java.util.Map;
    
    
    public class ActionOpJava { 
    
        public static void main(String[] args) { 
            JavaSparkContext sc = getSparkContext();
            // reduce：聚合计算
            reduceOp(sc);
            // collect：获取元素集合
            collectOp(sc);
            // take(n)：获取前n个元素
            takeOp(sc);
            // count：获取元素总数
            countOp(sc);
            // saveAsTextFile：保存文件
            saveAsTextFileOp(sc);
            // countByKey：统计相同的key出现多少次
            countByKeyOp(sc);
            // foreach：迭代遍历元素
            foreachOp(sc);
    
            sc.stop();
        }
    
        private static void foreachOp(JavaSparkContext sc) { 
            JavaRDD<Integer> dataRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5));
            dataRDD.foreach(new VoidFunction<Integer>() { 
                @Override
                public void call(Integer i) throws Exception { 
                    System.out.println(i);
                }
            });
        }
    
        private static void countByKeyOp(JavaSparkContext sc) { 
            Tuple2<String, Integer> t1 = new Tuple2<>("A", 1001);
            Tuple2<String, Integer> t2 = new Tuple2<>("B", 1002);
            Tuple2<String, Integer> t3 = new Tuple2<>("A", 1003);
            Tuple2<String, Integer> t4 = new Tuple2<>("C", 1004);
    
            JavaRDD<Tuple2<String, Integer>> dataRDD = sc.parallelize(Arrays.asList(t1, t2, t3, t4));
            //想要使用countByKey，需要先使用mapToPair对RDD进行转换
            Map<String, Long> res = dataRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, String, Integer>() { 
                @Override
                public Tuple2<String, Integer> call(Tuple2<String, Integer> tup)
                        throws Exception { 
                    return new Tuple2<String, Integer>(tup._1, tup._2);
                }
            }).countByKey();
            for(Map.Entry<String,Long> entry : res.entrySet()){ 
                System.out.println(entry.getKey()+","+entry.getValue());
            }
    
        }
    
        private static void saveAsTextFileOp(JavaSparkContext sc) { 
            JavaRDD<Integer> dataRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5));
            dataRDD.saveAsTextFile("hdfs://bigdata01:9000/out002");
        }
    
        private static void countOp(JavaSparkContext sc) { 
            JavaRDD<Integer> dataRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5));
            long res = dataRDD.count();
            System.out.println(res);
        }
    
        private static void takeOp(JavaSparkContext sc) { 
            JavaRDD<Integer> dataRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5));
            List<Integer> res = dataRDD.take(2);
            for(Integer item : res){ 
                System.out.println(item);
            }
        }
    
        private static void collectOp(JavaSparkContext sc) { 
            JavaRDD<Integer> dataRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5));
            List<Integer> res = dataRDD.collect();
            for(Integer item : res){ 
                System.out.println(item);
            }
        }
    
        private static void reduceOp(JavaSparkContext sc) { 
            JavaRDD<Integer> dataRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5));
            Integer num = dataRDD.reduce(new Function2<Integer, Integer, Integer>() { 
                @Override
                public Integer call(Integer i1, Integer i2) throws Exception { 
                    return i1 + i2;
                }
            });
    
            System.out.println(num);
        }
    
        private static JavaSparkContext getSparkContext() { 
            //创建JavaSparkContext
            SparkConf conf = new SparkConf();
            conf.setAppName("ActionOpJava")
                    .setMaster("local");
            return new JavaSparkContext(conf);
        }
    }

[WordCount]: https://jiandansuifeng.blog.csdn.net/article/details/119747590