學習隨筆--Spark java開發入門

package stuSpark.com;

import scala.Tuple2;

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.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.storage.StorageLevel;

import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.regex.Pattern;

public final class JavaWordCount {
    private static final Pattern SPACE = Pattern.compile(" ");
    //pattern 對象是一個正則表達式的編譯表示
    //compile()方法表示編譯此正則表達式regExp，返回regExp被編譯后的pattern

    public static void main(String[] args) throws Exception {

        // file 代表本地路徑，反之代表hdfs路徑
        String filePath = "file:\\E:\\test.txt";

        SparkConf sparkConf = new SparkConf().setAppName("JavaWordCount")
                .setMaster("local[2]");
        //設置該程序名稱   設置本地模式
        JavaSparkContext ctx = new JavaSparkContext(sparkConf);
        //創建JavaSparkContext對象實例sc
        
        JavaRDD<String> lines = ctx.textFile(filePath, 1);
        //直接從集合轉化 sc.parallelize(List(1,2,3,4,5,6,7,8,9,10))
        //從HDFS文件轉化 sc.textFile("hdfs://")
        //從本地文件轉化 sc.textFile("file:/")
        
        lines.cache();
        lines.persist(StorageLevel.MEMORY_ONLY());
        //持久化RDD
        /*
         * cache()方法表示：使用非序列化的方式將RDD的數據全部嘗試持久化到內存中，
         * cache是一個transformtion，是lazy的，必須通過一個action觸發，
         * 才能真正的將該RDD cache到內存中。
         * 
         * persist()方法表示：手動選擇持久化級別，並使用指定的方式進行持久化
         * DISK_ONLY：磁盤
        DISK_ONLY_2：磁盤；雙副本
        MEMORY_ONLY： 內存；反序列化；把RDD作為反序列化的方式存儲，假如RDD的內容存不下，剩余的分區在以后需要時會重新計算，不會刷到磁盤上。
        MEMORY_ONLY_2：內存；反序列化；雙副本
        MEMORY_ONLY_SER：內存；序列化；這種序列化方式，每一個partition以字節數據存儲，好處是能帶來更好的空間存儲，但CPU耗費高
        MEMORY_ONLY_SER_2 : 內存；序列化；雙副本
        MEMORY_AND_DISK：內存 + 磁盤；反序列化；雙副本；RDD以反序列化的方式存內存，假如RDD的內容存不下，剩余的會存到磁盤
        MEMORY_AND_DISK_2 : 內存 + 磁盤；反序列化；雙副本
        MEMORY_AND_DISK_SER：內存 + 磁盤；序列化  
        MEMORY_AND_DISK_SER_2：內存 + 磁盤；序列化；雙副本
         * */
        

        // 並行化集合
        //並行數組中一個很重要的參數是partitions，它來描述數組被切割的數據集數量。Spark會在每一個partitions上運行任務
        List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
        //數組連接list，當更新其中之一時，另一個自動更新
        JavaRDD<Integer> distData = ctx.parallelize(data);
        //分發本地Scala集合以形成RDD
        //初始化一個已經存在的集合
        
        //filter()參數是函數，函數會過濾掉不符合條件的元素，返回值是新的RDD
        lines.filter(new Function<String,Boolean>(){

            public Boolean call(String arg0) throws Exception {
                // TODO Auto-generated method stub
                return null;
            }
            
        });
        //map參數是函數，函數應用於RDD每一個元素，返回值是新的RDD
        JavaRDD<Integer> lineLengths = lines
                .map(new Function<String, Integer>() {
                    public Integer call(String s) {
                        return s.length();
                    }
                });
        //reduce聚集，但是傳入的函數是兩個參數輸入返回一個值，這個函數必須是滿足交換律和結合律
        int totalLength = lineLengths
                .reduce(new Function2<Integer, Integer, Integer>() {
                    public Integer call(Integer a, Integer b) {
                        return a + b;
                    }
                });
        //flatMap和map差不多，但是flatMap生成的是多個結果
        JavaRDD<String> words = lines
                .flatMap(new FlatMapFunction<String, String>() {
                    //Iterable迭代的
                    public Iterable<String> call(String s) {
                        return Arrays.asList(SPACE.split(s));
                    }
                });
        //maptopair 將集合數據存為key value
        JavaPairRDD<String, Integer> ones = words
                .mapToPair(new PairFunction<String, String, Integer>() {
                    public Tuple2<String, Integer> call(String s) {
                        return new Tuple2<String, Integer>(s, 1);
                    }
                });
        //reduceBykey 根據key聚集，對value進行操作
        JavaPairRDD<String, Integer> counts = ones
                .reduceByKey(new Function2<Integer, Integer, Integer>() {
                    public Integer call(Integer i1, Integer i2) {
                        return i1 + i2;
                    }
                });
        //collect封裝返回一個數組
        List<Tuple2<String, Integer>> output = counts.collect();
        for (Tuple2<?, ?> tuple : output) {
            System.out.println(tuple._1() + ": " + tuple._2());
        }
        ctx.stop();
    }
}

簡單的java項目開發，所需Jar包見百度網盤

鏈接：https://pan.baidu.com/s/1jqWwBBNIm1kbQoFSCppEZQ 密碼：y4xr