Spark Streaming程序基本步骤

编写Spark Streaming程序的基本步骤是：

1.通过创建输入DStream来定义输入源
2.通过对DStream应用转换操作和输出操作来定义流计算。
3.用streamingContext.start()来开始接收数据和处理流程。
4.通过streamingContext.awaitTermination()方法来等待处理结束（手动结束或因为错误而结束）。
5.可以通过streamingContext.stop()来手动结束流计算进程。

创建StreamingContext对象

请登录Linux系统，启动pyspark。进入pyspark以后，就已经获得了一个默认的SparkConext，也就是sc。因此，可以采用如下方式来创建StreamingContext对象：

>>> from pyspark import SparkContext
>>> from pyspark.streaming import StreamingContext
>>> ssc = StreamingContext(sc, 1)

1表示每隔1秒钟就自动执行一次流计算，这个秒数可以自由设定。
如果是编写一个独立的Spark Streaming程序，而不是在pyspark中运行，则需要通过如下方式创建StreamingContext对象：

from pyspark import SparkContext, SparkConf
from pyspark.streaming import StreamingContext
conf = SparkConf()
conf.setAppName('TestDStream')
conf.setMaster('local[2]')
sc = SparkContext(conf = conf)
ssc = StreamingContext(sc, 1)

setAppName(“TestDStream”)– 设置应用程序名称;

setMaster(“local[2]”) – ”local[2]’表示本地模式，启动2个工作线程。

文件流(DStream) – 命令行中监听

Spark支持从兼容HDFS API的文件系统中读取数据，创建数据流。

为了能够演示文件流的创建，我们需要首先创建一个日志目录(/usr/local/spark/python_code/streaming)，并在里面放置两个模拟的日志文件。log1.txt输入：

I love Hadoop
I love Spark
Spark is fast

请另外打开一个终端窗口，启动进入pyspark。

>>> from operator import add
>>> from pyspark import SparkContext
>>> from pyspark.streaming import StreamingContext
>>> ssc = StreamingContext(sc,20)
>>> lines = ssc.textFileStream('file:///usr/local/spark/python_code/streaming/logfile')
>>> words = lines.flatMap(lambda line:line.split(' '))
>>> wordCounts = words.map(lambda word:(word,1)).reduceByKey(add)
>>> wordCounts.pprint()
>>> ssc.start()
>>> ssc.awaitTermination()-------------------------------------------
Time: 2018-08-13 20:40:40
-------------------------------------------

>>> ssc.awaitTermination()-------------------------------------------
Time: 2018-08-13 20:41:00
-------------------------------------------

-------------------------------------------
Time: 2018-08-13 20:41:20
-------------------------------------------

-------------------------------------------
Time: 2018-08-13 20:41:40
-------------------------------------------

输入ssc.start()以后，程序就开始自动进入循环监听状态，屏幕上会显示一堆的信息,下面的ssc.awaitTermination()是无法输入到屏幕上的。

Spark Streaming每隔20秒就监听一次。但是，监听程序只监听”/usr/local/spark/mycode/streaming/logfile”目录下在程序启动后新增的文件，不会去处理历史上已经存在的文件。所以，为了让我们能够看到效果，需要到”/usr/local/spark/mycode/streaming/logfile”目录下再新建一个log3.txt文件。请打开另外一个终端窗口再新建一个log3.txt文件，里面随便输入一些英文单词，保存后再切换回到spark-shell窗口。
现在你会发现屏幕上不断输出新的信息，导致你无法看清。所以必须停止这个监听程序，按键盘Ctrl+D，或者Ctrl+C。
你可以看到屏幕上，在一大堆输出信息中，你可以找到打印出来的单词统计信息。

文件流(DStream) – 编写独立应用程序进行监听

打开一个Linux终端窗口，进入shell命令提示符状态，然后，执行下面命令：

cd /usr/local/spark/mycode/streaming
vim TestStreaming.py

在TestStreaming.py中输入以下代码：

from operator import add
from pyspark import SparkContext, SparkConf
from pyspark.streaming import StreamingContext
conf = SparkConf()
conf.setAppName('TestDStream')
conf.setMaster('local[1]')
sc = SparkContext(conf = conf)
ssc = StreamingContext(sc, 20)
lines = ssc.textFileStream('file:///usr/local/spark/mycode/streaming/logfile')
words = lines.flatMap(lambda line: line.split(' '))
wordCounts = words.map(lambda x : (x,1)).reduceByKey(add)
wordCounts.pprint()
ssc.start()
ssc.awaitTermination()

保存成功后，执行python TestStreaming.py

套接字流(DStream)

Spark Streaming可以通过Socket端口监听并接收数据，然后进行相应处理。
在/usr/local/spark/python_code/streaming目录下新建NetworkWordCount.py，输入如下内容：

from __future__ import print_function

import sys

from pyspark import SparkContext
from pyspark.streaming import StreamingContext

if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Usage: network_wordcount.py <hostname> <port>", file=sys.stderr)
        exit(-1)
    sc = SparkContext(appName="PythonStreamingNetworkWordCount")
    ssc = StreamingContext(sc, 1)

    lines = ssc.socketTextStream(sys.argv[1], int(sys.argv[2]))
    counts = lines.flatMap(lambda line: line.split(" "))\
                  .map(lambda word: (word, 1))\
                  .reduceByKey(lambda a, b: a+b)
    counts.pprint()

    ssc.start()
    ssc.awaitTermination()

保存后，执行：
sudo nc -lk port #永久监听TCP端口

第二个终端作为监听窗口，执行如下代码：

cd /usr/local/spark/mycode/streaming
python3 NetworkWordCount.py localhost 9999

这样，就可以在nc第一个终端窗口窗口中随意输入一些单词，在监听窗口每隔1秒就会打印出词频统计信息。

RDD队列流（DStream）

我们可以使用streamingContext.queueStream(queueOfRDD)创建基于RDD队列的DStream。

打开一个终端，进入Shell命令提示符状态，然后执行下面命令新建代码文件：
在/usr/local/spark/mycode/streaming目录下创建TestRDDQueueStream.py
输入以下代码：

import time
 
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
 
if __name__ == "__main__":
 
    sc = SparkContext(appName="PythonStreamingQueueStream")
    ssc = StreamingContext(sc, 1)
 
    # Create the queue through which RDDs can be pushed to
    # a QueueInputDStream
    rddQueue = []
    for i in range(5):
        rddQueue += [ssc.sparkContext.parallelize([j for j in range(1, 1001)], 10)]
 
    # Create the QueueInputDStream and use it do some processing
    inputStream = ssc.queueStream(rddQueue)
    mappedStream = inputStream.map(lambda x: (x % 10, 1))
    reducedStream = mappedStream.reduceByKey(lambda a, b: a + b)
    reducedStream.pprint()
 
    ssc.start()
    time.sleep(6)
    ssc.stop(stopSparkContext=True, stopGraceFully=True)

保存并回到命令行，输入python3 TestRDDQueueStream.py，得到结果如下：