需求

现在想要通过socket发送数据, 然后Spark Streaming接收数据并且统计socket发送的每个单词的个数.

1. 架构图

2. 实现流程

安装并启动生产者
首先在linux服务器上用yum安装netcat工具,netcat工具用了发送socket套接字,netcat的命令是nc, 它使用来设置路由器的,我们可以利用它向某个端口发送数据.
linux安装netcat命令如下:

yum intstall -y nc

通过netcat工具向指定的端口发送数据

nc -lk 9999

IDEA编写Spark Streaming代码

package cn.acec.sparkStreamingtestimport org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
/*** sparkStreming流式处理接受socket数据，实现单词统计*/
object SparkStreamingNC{def main(args: Array[String]): Unit = {//配置sparkConf参数val sparkConf: SparkConf = new   SparkConf().setAppName("SparkStreamingTCP").setMaster("local[2]")//构建sparkContext对象val sc: SparkContext = new SparkContext(sparkConf)//设置日志输出级别sc.setLogLevel("WARN")//构建StreamingContext对象，每个批处理的时间间隔val scc: StreamingContext = new StreamingContext(sc,Seconds(5))//注册一个监听的IP地址和端口  用来收集数据val lines: ReceiverInputDStream[String] = scc.socketTextStream("192.168.200.160",9999)//切分每一行记录val words: DStream[String] = lines.flatMap(_.split(" "))//每个单词记为1val wordAndOne: DStream[(String, Int)] = words.map((_,1))//分组聚合val result: DStream[(String, Int)] = wordAndOne.reduceByKey(_+_)//打印数据result.print()scc.start()scc.awaitTermination()}
}

注意:
由于使用的是本地模式local[2], 所以可以直接在本地运行程序
要指定并行度, 如在本地运行设置setMaster(“local[2]”)，相当于启动两个线程，一个给receiver，一个给computer。如果是在集群中运行，必须要求集群中可用core数大于1。

3. 执行查看结果

先执行nc -lk 9999
然后再在IDEA中启动代码
不断的在1.中输入不同的单词，观察IDEA控制台输出

4. 结果

现象：sparkStreaming每隔5s计算一次当前5s内的数据，然后将每个批次的数据输出。

DStream实战之Spark Streaming接收socket数据实现WordCount 31相关推荐

DStream实战之Spark Streaming整合fulme实战, Flume向Spark Streaming中push推数据 36
前言本文所需要的安装包&Flume配置文件,博主都已上传,链接为本文涉及安装包&Flume配置文件本文涉及的安装包&Flume配置文件,请自行下载~ flume作为日志实时采 ...
spark streaming 接收 kafka 数据java代码WordCount示例
1. 首先启动zookeeper 2. 启动kafka 3. 核心代码生产者生产消息的java代码,生成要统计的单词 package streaming;import java.util.Prope ...
sparkstreaming监听hdfs目录_大数据系列之Spark Streaming接入Kafka数据
Spark Streaming官方提供Receiver-based和Direct Approach两种方法接入Kafka数据,本文简单介绍两种方式的pyspark实现. 1.Spark Streami ...
Spark Streaming揭秘 Day16 数据清理机制
Spark Streaming揭秘 Day16 数据清理机制今天主要来讲下Spark的数据清理机制,我们都知道,Spark是运行在jvm上的,虽然jvm本身就有对象的自动回收工作,但是,如果自己不进 ...
Spark Streaming通过Socket检测空气质量
作业描述: 针对当前空⽓质量监测数据,环保部门需要根据监测数据实时发布预警信息,需要我们在实时到达的六种污染物监测数据中,根据每⼀种数据的监测值进⾏报警检测. 输⼊数据:"空⽓质量监控数据& ...
Spark Streaming处理Socket流简单实例
在本文中我将在IDEA工具中开发一个SparkStream程序用于监听本机9999端口所接收的数据首先,我们将Spark Streaming类的名称以及从StreamingContext进行的一些隐 ...
Spark Streaming读取Kafka数据的两种方式
Kafka在0.8和0.10之间引入了一种新的消费者API,因此,Spark Streaming与Kafka集成,有两种包可以选择: spark-streaming-kafka-0-8与spark-s ...
Spark大数据分析与实战：Spark Streaming编程初级实践
Spark Streaming编程初级实践一.安装Hadoop和Spark 具体的安装过程在我以前的博客里面有,大家可以通过以下链接进入操作: Hadoop的安装:https://blog.csdn ...
Flume+Kafka+Spark Streaming实现大数据实时流式数据采集
近年来,随着企业信息化建设的飞速发展,大数据应用的问题越来越备受关注.很多企业投入大量的人力.物力和财力建设企业大数据平台,平台建设工作涵盖数据采集.数据处理.数据存储.数据服务.数据展示以及数据质量 ...

DStream实战之Spark Streaming接收socket数据实现WordCount 31

需求

1. 架构图

2. 实现流程

3. 执行查看结果

4. 结果

DStream实战之Spark Streaming接收socket数据实现WordCount 31相关推荐

最新文章

热门文章