(1)StreamingContext
与spark core的编程类似,在编写SparkStreaming的程序时,也需要一个通用的编程入口—-StreamingContext。
StreamingContext的创建:
object StreamingContextTest {
def main(args: Array[String]): Unit = {
val sparkConf = new SparkConf().setAppName("SCTest").setMaster("local[4]")
val streamingContext = new StreamingContext(sparkConf, Seconds(2))
}
}
注意:
如果在计算的时候,指定–master时 使用的是local 并且只指定了一个线程,那么只有receiver线程工作,计算的线程不会工作,所以在指定线程数的时候,最少指定2个。
(2)通过输入源创建InputDStream:
在构建好StreamingContext之后,首先我们要读取数据源的数据进行实时处理:
InputDStreams指的是从数据流的源头接收的输入数据流,每个 InputDStream 都关联一个 Receiver 对象,该 Receiver 对象接收数据源传来的数据并将其保存在内存中以便后期 Spark 处理。
Spark Streaming 提供两种原生支持的流数据源和自定义的数据源:
– 直接通过 StreamingContext API 创建,例如文件系统(本地文件系统及分布式文件系统)、 Socket 连接及 Akka 的 Actor。
– Kafka, Flume, Kinesis, Twitter 等,需要借助外部工具类,在运行时需要外部依赖
-Spark Streaming 还支持用户自定义数据源,它需要用户定义 receiver
注意:
- 在本地运行 Spark Streaming 时,master URL 不能使用”local”或”local[1] ”,因为当 Input DStream 与 Receiver(如 sockets, Kafka, Flume 等)关联时,Receiver 自身就需要一个线程 来运行,此时便没有线程去处理接收到的数据。因此,在本地运行 SparkStreaming 程序时,要使用”local[n]”作为 master URL,n 要大于 receiver 的数量。
- 在集群上运行 Spark Streaming 时,分配给 Spark Streaming 程序的 CPU 核数也必须大于 receiver 的数量,否则系统将只接受数据,无法处理数据。
(3)对DStream进行transformation 和 output 操作,这样操作构成了后期流式计算的逻辑
(4)通过streamingContext.start()方法启动接收和处理数据的流程
(5)使用streamingContext.awaitTermination()方法等待程序结束(手动停止或出错停止)
(6)调用streamingContext.stop()方法来结束程序的运行。
在编写sparkStreaming时的注意点:
- streamingContext启动后,增加新的操作将不起作用,一定要在启动之前定义好逻辑,也就是说在调用start方法之后,在对sparkStreaming程序进行逻辑操作是不被允许的
- StreamingContext 是单例对象停止后,不能重新启动,除非重新启动任务,重新执行计算
- 在单个jvm中,一段时间内不能出现两个active状态的StreamingContext
- 当在调用 StreamingContext 的 stop 方法时,默认情况下 SparkContext 也将被 stop 掉, 如果希望 StreamingContext 关闭时,能够保留 SparkContext,则需要在 stop 方法中传入参 数 stop SparkContext=false
- 一个 SparkContext 可以用来创建多个 StreamingContext,只要前一个 StreamingContext 已经停止了。
原创文章,作者:carmelaweatherly,如若转载,请注明出处:https://blog.ytso.com/192307.html