前言

前面我已经学习过怎么使用RDD和spark sql，在这里，就学习一下spark Streaming吧。

原理

在Spark中，一个应用（Application）由一个任务控制节点（Driver）和若干个作业（Job）构成，一个作业由多个阶段（Stage）构成，一个阶段由多个任务（Task）组成。

当执行一个应用时，任务控制节点会向集群管理器（Cluster Manager）申请资源，启动Executor，并向Executor发送应用程序代码和文件，然后在Executor上执行task。

在Spark Streaming中，会有一个组件Receiver，作为一个长期运行的task跑在一个Executor上。每个Receiver都会负责一个input DStream（比如从文件中读取数据的文件流，比如套接字流，或者从Kafka中读取的一个输入流等等）。

Spark Streaming通过input DStream与外部数据源进行连接，读取相关数据。

Spark Streaming程序基本步骤

1.通过创建DStream来定义输入源

2.通过对DStream应用转换操作和输出操作来定义流计算

3.用streamingContext.strart()来开始接收数据和处理流程

4.通过streamingContext.awaitTermination()方法来等待处理结束

5.也可以通过streamingContext.stop()来手动结束流计算

创建StreamingContext对象

登录linux，python启动spark-shell

>>> from pyspark import SparkContext
>>> from pyspark.streaming import StreamingContext
>>> ssc = StreamingContext(sc, 1) #1表示每隔1秒钟就自动执行一次流计算，这个秒数可以自由设定。

转换操作

DStream转换操作包括无状态转换（ 每个批次的处理不依赖于之前批次的数据）和有状态转换（ 前批次的处理需要使用之前批次的数据或者中间结果。

有状态转换包括基于滑动窗口的转换和追踪状态变化的转换）。

无状态转换有

* map(func) ：对源DStream的每个元素，采用func函数进行转换，得到一个新的DStream；

* flatMap(func)：与map相似，但是每个输入项可用被映射为0个或者多个输出项；

* filter(func)：返回一个新的DStream，仅包含源DStream中满足函数func的项；

* repartition(numPartitions)：通过创建更多或者更少的分区改变DStream的并行程度；

* union(otherStream)：返回一个新的DStream，包含源DStream和其他DStream的元素；

* count()：统计源DStream中每个RDD的元素数量；

* reduce(func)：利用函数func聚集源DStream中每个RDD的元素，返回一个包含单元素RDDs的新DStream；

* countByValue()：应用于元素类型为K的DStream上，返回一个（K，V）键值对类型的新DStream，每个键的值是在原DStream的每个RDD中的出现次数；

* reduceByKey(func, [numTasks])：当在一个由(K,V)键值对组成的DStream上执行该操作时，返回一个新的由(K,V)键值对组成的DStream，每一个key的值均由给定的recuce函数（func）聚集起来；

* join(otherStream, [numTasks])：当应用于两个DStream（一个包含（K,V）键值对,一个包含(K,W)键值对），返回一个包含(K, (V, W))键值对的新DStream；

* cogroup(otherStream, [numTasks])：当应用于两个DStream（一个包含（K,V）键值对,一个包含(K,W)键值对），返回一个包含(K, Seq[V], Seq[W])的元组；

* transform(func)：通过对源DStream的每个RDD应用RDD-to-RDD函数，创建一个新的DStream。支持在新的DStream中做任何RDD操作。

有状态转换

有状态转换包括基于滑动窗口的转换和追踪状态变化(updateStateByKey)的转换。

基于滑动窗口的转换

* window(windowLength, slideInterval) 基于源DStream产生的窗口化的批数据，计算得到一个新的DStream；

* countByWindow(windowLength, slideInterval) 返回流中元素的一个滑动窗口数；

* reduceByWindow(func, windowLength, slideInterval) 返回一个单元素流。利用函数func聚集滑动时间间隔的流的元素创建这个单元素流。函数func必须满足结合律，从而可以支持并行计算；

* reduceByKeyAndWindow(func, windowLength, slideInterval, [numTasks]) 应用到一个(K,V)键值对组成的DStream上时，会返回一个由(K,V)键值对组成的新的DStream。每一个key的值均由给定的reduce函数(func函数)进行聚合计算。注意：在默认情况下，这个算子利用了Spark默认的并发任务数去分组。可以通过numTasks参数的设置来指定不同的任务数；

* reduceByKeyAndWindow(func, invFunc, windowLength, slideInterval, [numTasks]) 更加高效的reduceByKeyAndWindow，每个窗口的reduce值，是基于先前窗口的reduce值进行增量计算得到的；它会对进入滑动窗口的新数据进行reduce操作，并对离开窗口的老数据进行“逆向reduce”操作。但是，只能用于“可逆reduce函数”，即那些reduce函数都有一个对应的“逆向reduce函数”（以InvFunc参数传入）；

* countByValueAndWindow(windowLength, slideInterval, [numTasks]) 当应用到一个(K,V)键值对组成的DStream上，返回一个由(K,V)键值对组成的新的DStream。每个key的值都是它们在滑动窗口中出现的频率。

追踪状态变化的转换

举个例子

from __future__ import print_functionimport sysfrom pyspark import SparkContextfrom pyspark.streaming import StreamingContextif __name__ == "__main__":    if len(sys.argv) != 3:        print("Usage: stateful_network_wordcount.py <hostname> <port>", file=sys.stderr)        exit(-1)    sc = SparkContext(appName="PythonStreamingStatefulNetworkWordCount")    ssc = StreamingContext(sc, 1)    ssc.checkpoint("file:///opt/spark/mycode/streaming/")    # RDD with initial state (key, value) pairs    initialStateRDD = sc.parallelize([(u'hello', 1), (u'world', 1)])    def updateFunc(new_values, last_sum):        return sum(new_values) + (last_sum or 0)    lines = ssc.socketTextStream(sys.argv[1], int(sys.argv[2]))    running_counts = lines.flatMap(lambda line: line.split(" "))\                          .map(lambda word: (word, 1))\                          .updateStateByKey(updateFunc, initialRDD=initialStateRDD)    running_counts.pprint()    ssc.start() ssc.awaitTermination()

输出操作

把DStream输出到文本文件

加上这条语句就行

running_counts.saveAsTextFiles("file:///opt/spark/mycode/streaming/output.txt")

把DStream写入mysql

创建表

create table wordcount (word char(20), count int(4));

导入mysql模块

pip3 install PyMySQL

加入以下代码

def dbfunc(records):        db = pymysql.connect("localhost","root","root","spark")        cursor = db.cursor()        def doinsert(p):            sql = "insert into wordcount(word,count) values ('%s', '%s')" % (str(p[0]), str(p[1]))            try:                cursor.execute(sql)                db.commit()            except:                db.rollback()        for item in records:            doinsert(item)    def func(rdd):        repartitionedRDD = rdd.repartition(3)        repartitionedRDD.foreachPartition(dbfunc)    running_counts.foreachRDD(func)

总结

对于spark Streaming的学习就学完了，这里学的概念比较多，实践比较少，下次有机会多多实践。

THE END