原文引自：http://blog.csdn.net/fengzhimohan/article/details/78564610

a. 案例描述

本案例假設我們需要對某個省的人口 (10萬) 性別還有身高進行統計，需要計算出男女人數，男性中的最高和最低身高，以及女性中的最高和最低身高。本案例中用到的源文件有以下格式, 三列分別是 ID，性別，身高 (cm)，格式如下：?

b.人口數據的生成

利用Java語言隨機生成一組人口數據，包括序列ID，性別M/F，身高cm，代碼如下：

1 import java.io.File; 2 import java.io.FileWriter; 3 import java.io.IOException; 4 import java.util.Random; 5 6 /** 7 * Created by Administrator on 2017/11/13. 8 */ 9 public class PeopleInfoFileGenerator { 10 public static void main(String[] args){ 11 File file = new File("F:\\PeopleInfo.txt"); 12 13 try { 14 Random random = new Random();//生成隨機數 15 FileWriter fileWriter = new FileWriter(file);//新建一個文件 16 for (int i=1;i<=1000000;i++){ //生成10萬個數字 17 int height = random.nextInt(220); 18 if (height < 50) { 19 height = height + 50; 20 } 21 String gender = getRandomGender(); //性別方法 22 if (height < 100 && gender == "M") { 23 height = height + 100; 24 } 25 if (height < 100 && gender == "F") { 26 height = height + 40; 27 } 28 fileWriter.write( i + " " + getRandomGender() + " " + height); //文件格式：ID 性別 身高 29 fileWriter.write(System.getProperty("line.separator")); 30 } 31 fileWriter.flush(); 32 fileWriter.close(); 33 System.out.println("People Information File generated successfully."); 34 }catch (IOException e){ 35 e.printStackTrace(); 36 } 37 } 38 39 public static String getRandomGender(){ //構建一個隨機生成性別方法 40 Random random = new Random(); 41 int randomNum = random.nextInt(2) + 1; 42 if( randomNum % 2 == 0){ 43 return "M"; 44 }else{ 45 return "F"; 46 } 47 } 48 }

c. 實例過程分析

對于這個案例，我們要分別統計男女的信息，那么很自然的想到首先需要對于男女信息從源文件的對應的 RDD 中進行分離，這樣會產生兩個新的 RDD，分別包含男女信息；其次是分別對男女信息對應的 RDD 的數據進行進一步映射，使其只包含身高數據，這樣我們又得到兩個 RDD，分別對應男性身高和女性身高；最后需要對這兩個 RDD 進行排序，進而得到最高和最低的男性或女性身高。?
第一步，先分離男女信息，使用 filter 算子過濾條件包含”M” 的行是男性，包含”F”的行是女性；第二步我們需要使用 map 算子把男女各自的身高數據從 RDD 中分離出來；第三步我們需要使用 sortBy 算子對男女身高數據進行排序。

特別注意：RDD 轉化的過程中需要把身高數據轉換成整數，否則 sortBy 算子會把它視為字符串，那么排序結果就會受到影響，例如 身高數據如果是：123,110,84,72,100，那么升序排序結果將會是 100,110,123,72,84，顯然這是不對的。

d.求出身高統計代碼實現

1 import org.apache.spark.SparkConf; 2 import org.apache.spark.api.java.JavaRDD; 3 import org.apache.spark.api.java.JavaSparkContext; 4 import org.apache.spark.api.java.function.FlatMapFunction; 5 import org.apache.spark.api.java.function.Function; 6 import java.util.Arrays; 7 /** 8 * Created by Administrator on 2017/11/17. 9 */ 10 public class PeopleInfoCalculator { 11 public static void main(String[] args){ 12 SparkConf sparkConf = new SparkConf().setAppName("PeopleInfoCalculator").setMaster("local[3]"); 13 JavaSparkContext sc = new JavaSparkContext(sparkConf); 14 JavaRDD<String> dataFile = sc.textFile("F:\\PeopleInfo.txt"); 15 16 JavaRDD<String> maleFilterData = dataFile.filter(new Function<String, Boolean>() {//過濾出性別為M的數據 17 @Override 18 public Boolean call(String s) throws Exception { 19 return s.contains("M"); 20 } 21 }); 22 JavaRDD<String> femaleFilterData = dataFile.filter(new Function<String, Boolean>() {//過濾出性別為F的數據 23 @Override 24 public Boolean call(String s) throws Exception { 25 return s.contains("F"); 26 } 27 }); 28 JavaRDD<String> maleHeightData = maleFilterData.flatMap(new FlatMapFunction<String, String>() {//得到性別為M的身高數據 29 @Override 30 public Iterable<String> call(String s) throws Exception { 31 return Arrays.asList(s.split(" ")[2]); 32 } 33 }); 34 JavaRDD<String> femaleHeightData = femaleFilterData.flatMap(new FlatMapFunction<String, String>() {//得到性別為F的身高數據 35 @Override 36 public Iterable<String> call(String s) throws Exception { 37 return Arrays.asList(s.split(" ")[2]); 38 } 39 }); 40 JavaRDD<Integer> maleHeightDataInt = maleHeightData.map(new Function<String, Integer>() {//將字符串格式轉化為整型格式 41 @Override 42 public Integer call(String s) throws Exception { 43 return Integer.parseInt(String.valueOf(s)); 44 } 45 }); 46 JavaRDD<Integer> femaleHeightDataInt = femaleHeightData.map(new Function<String, Integer>() {//將字符串格式轉化為整型格式 47 @Override 48 public Integer call(String s) throws Exception { 49 return Integer.parseInt(String.valueOf(s)); 50 } 51 }); 52 //sortBy(<T>,ascending,numPartitions) 解釋: 53 //第一個參數是一個函數，該函數的也有一個帶T泛型的參數，返回類型和RDD中元素的類型是一致的； 54 //第二個參數是ascending，這參數決定排序后RDD中的元素是升序還是降序，默認是true，也就是升序； 55 //第三個參數是numPartitions，該參數決定排序后的RDD的分區個數，默認排序后的分區個數和排序之前的個數相等，即為this.partitions.size。 56 JavaRDD<Integer> maleHeightLowSort = maleHeightDataInt.sortBy(new Function<Integer,Integer>(){// true表示默認排序，為升序排序，從低到高排 57 public Integer call(Integer s) throws Exception { 58 return s; 59 } 60 },true,3); 61 JavaRDD<Integer> femaleHeightLowSort = femaleHeightDataInt.sortBy(new Function<Integer,Integer>(){// true表示默認排序，為升序排序，從低到高排 62 public Integer call(Integer s) throws Exception { 63 return s; 64 } 65 },true,3); 66 JavaRDD<Integer> maleHeightHightSort = maleHeightDataInt.sortBy(new Function<Integer,Integer>(){// false表示為降序排序，從高到低 67 public Integer call(Integer s) throws Exception { 68 return s; 69 } 70 },false,3); 71 JavaRDD<Integer> femaleHeightHightSort = femaleHeightDataInt.sortBy(new Function<Integer,Integer>(){// true表示默認排序，為降序排序，從低到高排 72 public Integer call(Integer s) throws Exception { 73 return s; 74 } 75 },false,3); 76 Integer lowestMale = maleHeightLowSort.first(); //求出升序的第一個數，即最小值 77 Integer lowestFemale = femaleHeightLowSort.first();//求出升序的第一個數，即最小值 78 Integer highestMale = maleHeightHightSort.first();//求出降序的第一個數，即最大值 79 Integer highestFemale = femaleHeightHightSort.first();//求出降序的第一個數，即最大值 80 81 System.out.println("Number of Female Peole:" + femaleHeightData.count());//求出女性的總個數 82 System.out.println("Number of Male Peole:" + maleHeightData.count());//求出男性的總個數 83 System.out.println("Lowest Male:" + lowestMale);//求出男性最矮身高 84 System.out.println("Lowest Female:" + lowestFemale);//求出女性最矮身高 85 System.out.println("Highest Male:" + highestMale);//求出男性最高身高 86 System.out.println("Highest Female:" + highestFemale);//求出女性最高身高 87 88 } 89 }

?

e.運行結果：

?

轉載于:https://www.cnblogs.com/jinggangshan/p/8109329.html

總結

以上是生活随笔為你收集整理的spark 应用场景2-身高统计的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。