本篇内容介绍了“java/scala如何实现WordCount程序”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！
公司主营业务：成都网站设计、成都做网站、移动网站开发等业务。帮助企业客户真正实现互联网宣传，提高企业的竞争能力。创新互联是一支青春激扬、勤奋敬业、活力青春激扬、勤奋敬业、活力澎湃、和谐高效的团队。公司秉承以“开放、自由、严谨、自律”为核心的企业文化，感谢他们对我们的高要求，感谢他们从不同领域给我们带来的挑战，让我们激情的团队有机会用头脑与智慧不断的给客户带来惊喜。创新互联推出鄂尔多斯免费做网站回馈大家。
程序从windows一个socket端的9999端口接收以换行符分隔的多行文本，每两秒一个时间窗口，打印字数统计。
Socket数据发送命令
window发送命令 nc -l -p 9999
linux 发送命令 nc -lk 9999
Java版本：
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 sourceDStream = jsc.socketTextStream("localhost", 9999);
/*接下来就是对每个批次就行处理 这里是每2秒钟一个批次 这样一行行的数据流都被拆分为一个个的单词流*/ JavaDStream wordDStream = sourceDStream.flatMap(new FlatMapFunction() { @Override public Iterator call(String line) throws Exception { return Arrays.asList(line.split(" ")).iterator(); } }); /*转换成 hello 1 * world 1 * a 1 * b 1 格式*/ JavaPairDStream wordPairDStream = wordDStream.mapToPair(new PairFunction() { @Override public Tuple2 call(String word) throws Exception { return new Tuple2<>(word, 1); } }); JavaPairDStream wordCountResult = wordPairDStream.reduceByKey(new Function2() { @Override public Integer call(Integer v1, Integer v2) throws Exception { return v1+v2; } });
/*打印结果*/ wordCountResult.print();
/*jsc这里必须要调用start()函数application才会启动执行，接收数据*/ jsc.start(); jsc.awaitTermination(); /*停止*/ jsc.stop(); }}
Scala版本：
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() }}
“java/scala如何实现WordCount程序”的内容就介绍到这里了，感谢大家的阅读。如果想了解更多行业相关的知识可以关注创新互联网站，小编将为大家输出更多高质量的实用文章！
文章标题：java/scala如何实现WordCount程序
当前地址：
http://cxhlcq.com/article/jhdhoe.html
重庆分公司
重庆分公司
