spark 應用場景1-求年齡平均值

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

該案例中，我們將假設我們需要統計一個 10 萬人口的所有人的平均年齡，當然如果您想測試 Spark 對於大數據的處理能力，您可以把人口數放的更大，比如 1 億人口，當然這個取決於測試所用集群的存儲容量。假設這些年齡信息都存儲在一個文件里，並且該文件的格式如下，第一列是 ID，第二列是年齡。如下圖格式： 

 

以下利用java隨機生成10萬個人口年齡文件

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 DataFileGenerator {
    public static void main(String[] args){
        File file = new File("F:\\DataFile.txt");
        try {
            FileWriter fileWriter = new FileWriter(file);
            Random rand = new Random();
            for (int i=1;i<=100000;i++){
                fileWriter.write(i +" " + (rand.nextInt(100)+1));
                fileWriter.write(System.getProperty("line.separator"));
            }
            fileWriter.flush();
            fileWriter.close();

        }catch(IOException e){
            e.printStackTrace();
        }
    }
}

 

場景分析：

要計算平均年齡，那么首先需要對源文件對應的 RDD 進行處理，也就是將它轉化成一個只包含年齡信息的 RDD，其次是計算元素個數即為總人數，然后是把所有年齡數加起來，最后平均年齡=總年齡/人數。 
對於第一步我們需要使用 map 算子把源文件對應的 RDD 映射成一個新的只包含年齡數據的 RDD，很顯然需要對在 map 算子的傳入函數中使用 split 方法，得到數組后只取第二個元素即為年齡信息；第二步計算數據元素總數需要對於第一步映射的結果 RDD 使用 count 算子；第三步則是使用 reduce 算子對只包含年齡信息的 RDD 的所有元素用加法求和；最后使用除法計算平均年齡即可。

以下實現對平均年齡的計算的代碼：

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 org.apache.spark.api.java.function.Function2;
import java.util.Arrays;

/**
 * Created by Administrator on 2017/11/13.
 */
public class AvgAgeCalculator {
    public static void main(String[] args){

        SparkConf sparkConf = new SparkConf().setAppName("AvgAgeCalculator").setMaster("local[3]");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        //讀取文件
        JavaRDD<String> dataFile = sc.textFile("F:\\DataFile.txt");
        //數據分片並取第二個數
        JavaRDD<String> ageData = dataFile.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterable<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" ")[1]);
            }
        });
        //求出所有年齡個數。
        long count = ageData.count();
        //轉換數據類型
        JavaRDD<Integer> ageDataInt = ageData.map(new Function<String, Integer>() {
            @Override
            public Integer call(String s) throws Exception {
                return Integer.parseInt(String.valueOf(s));
            }
        });
        //求出年齡的和
        Integer totalAge = ageDataInt.reduce(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer x, Integer y) throws Exception {
                return x+y;
            }
        });
        //平均值結果為double類型
        Double avgAge = totalAge.doubleValue()/count;
        /*System.out.println(ageData.collect());
        System.out.println(count);*/
        System.out.println("Total Age:" + totalAge + ";    Number of People:" + count );
        System.out.println("Average Age is " + avgAge);

    }
}

 

 

運行結果：

 

從結果可以看出，計算出所以年齡的總和，以及總人數，以及平均年齡值。看似簡單的例子，在對數組取值和數據類型轉換時候需要特別的注意。