学习随笔--Spark java开发入门

Posted 2020-12-31 沙沙沙啊啊皮

  1 package stuSpark.com;
  2 
  3 import scala.Tuple2;
  4 
  5 import org.apache.spark.SparkConf;
  6 import org.apache.spark.api.java.JavaPairRDD;
  7 import org.apache.spark.api.java.JavaRDD;
  8 import org.apache.spark.api.java.JavaSparkContext;
  9 import org.apache.spark.api.java.function.FlatMapFunction;
 10 import org.apache.spark.api.java.function.Function;
 11 import org.apache.spark.api.java.function.Function2;
 12 import org.apache.spark.api.java.function.PairFunction;
 13 import org.apache.spark.storage.StorageLevel;
 14 
 15 import java.util.Arrays;
 16 import java.util.Iterator;
 17 import java.util.List;
 18 import java.util.regex.Pattern;
 19 
 20 public final class JavaWordCount {
 21     private static final Pattern SPACE = Pattern.compile(" ");
 22     //pattern 对象是一个正则表达式的编译表示
 23     //compile()方法表示编译此正则表达式regExp，返回regExp被编译后的pattern
 24 
 25     public static void main(String[] args) throws Exception {
 26 
 27         // file 代表本地路径，反之代表hdfs路径
 28         String filePath = "file:\E:\test.txt";
 29 
 30         SparkConf sparkConf = new SparkConf().setAppName("JavaWordCount")
 31                 .setMaster("local[2]");
 32         //设置该程序名称   设置本地模式
 33         JavaSparkContext ctx = new JavaSparkContext(sparkConf);
 34         //创建JavaSparkContext对象实例sc
 35         
 36         JavaRDD<String> lines = ctx.textFile(filePath, 1);
 37         //直接从集合转化 sc.parallelize(List(1,2,3,4,5,6,7,8,9,10))
 38         //从HDFS文件转化 sc.textFile("hdfs://")
 39         //从本地文件转化 sc.textFile("file:/")
 40         
 41         lines.cache();
 42         lines.persist(StorageLevel.MEMORY_ONLY());
 43         //持久化RDD
 44         /*
 45          * cache()方法表示：使用非序列化的方式将RDD的数据全部尝试持久化到内存中，
 46          * cache是一个transformtion，是lazy的，必须通过一个action触发，
 47          * 才能真正的将该RDD cache到内存中。
 48          * 
 49          * persist()方法表示：手动选择持久化级别，并使用指定的方式进行持久化
 50          * DISK_ONLY：磁盘
 51         DISK_ONLY_2：磁盘；双副本
 52         MEMORY_ONLY： 内存；反序列化；把RDD作为反序列化的方式存储，假如RDD的内容存不下，剩余的分区在以后需要时会重新计算，不会刷到磁盘上。
 53         MEMORY_ONLY_2：内存；反序列化；双副本
 54         MEMORY_ONLY_SER：内存；序列化；这种序列化方式，每一个partition以字节数据存储，好处是能带来更好的空间存储，但CPU耗费高
 55         MEMORY_ONLY_SER_2 : 内存；序列化；双副本
 56         MEMORY_AND_DISK：内存 + 磁盘；反序列化；双副本；RDD以反序列化的方式存内存，假如RDD的内容存不下，剩余的会存到磁盘
 57         MEMORY_AND_DISK_2 : 内存 + 磁盘；反序列化；双副本
 58         MEMORY_AND_DISK_SER：内存 + 磁盘；序列化  
 59         MEMORY_AND_DISK_SER_2：内存 + 磁盘；序列化；双副本
 60          * */
 61         
 62 
 63         // 并行化集合
 64         //并行数组中一个很重要的参数是partitions，它来描述数组被切割的数据集数量。Spark会在每一个partitions上运行任务
 65         List<Integer> data = Arrays.asList(1, 2, 3, 4, 5);
 66         //数组连接list，当更新其中之一时，另一个自动更新
 67         JavaRDD<Integer> distData = ctx.parallelize(data);
 68         //分发本地Scala集合以形成RDD
 69         //初始化一个已经存在的集合
 70         
 71         //filter()参数是函数，函数会过滤掉不符合条件的元素，返回值是新的RDD
 72         lines.filter(new Function<String,Boolean>(){
 73 
 74             public Boolean call(String arg0) throws Exception {
 75                 // TODO Auto-generated method stub
 76                 return null;
 77             }
 78             
 79         });
 80         //map参数是函数，函数应用于RDD每一个元素，返回值是新的RDD
 81         JavaRDD<Integer> lineLengths = lines
 82                 .map(new Function<String, Integer>() {
 83                     public Integer call(String s) {
 84                         return s.length();
 85                     }
 86                 });
 87         //reduce聚集，但是传入的函数是两个参数输入返回一个值，这个函数必须是满足交换律和结合律
 88         int totalLength = lineLengths
 89                 .reduce(new Function2<Integer, Integer, Integer>() {
 90                     public Integer call(Integer a, Integer b) {
 91                         return a + b;
 92                     }
 93                 });
 94         //flatMap和map差不多，但是flatMap生成的是多个结果
 95         JavaRDD<String> words = lines
 96                 .flatMap(new FlatMapFunction<String, String>() {
 97                     //Iterable迭代的
 98                     public Iterable<String> call(String s) {
 99                         return Arrays.asList(SPACE.split(s));
100                     }
101                 });
102         //maptopair 将集合数据存为key value
103         JavaPairRDD<String, Integer> ones = words
104                 .mapToPair(new PairFunction<String, String, Integer>() {
105                     public Tuple2<String, Integer> call(String s) {
106                         return new Tuple2<String, Integer>(s, 1);
107                     }
108                 });
109         //reduceBykey 根据key聚集，对value进行操作
110         JavaPairRDD<String, Integer> counts = ones
111                 .reduceByKey(new Function2<Integer, Integer, Integer>() {
112                     public Integer call(Integer i1, Integer i2) {
113                         return i1 + i2;
114                     }
115                 });
116         //collect封装返回一个数组
117         List<Tuple2<String, Integer>> output = counts.collect();
118         for (Tuple2<?, ?> tuple : output) {
119             System.out.println(tuple._1() + ": " + tuple._2());
120         }
121         ctx.stop();
122     }
123 }

简单的java项目开发，所需Jar包见百度网盘

链接：https://pan.baidu.com/s/1jqWwBBNIm1kbQoFSCppEZQ 密码：y4xr