一 基本概念
streaming的重要特点是使用了spark DataFrame
spark streaming 是数据流式系统，采用RDD批量模式加速处理数据，以小批量或者批次间隔运行
spark streaming接受输入数据流 将其分为多个较小的batch spark引擎将这些数据的batch处理后，生成处理过数据的batch结果集
spark streaming的主要抽象是离散流（Dstream） 代表了数据流中的一个小批量数据 Dstream建立在RDD上
可与MLLIB SQL DataFrame GraphX集成
目前 spark streaming 有四种广泛的应用场景
- 流ETL 将数据推入下游分析系统之前对其进行持续的数据清洗和聚合
- 触发器 实时检测行为或异常事件 及时触发下游动作
- 数据浓缩 将实时数据和其他数据集合连接 进行更加丰富的分析
- 复杂会话和持续学习 与实时流相关的多组事件被持续分析，从而更新机器学习模型 例如与在线游戏相关联的用户活动流
二 streaming工作流程
【1】当spark streaming启动上下文时，驱动进程会对工作节点excutor执行长时间运行的任务
【2】excutor中的receiver进程从streaming源中获取数据 并切分成多个数据块保存至内存中
【3】这些数据块被复制到另一个excutor中进行数据备份保存
【4】数据块的ID信息被传送到driver上的块管理Master
【5】对于在streaming context内配置的批次间隔，driver将启动spark任务对每个批次的数据块进行处理，然后这些数据被持久化到目标存储中 如NoSQL
代码实践
1.Dstream完成wordcount
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
sc = SparkContext("local[2]","NewWordCount")
ssc = StreamingContext(sc,1) # 1 表示1秒间隔
lines = ssc.socketTextStream("localhost",9999)
words = lines.flatMap(lambda line:line.strip().split(" "))
pairs = words.map(lambda x:(x,1))
wordCounts = pairs.reduceByKey(lambda x, y: x + y)
wordCounts.pprint()
ssc.start() #启动spark
ssc.awaitTermination()
2.全局聚合 保留信息的状态 返回特定时间窗口长度内的所有数据UpdateStateByKey/mapWithState
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
sc = SparkContext("local[2]","statefulNetworkWordCount")
ssc = StreamingContext(sc,1)
ssc.checkpoint("checkpoint")
def updateFunc(new_values,last_sum):
return sum(new_values,last_sum)
lines = ssc.socketTextStream("localhost",9999)
running_counts = lines.flatMap(lambda line:line.split(" ")).map(lambda x:(x,1)).updateStateByKey(updateFunc)
running_counts.pprint()
ssc.start()
ssc.awaitTermination()
注意：通常在spark1.5 之前的版本使用updateStateByKey
spark1.6之后应该使用mapWithState 性能和批量的大小成正比
3.而在spark2.0 中 引入了结构化流 structured streaming 将streaming的概念与Dataset/DataFrame进行了整合
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode
from pyspark.sql.functions import split
spark = SparkSession.builder \
.appName("ATestName") \
.getOrcreate()
lines = spark.readStream \
.format("socket") \
.option("host","localhost") \
.option("port",9999) \
.load()
words = lines.select(explode(split(lines.value," ")).alias("word"))
wordCounts = words.groupBy("words").count()
query = wordCounts.writeStream \
.outputMode("complete") \
.format("console") \
.start()
query.awaitTermination()
