a. 案例描述

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

b.人口資料的生成

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

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Random;

/**
 * Created by Administrator on 2017/11/13.
 */
 

public class PeopleInfoFileGenerator {
    public static void main(String[] args){
        File file = new File("F:\\PeopleInfo.txt");

        try {
            Random random = new Random();//生成隨機數
            FileWriter fileWriter = new FileWriter(file);//新建一個檔案
            for (int i=1;i<=1000000;i++){   //生成10萬個數字
 

                int height = random.nextInt(220); 
                if (height < 50) {
                    height = height + 50;
                }
             String  gender = getRandomGender(); //性別方法
             if (height < 100 && gender == "M") {
                 height = height + 100;
             }
              if
 
 (height < 100 && gender == "F") {
                  height = height + 40;
              }
             fileWriter.write( i + " " + getRandomGender() + " " + height); //檔案格式：ID 性別 身高
             fileWriter.write(System.getProperty("line.separator"));
            }
            fileWriter.flush();
            fileWriter.close();
            System.out.println("People Information File generated successfully.");
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    public static String getRandomGender(){ //構建一個隨機生成性別方法
        Random random = new Random();
        int randomNum = random.nextInt(2) + 1;
        if( randomNum % 2 == 0){
            return "M";
        }else{
            return "F";
        }
    }
}

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.求出身高統計程式碼實現

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.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import java.util.Arrays;
/**
 * Created by Administrator on 2017/11/17.
 */
public class PeopleInfoCalculator {
    public static void main(String[] args){
        SparkConf sparkConf = new SparkConf().setAppName("PeopleInfoCalculator").setMaster("local[3]");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        JavaRDD<String> dataFile = sc.textFile("F:\\PeopleInfo.txt");

        JavaRDD<String> maleFilterData = dataFile.filter(new Function<String, Boolean>() {//過濾出性別為M的資料
            @Override
            public Boolean call(String s) throws Exception {
                return s.contains("M");
            }
        });
        JavaRDD<String> femaleFilterData = dataFile.filter(new Function<String, Boolean>() {//過濾出性別為F的資料
            @Override
            public Boolean call(String s) throws Exception {
                return s.contains("F");
            }
        });
        JavaRDD<String> maleHeightData = maleFilterData.flatMap(new FlatMapFunction<String, String>() {//得到性別為M的身高資料
            @Override
            public Iterable<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" ")[2]);
            }
        });
        JavaRDD<String> femaleHeightData = femaleFilterData.flatMap(new FlatMapFunction<String, String>() {//得到性別為F的身高資料
            @Override
            public Iterable<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" ")[2]);
            }
        });
        JavaRDD<Integer> maleHeightDataInt = maleHeightData.map(new Function<String, Integer>() {//將字串格式轉化為整型格式
            @Override
            public Integer call(String s) throws Exception {
                return Integer.parseInt(String.valueOf(s));
            }
        });
        JavaRDD<Integer> femaleHeightDataInt = femaleHeightData.map(new Function<String, Integer>() {//將字串格式轉化為整型格式
            @Override
            public Integer call(String s) throws Exception {
                return Integer.parseInt(String.valueOf(s));
            }
        });
        //sortBy(<T>,ascending,numPartitions) 解釋:
        //第一個引數是一個函式，該函式的也有一個帶T泛型的引數，返回型別和RDD中元素的型別是一致的；
        //第二個引數是ascending，這引數決定排序後RDD中的元素是升序還是降序，預設是true，也就是升序；
        //第三個引數是numPartitions，該引數決定排序後的RDD的分割槽個數，預設排序後的分割槽個數和排序之前的個數相等，即為this.partitions.size。
        JavaRDD<Integer> maleHeightLowSort = maleHeightDataInt.sortBy(new Function<Integer,Integer>(){// true表示預設排序，為升序排序，從低到高排
            public Integer call(Integer s) throws Exception {
                return s;
            }
        },true,3);
        JavaRDD<Integer> femaleHeightLowSort = femaleHeightDataInt.sortBy(new Function<Integer,Integer>(){// true表示預設排序，為升序排序，從低到高排
            public Integer call(Integer s) throws Exception {
                return s;
            }
        },true,3);
        JavaRDD<Integer> maleHeightHightSort = maleHeightDataInt.sortBy(new Function<Integer,Integer>(){// false表示為降序排序，從高到低
            public Integer call(Integer s) throws Exception {
                return s;
            }
        },false,3);
        JavaRDD<Integer> femaleHeightHightSort = femaleHeightDataInt.sortBy(new         Function<Integer,Integer>(){// true表示預設排序，為降序排序，從低到高排
            public Integer call(Integer s) throws Exception {
                return s;
            }
        },false,3);
        Integer lowestMale = maleHeightLowSort.first(); //求出升序的第一個數，即最小值
        Integer lowestFemale = femaleHeightLowSort.first();//求出升序的第一個數，即最小值
        Integer highestMale = maleHeightHightSort.first();//求出降序的第一個數，即最大值
        Integer highestFemale = femaleHeightHightSort.first();//求出降序的第一個數，即最大值

        System.out.println("Number of Female Peole:" + femaleHeightData.count());//求出女性的總個數
        System.out.println("Number of Male Peole:" + maleHeightData.count());//求出男性的總個數
        System.out.println("Lowest Male:" + lowestMale);//求出男性最矮身高
        System.out.println("Lowest Female:" + lowestFemale);//求出女性最矮身高
        System.out.println("Highest Male:" + highestMale);//求出男性最高身高
        System.out.println("Highest Female:" + highestFemale);//求出女性最高身高

    }
}

e.執行結果：