Spark Dstream
2.Dstream 入门
Spark Streaming的入口
StreamingContext
val conf = new SparkConf().setAppName(appName).setMaster(master) val ssc = new StreamingContext(conf, Seconds(1)) // 可以通过ssc.sparkContext 来访问SparkContext
// 或者通过已经存在的SparkContext来创建StreamingContext val sc = ... val ssc = new StreamingContext(sc, Seconds(1))
初始化完Context之后:
1)定义消息输入源来创建DStreams.
2)定义DStreams的转化操作和输出操作。
3)通过 streamingContext.start()来启动消息采集和处理.
4)等待程序终止,可以通过
streamingContext.awaitTermination()来设置
5)通过streamingContext.stop()来手动终止处理程序。
2.1 WordCount 案例实操
1.需求:使用 netcat 工具向 9999 端口不断的发送数据,通过 SparkStreaming 读取端口数据并统
计不同单词出现的次数
2.添加依赖
<dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming_2.11</artifactId> <version>2.1.1</version> </dependency>
3.编写代码
package com.lxl
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream} import org.apache.spark.streaming.{Seconds, StreamingContext} import org.apache.spark.SparkConf
object StreamWordCount { def main(args: Array[String]): Unit = {
//1.初始化 Spark 配置信息 val sparkConf = new SparkConf().setMaster("local[*]").setAppName("StreamWordCount")
//2.初始化 SparkStreamingContext val ssc = new StreamingContext(sparkConf, Seconds(5))
//3.通过监控端口创建 DStream,读进来的数据为一行行 val lineStreams = ssc.socketTextStream("hadoop102", 9999)
//将每一行数据做切分,形成一个个单词 val wordStreams = lineStreams.flatMap(_.split(" "))
//将单词映射成元组(word,1) val wordAndOneStreams = wordStreams.map((_, 1))
//将相同的单词次数做统计 val wordAndCountStreams = wordAndOneStreams.reduceByKey(_+_)
//打印 wordAndCountStreams.print()
//启动 SparkStreamingContext ssc.start() ssc.awaitTermination() } }
4.启动程序并通过 NetCat 发送数据:
[[email protected] spark]$ nc -lk 9999 hello atguigu
注意:如果程序运行时,log 日志太多,可以将 spark conf 目录下的 log4j 文件里面的日志级别改
成 WARN。
2.2 WordCount 解析
Discretized Stream 是 Spark Streaming 的基础抽象,代表持续性的数据流和经过各种 Spark 原
语操作后的结果数据流。在内部实现上,DStream 是一系列连续的 RDD 来表示。每个 RDD 含有
一段时间间隔内的数据,如下图:
对数据的操作也是按照 RDD 为单位来进行的
计算过程由 Spark engine 来完成