使用Java开发本地测试的wordcount

ゝ一世哀愁。 2023-05-31 03:25 5阅读 0赞

package spark;
    
    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 org.apache.spark.api.java.function.VoidFunction;
    import scala.Tuple2;
    
    import java.util.Arrays;
    
    public class worldcount {
        public static void main(String[] args) {
            //编写Spark应用程序
    
            //第一步：创建sparkconf对象，设置spark应用的配置信息
            //使用setMaster() 可以设置spark应用程序要连接的spark集群的master节点的url
            //如果设置为local，则代表在本地运行
            SparkConf conf = new SparkConf()
                    .setAppName("worldcount")
                    .setMaster("local");
    
            //第二步：创建JavaSparkContent对象
            //在Spark中，SparkContext是Spark所有功能的一个入口，你无论是用java，scala，
            // 甚至是python编写，都必须要有一个SparkContext，它的主要作用是初始化spark应用程序一些核心组件
    
            //sparkContext是Spark应用中最重要的一个对象
            //编写不同类型的spark应用程序，使用的SparkContext是不同的，如果使用scala，使用的就是原生的SparkContext对象
            //如果使用Java就是JavaSparkContext对象，如果是Spark SQL程序，那就是SQLContext、HiveContext
            //如果是Spark Streaming程序，那么就是它独有的SparkContext...
    
            JavaSparkContext sc = new JavaSparkContext(conf);
    
            //第三步：要针对输入源（hdfs文件、本地文件，等等），创建一个初始化RDD
            //输入源中的数据会打散，分配到RDD的每个partition中，从而形成一个初始的分布式的数据集
            //这里是本地测试，所以针对本地文件
            //SparkContext中，用于根据文件类型的输入源创建RDD的方法，叫做textFile()方法
            //在Java中创建的普通RDD，都叫做JavaRDD
    
            JavaRDD<String> lines = sc.textFile("E:\\hadoop\\spark.txt");
    
            //第四步：对初始化RDD进行transformation操作，也就是一些计算操作
            //通常操作会通过创建function，并配合RDD的map、flatMap等算计来执行
            //function 如果比较简单，则创建Function的匿名内部类.
            // 如果比较复杂，则会单独创建一个 类，作为实现这个function接口的类
    
            //先将每一行拆分成单个的单词
            //FlatMapFunction，有两个泛型参数，分别代表了输入和输出类型，
            //flatMap算子的作用就是将RDD的一个元素，拆分成一个或多个元素
            final JavaRDD<String> worlds = lines.flatMap(new FlatMapFunction<String, String>() {
                @Override
                public Iterable<String> call(String lines) throws Exception {
                    return Arrays.asList(lines.split(" "));
                }
            });
            //第五步需要将一个单词映射为<单词，1>的这种格式
            //这样就可以对单词出现的次数累加
            //mapTopair,就是将每个元素，映射成为一个（v1,v2）这样的Tuple2类型的元素
            JavaPairRDD<String, Integer> pairs = worlds.mapToPair(new PairFunction<String, String, Integer>() {
                @Override
                public Tuple2<String, Integer> call(String word) throws Exception {
                    return new Tuple2<String, Integer>( word,1);
                }
            });
            //第六步：以单词作为key，统计每个单词出现的次数
            //这里使用reduceByKey这个算子，对每个key对应的value，都进行reduce操作
            //
            JavaPairRDD<String, Integer> wordcounts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
                @Override
                public Integer call(Integer v1, Integer v2) throws Exception {
                    return v1 + v2;
                }
            });
    
            //前面用的几个Spark算子操作，已经统计出了单词的次数
            //之前使用的flatMap、mapTopair、reduceByKey都是transformation操作
            //光有transformation是不会执行的，还需要action操作，
            //比如说，foreach来触发程序的执行
    
            wordcounts.foreach(new VoidFunction<Tuple2<String, Integer>>() {
                @Override
                public void call(Tuple2<String, Integer> t) throws Exception {
                    System.out.println(t._1 + ":" + t._2);
                }
            });
    
            //关闭JavaSparkContext链接
            sc.close();
        }
    }