java/scala如何实现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        * 带有两个参数        *           第一个参数：SparkConf 配置        *           第二个参数: 每次收取的数据流的时间间隔  作为一个批次进行处理        */        JavaStreamingContext jsc=new JavaStreamingContext(sparkConf, Durations.seconds(2));        /*指定从socket数据源接收数据        * 指定两个参数 1：主机名   2：端口        * window发送命令 nc -l  -p  9999        * linux 发送命令  nc -lk 9999*/
        JavaReceiverInputDStream<String> sourceDStream = jsc.socketTextStream("localhost", 9999);
        /*接下来就是对每个批次就行处理  这里是每2秒钟一个批次 这样一行行的数据流都被拆分为一个个的单词流*/        JavaDStream<String> wordDStream = sourceDStream.flatMap(new FlatMapFunction<String, String>() {            @Override            public Iterator<String> call(String line) throws Exception {                return Arrays.asList(line.split(" ")).iterator();            }        });        /*转换成  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;            }        });
        /*打印结果*/        wordCountResult.print();
        /*jsc这里必须要调用start()函数application才会启动执行，接收数据*/        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 = {
    /*构建SparkConf配置*/    val sparkConf  =new SparkConf().setMaster("local[*]").setAppName("StreamingWordCountScala")    val ssc =new StreamingContext(sparkConf,Seconds(2))
    /*指定socket数据源*/    val sourceDStream=ssc.socketTextStream("localhost",9999)
    val  wordDStream=sourceDStream.flatMap(x=>x.split(" "))
    val  wordPairDStream=wordDStream.map(x=>(x,1))    val  wordCountResult=wordPairDStream.reduceByKey(_+_)
    /*打印结果*/    wordCountResult.print()    /*启动*/    ssc.start()    ssc.awaitTermination()    /*停止*/    ssc.stop()  }}