Spark1.4从HDFS读取文件运行Java语言WordCounts并将结果保存至HDFS

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.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;public final class JavaWordCount {
  private static final Pattern SPACE = Pattern.compile(" ");  public static void main(String[] args) throws Exception {    if (args.length < 1) {
      System.err.println("Usage: JavaWordCount <file>");
      System.exit(1);
    }    //创建SparkConf，包含application的相关信息
    SparkConf sparkConf = new SparkConf().setAppName("JavaWordCount");
    //创建一个JavaSparkContext对象
    JavaSparkContext ctx = new JavaSparkContext(sparkConf);
    //textFile()方法可将本地文件或HDFS文件转换成RDD，读取本地文件需要各节点上都存在，或者通过网络共享该文件
    //读取一行
    JavaRDD<String> lines = ctx.textFile(args[0], 1);
    //flatMap与map的区别是，对每个输入，flatMap会生成一个或多个的输出，而map只是生成单一的输出
    //用空格分割各个单词,输入一行,输出多个对象,所以用flatMap
    JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
      @Override
      public Iterable<String> call(String s) {
        return Arrays.asList(SPACE.split(s));
      }
    });
    //对每个单词生成key-value对,PairFunction<T,K,V>
    //表示输入类型为T,生成的key-value对中的key类型为k,value类型为v,对本例,T=String, K=String, V=Integer(计数)
    //重写scala的Tupple2方法
    JavaPairRDD<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() {
      @Override
      //scala.Tuple2<K,V> call(T t)
      //Tuple2为scala中的一个对象,call方法的输入参数为T,即输入一个单词s,新的Tuple2对象的key为这个单词,计数为1
      public Tuple2<String, Integer> call(String s) {
        return new Tuple2<String, Integer>(s, 1);
      }
    });
    //调用reduceByKey方法,按key值进行reduce
    //调用Function2对象,Function2<T1,T2,R>
    //输入两个参数,T1,T2,返回R
    //若ones有<"one", 1>, <"one", 1>,会根据"one"将相同的pair单词个数进行统计,输入为Integer,输出也为Integer
    //输出<"one", 2>
    JavaPairRDD<String, Integer> counts = ones.reduceByKey(new Function2<Integer, Integer, Integer>() {
      @Override
      public Integer call(Integer i1, Integer i2) {
        return i1 + i2;
      }
    });
    //将结果保存到HDFS中
    counts.saveAsTextFile(args[1]);
    //collect返回一个包含RDD内所有元素的Array
    List<Tuple2<String, Integer>> output = counts.collect();
    for (Tuple2<?, ?> tuple : output) {
      System.out.println(tuple._1() + ": " + tuple._2());
    }
    ctx.stop();
  }
}