Spark学习笔记

		JavaRDD<Integer> nums_1 = jsc.parallelize(Arrays.asList(1,3,5,6));
		//reduce
		Integer sum =nums_1.reduce(new Function2<Integer, Integer, Integer>() {
			@Override
			public Integer call(Integer arg0, Integer arg1) throws Exception {
				return arg0+arg1;
			}
		});
		/**
		 * 1: 1+3;
		 * 2: 4+5;
		 * 3: 9+6;
		 * 结果15
		 */
		System.out.println(sum);
		JavaRDD<Integer> nums_2 = jsc.parallelize(Arrays.asList(1,5,7,9));
		//fold 与reduce操作类似，可以提供一个计算中使用到的初始值，初始值可以不使用，等同于reduce
		Integer sum2 = nums_2.fold(10, new Function2<Integer, Integer, Integer>() {
			@Override
			public Integer call(Integer arg0, Integer arg1) throws Exception {
				return arg0+arg1;
			}
		});
		/**
		 * fold的计算过程：
		 * 1: 10+1;
		 * 2: 11+5;
		 * 3: 16+7;
		 * 4: 23+9;
		 * 5: 10+32
		 * 结果42
		 */
		System.out.println(sum2);

		JavaRDD<Integer> nums_2 = jsc.parallelize(Arrays.asList(1,5,7,9));

		Function2<Double, Integer, Double> add = new Function2<Double, Integer, Double>() {
			@Override
			public Double call(Double arg0, Integer arg1) throws Exception {
				return arg0+arg1;
			}
		};
		Function2<Double, Double, Double> combin = new Function2<Double, Double, Double>() {
			@Override
			public Double call(Double arg0, Double arg1) throws Exception {
				return arg0+arg1;
			}
		};
		double res = nums_2.aggregate(10.0, add, combin);
		System.out.println(res);

		JavaRDD<Integer> nums_2 = jsc.parallelize(Arrays.asList(1,5,7,9));
		
		class AvgCount implements Serializable{
			public int total;
			public int num;
			public AvgCount(int total,int num){
				this.total = total;
				this.num = num;
			}
			public double avg(){
				return total/(double)num;
			}
		}
		
		AvgCount initial = new AvgCount(0, 0);
		Function2<AvgCount,Integer,AvgCount> addAndCount = 
				new Function2<AvgCount, Integer, AvgCount>() {
					@Override
					public AvgCount call(AvgCount arg0, Integer arg1)
							throws Exception {
						arg0.total+=arg1;
						arg0.num++;
						return arg0;
					}
				};
		Function2<AvgCount,AvgCount,AvgCount> combine=
				new Function2<AvgCount,AvgCount,AvgCount>(){
					@Override
					public AvgCount call(AvgCount arg0, AvgCount arg1)
							throws Exception {
						arg0.total+=arg1.total;
						arg0.num+=arg1.num;
						return arg0;
					}
		};
		AvgCount result = nums_2.aggregate(initial, addAndCount, combine);
		/**
		 * 1. initial 在addAndCount中与 RDD的元素进行计算，返回初始值类型的结果res1
		 *    所以Function2<AvgCount(初始值),Integer(RDD元素),AvgCount(返回值)> addAndCount;
		 * 2. initial 与在addAndCount中返回的结果res1进行计算，返回初始值类型的结果res2
		 *    所以Function2<AvgCount(初始值),AvgCount(res1),AvgCount(返回值)> combine;
		 *    可以看出aggregate的返回结果可以与RDD元素的类型不同
		 */
		System.out.println(result.avg());

		JavaRDD<Integer> nums_1 = jsc.parallelize(Arrays.asList(2,1,3,5,6,4,10,1,-2));
		//take
		List<Integer> list_take = nums_1.take(3);
		System.out.println(list_take);//[2, 1, 3]集群下，原则是尝试只访问尽量少的分区
		//top 默认
		List<Integer> list_top_def = nums_1.top(3);
		System.out.println(list_top_def);//[10, 6, 5]默认是按照降序
		//top 指定比较函数
		class SortASC implements Comparator<Integer> , Serializable{//java.util.Comparator
			@Override
			public int compare(Integer o1, Integer o2) {
				return o1<o2?1:-1;
			}
		}
		List<Integer> list_top_asc = nums_1.top(3, new SortASC());
		System.out.println(list_top_asc);//[-2, 1, 1]

		//foreach
		nums_1.foreach(new VoidFunction<Integer>() {
			@Override
			public void call(Integer arg0) throws Exception {
				System.out.println(arg0);
			}
		});

		System.out.println(nums_1.count());
		System.out.println(nums_1.countByValue());//返回Map 值：计数
		//{5=1, 10=1, 1=2, 6=1, 2=1, 3=1, 4=1, -2=1}
		jsc.stop();