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