spark教程(15)-Streaming

Spark Streaming 是一个分布式数据流处理框架，它可以近乎实时的处理流数据，它易编程，可以处理大量数据，并且能把实时数据与历史数据结合起来处理。

Streaming 使得 spark 具有了流式处理的能力，它为数据流式处理提供了高层抽象，底层仍然是 spark，所以它具有 spark 的可扩展、可容错、高吞吐量的特点，而且它可以与 spark 的各种库结合使用，如 sparkSQL、MLib、ml 等

总体架构

Spark Streaming 是一个伪实时的流处理框架，它处理的是一个微批次的数据流，就是说他把数据流按照非常小的时间间隔切分成一批一批的数据，然后每次处理一批数据；

每一批数据仍然以 RDD 方式存储，然后使用 spark core 进行处理；

RDD 操作结果也是一批一批的输出；

数据流来源

Streaming 支持多种数据流来源，比较常用的有 TCP网络传输、Kafa、Flume 等，还有 Twitter、ZeroMQ、MQTT 等，

它也可以把一个文件当成流来处理，

也可以自定义数据流来源；

流数据经过 spark 处理后可以流向各种地方；

总结一下如下图

接收器

也叫数据采集器，接收器收到数据后存储在内存中；

Streaming 在每个 worker 上为每个数据流来源创建一个接收器；

一个 spark 应用可以同时接收多个数据流来源，然后统一处理；

API

Streaming API 有两个高度抽象 StreamingContext 和 离散流 DStream

StreamingContext

他是 Streaming 库的入口点，使得 Streaming 连接到 spark 上；

每个 Streaming 应用必须先创建一个 StreamingContext 实例；

创建 StreamingContext 实例

创建方法和 SparkContext 一样，创建 sc 的方法都能用来创建 StreamingContext；

不同的是多了一个参数，指定划分数据流的时间间隔；

from pyspark import SparkContext, StreamingContext, SparkConf

conf = SparkConf().setAppName('myapp1').setMaster('local[4]')   # 设定 appname 和 master

ssc = StreamingContext(conf=conf, 10)    # 10s 间隔

## 或者这样

sc = SparkContext(conf=conf)

ssc = StreamingContext(sc, 10)  # 直接传入 sc

StreamingContext 实例的方法

ssc.start()     # 启动 流式计算，在这之前，什么也不会发生

ssc.checkpoint('hdfs path')     # 定期创建检查点数据，输入为 hdfs 的路径

ssc.stop()      # 停止 流式计算

ssc.awaitTermination()      # 等待流式计算结束

checkpoint

DStream     【内容比较多，故单独一章】

离散数据流；

他是 Streaming 处理数据流的一个高度抽象，也是一个抽象类，并且定义了一系列对该类的操作；

不同数据流来源有不同的 DStream 类；

DStream 实际上是一个 RDD 序列，Spark Streaming 把对 DStream 的操作转换成对 RDD 的操作；

因为他是 RDD 序列，所以具有 RDD 的特点：不可变、分区、容错

创建 DStream 实例

DStream 创建有两种方式，一种是从数据流来源直接创建，一种是从现有的 DStream 对象转换得到

socketTextStream：创建一个从 TCP 套接字连接 接收数据流的 DStream

3 个参数，host，port，第三个可选，指定接收数据的存储等级

def socketTextStream(self, hostname, port, storageLevel=StorageLevel.MEMORY_AND_DISK_2)

默认的存储等级为 StorageLevel.MEMORY_AND_DISK_2，顾名思义，表示接收到的数据先存储在内存中，如果内存放不下，多出来的数据会存放到硬盘上；

而且他会对接收到的数据 以 spark 序列化的方式 进行序列化操作；

所以这个存储等级会有序列化的开销，但是减少了 jvm 垃圾回收相关的问题；

接收到的数据会复制多份，提高容错；

选择合适的存储等级可以提高性能，比如 Streaming 采集周期很短，如几秒钟，也就是数据量很小，那么可以指定存储等级为只内存存储 StorageLevel.MEMORY_ONLY

textFileStream：创建一个 DStream 用于监控 hadoop 兼容的文件系统中是否有新文件创建

输入为被监控的目录；

如果有新文件创建，则将作为文本文件读出；

注意，写入被监控目录的文件必须是从同等文件系统中移动过来的，比如 linux 系统，新文件必须是用 mv 命令移动过来的

def textFileStream(self, directory)

actorStream：用户自己定义的 Akka actor 接收器的 DStream

python 好像没这个

KafkaUtils：创建从 Kafka 接收数据流的 DStream

