学习随笔--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