1.2 SparkStreaming实时处理入门
1.2.1 工程创建
导入maven依赖
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.12</artifactId>
<version>3.1.2</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
<version>3.1.2</version>
</dependency>1.2.2 入口类StreamingContext
SparkStreaming中的入口类，称之为StreamingContext,但是底层还是得需要依赖SparkContext。
object SparkStreamingWordCountOps {
def main(args: Array[String]): Unit = {
/*
StreamingContext的初始化，需要至少两个参数，SparkConf和BatchDuration
SparkConf不用多说
batchDuration：提交两次作业之间的时间间隔，每次会提交一个DStream，将数据转化batch--->RDD
所以说：sparkStreaming的计算，就是每隔多长时间计算一次数据
*/
val conf = new SparkConf()
.setAppName("SparkStreamingWordCount")
.setMaster("local[*]")
val duration = Seconds(2)
val ssc = new StreamingContext(conf, duration) //批次
//业务
//为了执行的流式计算，必须要调用start来启动
ssc.start()
//为了不至于start启动程序结束，必须要调用awaitTermination方法等待程序业务完成之后调用stop方法结束程序，或者异常
ssc.awaitTermination()
}
}
1.2.3 业务编写
SparkStreaming是一个流式计算的计算引擎，那么 就模拟一个对流式数据进行单词统计
代码实现
package com.qianfeng.sparkstreaming
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Durations, StreamingContext}
/**
* sparkStreaming的流程序
*/
object Demo01_SparkStreaming_WC {
def main(args: Array[String]): Unit = {
//1、获取streamingcontext
val conf = new SparkConf()
.setAppName("streaming wc")
.setMaster("local[*]")
val sc = new StreamingContext(conf, Durations.seconds(2)) //微批次微2s
//2、初始化数据
val ds = sc.socketTextStream("qianfeng01", 6666)
//3、对数据进行操作
val sumDS = ds.flatMap(_.split(" "))
#判断H开头 5位
.filter(x=>x.startsWith("H") && x.length == 5)
.map((_, 1))
.reduceByKey(_ + _)
//4、对数据做输出
sumDS.print()
//5、开启sc
sc.start()
//6、等待结束 --- 实时不能停止
sc.awaitTermination()
}
}
使用netcat进行测试（如果没有请先安装，有则忽略如下）
需要在任意一台节点上安装工具：
[root@qianfeng01 home]# yum install -y nc
启动监听端口：
[root@qianfeng01 home]# nc -lk 6666 hello nihao nihao hello hi hello nihao