class KafkaUtils(object):

    @staticmethod

    def createStream(ssc, zkQuorum, groupId, topics, kafkaParams=None,

                     storageLevel=StorageLevel.MEMORY_AND_DISK_2,

                     keyDecoder=utf8_decoder, valueDecoder=utf8_decoder):

        """

        Create an input stream that pulls messages from a Kafka Broker.

        :param ssc:  StreamingContext object

        :param zkQuorum:  Zookeeper quorum (hostname:port,hostname:port,..).

        :param groupId:  The group id for this consumer.

        :param topics:  Dict of (topic_name -> numPartitions) to consume.

                        Each partition is consumed in its own thread.

        :param kafkaParams: Additional params for Kafka

        :param storageLevel:  RDD storage level.

        :param keyDecoder:  A function used to decode key (default is utf8_decoder)

        :param valueDecoder:  A function used to decode value (default is utf8_decoder)

        :return: A DStream object

还有一些其他的，参考 pyspark

DStream 操作

内容比较多，后面会专门写一篇博客

DStream 输出

Dstream 可以输出到各种地方，如文件、数据库或者其他应用程序

输出到文件

saveAsTextFiles：将 DStream 保存成文件，他为每个 RDD 创建一个目录，并在每个目录里创建多个副本；

目录名称为 用户定义的前缀-时间戳-可选后缀

DStream.saveAsTextFiles('/usr/lib/text')

saveAsObjectFiles：将 DStream 以序列化对象的形式保存成二进制 SequenceFile 文件，用法同 saveAsTextFiles

saveAsHadoopFiles：只有由键值对组成的 DStream 才能使用该方法

saveAsNewAPIHadoopFiles：自己试试

输出到控制台

pprint(n)：打印指定输出元素个数

输出到数据库

foreachRDD：输入一个 func 逐个处理 DStream 中的每一个 RDD；

该方法无需返回任何东西；

    def foreachRDD(self, func):

        """

        Apply a function to each RDD in this DStream.

        """

        if func.__code__.co_argcount == 1:

            old_func = func

            func = lambda t, rdd: old_func(rdd)

        jfunc = TransformFunction(self._sc, func, self._jrdd_deserializer)

        api = self._ssc._jvm.PythonDStream

        api.callForeachRDD(self._jdstream, jfunc)

对于 foreachRDD 需要理解的是：

1. func 是把 RDD 作为输入，并且可以使用 RDD 的所有操作；

2. foreachRDD 执行在 Driver 中，而 func 执行在 Executor 中；

3. foreachRDD 不仅仅用于存到数据库

存储到数据库需要注意的是：

1. 数据库连接比较耗时，不要频繁的连接、关闭

2. 数据库连接无法序列化，也就是无法从 Driver 发送给 Executor，故数据库连接只能在 worker 上创建，并且复用

3. RDD 的 foreachPartition 操作可以使用同一个数据库连接保存多个 DStream 中的元素，可在 func 中使用该方法

4. 利用数据库连接池进行优化

5. 也可以采用批量写入的方法来优化数据库存储

示例代码

数据流来源是 TCP 套接字

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)

        sys.exit(-1)

    sc = SparkContext(appName="PythonStreamingNetworkWordCount")

    # 以指定时间为周期采集实时数据

    ssc = StreamingContext(sc, 30)   # 采集周期为 x s

    ### 连接已经开启的 socket 服务器，注意必须事先开启 socket server

    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()     # 打印 DStream

    ## 如

    # (u'a', 1)

    # (u'e', 1)

    # (u'd', 1)

    counts.pprint(2)     # 打印 DStream，并指定输出元素个数

    ## 如果 pprint 输出如上，pprint(2)输出如下

    # (u'a', 1)

    # (u'e', 1)

    counts.saveAsTextFiles('/usr/lib/text')     # DStream 保存至文件系统

    ssc.checkpoint('/spark')

    ssc.start()     # 启动采集器

    ssc.awaitTermination()      # 等待 socket ser 终止

TCP 服务器要事先存在，在 linux 上使用如下命令建立 TCP 服务器

nc -lk 9999

注意 nc 命令需要手动安装

数据流来源是 Kafka

import time

from pyspark import SparkContext

from pyspark.streaming import StreamingContext

from pyspark.streaming.kafka import KafkaUtils

from operator import add

sc = SparkContext(master="yarn",appName="PythonSparkStreamingRokidDtSnCount")

ssc = StreamingContext(sc, 2)

zkQuorum = '172.16.89.80:2181'      # broker list

topic = {'':1}          # topic name : partition

groupid = "test-consumer-group"     # consumer group

lines = KafkaUtils.createStream(ssc, zkQuorum, groupid, topic)

lines1 = lines.flatMap(lambda x: x.split("\n"))

valuestr = lines1.map(lambda x: x.value.decode())

valuedict = valuestr.map(lambda x:eval(x))

message = valuedict.map(lambda x: x["message"])

rdd2 = message.map(lambda x: (time.strftime("%Y-%m-%d",time.localtime(float(x.split("\u0001")[0].split("\u0002")[1])/1000))+"|"+x.split("\u0001")[1].split("\u0002")[1],1)).map(lambda x: (x[0],x[1]))

rdd3 = rdd2.reduceByKey(add)

rdd3.saveAsTextFiles("/tmp/wordcount")

rdd3.pprint()

ssc.start()

ssc.awaitTermination()

其他场景参考 spark 自带的样例

参考资料：

《Spark大数据分析核心概念技术及实践OCR-2017》  电子书