java/scala如何實(shí)現(xiàn)WordCount程序

package com.unicom.ljs.spark220.study.streaming;
import org.apache.spark.SparkConf;import org.apache.spark.api.java.JavaPairRDD;import org.apache.spark.api.java.function.*;import org.apache.spark.streaming.Durations;import org.apache.spark.streaming.Time;import org.apache.spark.streaming.api.java.JavaDStream;import org.apache.spark.streaming.api.java.JavaPairDStream;import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;import org.apache.spark.streaming.api.java.JavaStreamingContext;import scala.Tuple2;
import java.util.Arrays;import java.util.Iterator;
/** * @author: Created By lujisen * @company ChinaUnicom Software JiNan * @date: 2020-01-30 22:21 * @version: v1.0 * @description: com.unicom.ljs.spark220.study.streaming */public class StreamingWordCount {    public static void main(String[] args) throws InterruptedException {
        SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("StreamingWordCount");        /*這里JavaStreamingContext類似sparkCore的SparkContext        * 帶有兩個(gè)參數(shù)        *           第一個(gè)參數(shù)：SparkConf 配置        *           第二個(gè)參數(shù): 每次收取的數(shù)據(jù)流的時(shí)間間隔  作為一個(gè)批次進(jìn)行處理        */        JavaStreamingContext jsc=new JavaStreamingContext(sparkConf, Durations.seconds(2));        /*指定從socket數(shù)據(jù)源接收數(shù)據(jù)        * 指定兩個(gè)參數(shù) 1：主機(jī)名   2：端口        * window發(fā)送命令 nc -l  -p  9999        * linux 發(fā)送命令  nc -lk 9999*/
        JavaReceiverInputDStream<String> sourceDStream = jsc.socketTextStream("localhost", 9999);
        /*接下來(lái)就是對(duì)每個(gè)批次就行處理  這里是每2秒鐘一個(gè)批次 這樣一行行的數(shù)據(jù)流都被拆分為一個(gè)個(gè)的單詞流*/        JavaDStream<String> wordDStream = sourceDStream.flatMap(new FlatMapFunction<String, String>() {            @Override            public Iterator<String> call(String line) throws Exception {                return Arrays.asList(line.split(" ")).iterator();            }        });        /*轉(zhuǎn)換成  hello  1        *         world  1        *         a      1        *         b      1 格式*/        JavaPairDStream<String, Integer> wordPairDStream = wordDStream.mapToPair(new PairFunction<String, String, Integer>() {            @Override            public Tuple2<String, Integer> call(String word) throws Exception {                return new Tuple2<>(word, 1);            }        });        JavaPairDStream<String, Integer>  wordCountResult = wordPairDStream.reduceByKey(new Function2<Integer, Integer, Integer>() {            @Override            public Integer call(Integer v1, Integer v2) throws Exception {                return v1+v2;            }        });
        /*打印結(jié)果*/        wordCountResult.print();
        /*jsc這里必須要調(diào)用start()函數(shù)application才會(huì)啟動(dòng)執(zhí)行，接收數(shù)據(jù)*/        jsc.start();        jsc.awaitTermination();        /*停止*/        jsc.stop();    }}

package com.unicom.ljs.study.streaming
import org.apache.spark.SparkConfimport org.apache.spark.streaming.StreamingContextimport org.apache.spark.streaming.Seconds
/**  * @author: Created By lujisen  * @company ChinaUnicom Software JiNan  * @date: 2020-01-31 08:59  * @version: v1.0  * @description: com.unicom.ljs.study.streaming  */object StreamingWordCount {  def main(args: Array[String]): Unit = {
    /*構(gòu)建SparkConf配置*/    val sparkConf  =new SparkConf().setMaster("local[*]").setAppName("StreamingWordCountScala")    val ssc =new StreamingContext(sparkConf,Seconds(2))
    /*指定socket數(shù)據(jù)源*/    val sourceDStream=ssc.socketTextStream("localhost",9999)
    val  wordDStream=sourceDStream.flatMap(x=>x.split(" "))
    val  wordPairDStream=wordDStream.map(x=>(x,1))    val  wordCountResult=wordPairDStream.reduceByKey(_+_)
    /*打印結(jié)果*/    wordCountResult.print()    /*啟動(dòng)*/    ssc.start()    ssc.awaitTermination()    /*停止*/    ssc.stop()  }}