Spark Streaming学习笔记

时间:2022-08-20 20:49:13

Spark Streaming 编程学习笔记

简介

spark stream 基于spark 核心API扩展而来,提供了一种具有规模可伸缩、高吞吐、错误恢复的处理实时数据流的流式处理方法。数据来源可以从本地文件,hadoop等接受,可以使用该机函数类似map、reduce、window等函数进行处理。数据结果同样可以进行多种形式的存储。

Spark Streaming学习笔记
在spark stream内部,将接收到的实时数据进行切分成不同的数据段,然后处理引擎对一个个的数据快进行处理然后输出相应的结果。

这个地方有点疑问:如果原始数据被切分开,比如说,原始数据是“Hello, my boys", 但是在spark数据接收端对数据块进行切分时,从中间切分开来使"Hello"和", my boys"成为了两个数据块,这内部怎么保证切分永远是合理的??

Spark Streaming学习笔记
上面的流程图中的batches,spark将其抽象成一种数据结构叫做 DStream(discretized stream),Dstream既可以有输入的实时数据流创建,也可以由其他的函数对已有的DStream操作后得到。DStream在spark内部就是一系列的RDD的集合。

一个简单的例子

以下的例子是一个简单的wordCount的程序,改程序从TCP端口接收数据,然后统计接受到数据的文字的个数。


import org.apache.spark.
import org.apache.spark.streaming.

import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3

// Create a local StreamingContext with two working thread and batch interval of 1 second.
// The master requires 2 cores to prevent from a starvation scenario.
//采用的核数至少要大于1,否则spark没有办法去一边接受数据一边处理数据,会报错block input --- instead 0 peers,表示spark没有足够资源
val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
val ssc = new StreamingContext(conf, Seconds(1)) //对于使用交互模式的,由于只能有一个SparkContext, 此时可以采用 val ssc = new StreamingContext(sc, Seconds(1))

// Create a DStream that will connect to hostname:port, like localhost:9999
val lines = ssc.socketTextStream("localhost", 9999)

// Split each line into words
val words = lines.flatMap(_.split(" "))

import org.apache.spark.streaming.StreamingContext._ // not necessary since Spark 1.3
// Count each word in each batch
val pairs = words.map(word => (word, 1))
val wordCounts = pairs.reduceByKey(_ + _)

// Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.print()
ssc.start() // Start the computation
ssc.awaitTermination() // Wait for the computation to terminate

基础概念

初始化SparkStreaming

SparkStreaming 对象是整个spark实时数据处理的入口点,初始化的方法可以使用SparkConf作为参数进行创建:


val conf = new SparkConf().setAppName(appName).setMaster(master)
val ssc = new StreamingContext(conf, Seconds(1))

需要注意的是,上述的初始化过程中同样生成了一个SparkContext对象,此对象可以通过ssc.sparkContext进行访问,同样在交互模式中由于默认已经存在一个SparkContext对象,因此初始化StreamingContext的话可以采用上面的例子中的方式
未完,明日继续