定义及工作原理
Spark Streaming定义，官网翻译如下
Spark Streaming是核心Spark API的扩展，可实现实时数据流的可扩展，高吞吐量，容错流处理。
数据可以从许多来源（如Kafka，Flume，Kinesis或TCP套接字）中获取，并且可以使用以高级函数（如map，reduce，join和window）表示的复杂算法进行处理。
最后，处理后的数据可以推送到文件系统，数据库和实时仪表板。 实际上，您可以在数据流上应用Spark的机器学习和图形处理算法。
工作原理，官网翻译如下
在内部，它的工作原理如下：
Spark Streaming接收实时输入数据流并将数据分成批处理，然后由Spark引擎处理以批量生成最终结果流。
小结
Spark流数据处理就是将实时数据输入，按时间间隔分批次处理（形成一系列RDD）然后按需求输出。
流数据加载之初始化
Python
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
# Create a local StreamingContext with two working thread and batch interval of 1 second
sc = SparkContext("local[2]", "NetworkWordCount")
sc.setLogLevel("WARN")
ssc = StreamingContext(sc, 5)
Scala
import org.apache.spark._
import org.apache.spark.streaming._
// Create a local StreamingContext with two working thread and batch interval of 1 second.
// The master requires 2 cores to prevent a starvation scenario.
val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
val sc = new SparkContext(conf)
sc.setLogLevel("WARN")
val ssc = new StreamingContext(conf, Seconds(5))
小结
Spark Streaming初始化必须创建一个SparkContext流上下文对象，这是所有流处理的入口，接着创建StreamingContext对象。
如果后面结构化RDD，也就是DataFrame流操作，必须使用StreamingContext正在使用的 SparkContext 创建SparkSession
流数据加载之离散化流
Discretized Stream离散化流
Discretized Stream或DStream是Spark Streaming提供的基本抽象。
它表示连续的数据流，可以是从源接收的输入数据流，也可以是通过转换输入流生成的已处理数据流。 在内部，DStream由一系列连续的RDD表示，这是Spark对不可变分布式数据集的抽象。
DStream中的每个RDD都包含来自特定时间间隔的数据。
其实，DStram很好理解，就是按时间间隔组合的一系列RDD，其特点就是离散且持续。
离散流DStream的输入
Spark Streaming提供两类内置流媒体源。
基本来源：StreamingContext API中直接提供的源。 示例：文件系统和套接字连接。如上面StreamingContext。
高级资源：Kafka，Flume，Kinesis等资源可通过额外的实用程序类获得。 这些需要链接额外的依赖关系，如链接部分所述。
离散流DStream的转化
与RDD类似，转换允许修改来自输入DStream的数据。 DStreams支持普通Spark RDD上可用的许多转换。
DStream同样与RDD一样有map、flatmap、union、reduce、transform等等，但是区别于RDD，这些transform算子都不同于RDD算子的封装操作（针对于一些了RDD的操作），但其底层还是对每个RDD的操作。
离散流DStream的输出
输出操作允许将DStream的数据推送到外部系统，如数据库或文件系统。
由于输出操作实际上允许外部系统使用转换后的数据，因此它们会触发所有DStream转换的实际执行（类似于RDD的操作）
print（在Python API中pprint）、saveAsTextFiles、foreachRDD等等输出。
其中，值得注意的是，无论是什么输出底层调用的都是foreachRDD函数。
流数据处理扩展
了解更多，请参考 官网后续包括，结构化RDD的流操作、检查点、部署等等。
DataFrame流处理操作Demo
r"""
Use DataFrames and SQL to count words in UTF8 encoded, '\n' delimited text received from the
network every second.
Usage: sql_network_wordcount.py <hostname> <port>
<hostname> and <port> describe the TCP server that Spark Streaming would connect to receive data.
To run this on your local machine, you need to first run a Netcat server
`$ nc -lk 9999`
and then run the example
`$ bin/spark-submit examples/src/main/python/streaming/sql_network_wordcount.py localhost 9999`
"""
from __future__ import print_function
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.sql import Row, SparkSession
def getSparkSessionInstance(sparkConf):
# if ('sparkSessionSingletonInstance' not in globals()):
# 创建SparkSession
# 必须使用StreamingContext正在使用的SparkContext创建SparkSession
globals()['sparkSessionSingletonInstance'] = SparkSession \
.builder \
.config(conf=sparkConf) \
.getOrCreate()
return globals()['sparkSessionSingletonInstance']
def process(time, rdd):
# Convert RDDs of the words DStream to DataFrame and run SQL query
print("========= %s =========" % str(time))
try:
# Get the singleton instance of SparkSession
spark = getSparkSessionInstance(rdd.context.getConf())
# Convert RDD[String] to RDD[Row] to DataFrame
rowRdd = rdd.map(lambda w: Row(word=w))
wordsDataFrame = spark.createDataFrame(rowRdd)
# Creates a temporary view using the DataFrame.
wordsDataFrame.createOrReplaceTempView("words")
# Do word count on table using SQL and print it
wordCountsDataFrame = \
spark.sql("select word, count(*) as total from words group by word")
wordCountsDataFrame.show()
except:
pass
if __name__ == "__main__":
# SparkContext---StreamingContext---SparkSession
host, port = ["localhost", 9999]
sc = SparkContext(appName="PythonSqlNetworkWordCount")
# 默认log应该是info级，信息较多不利于观察；
sc.setLogLevel("WARN")
ssc = StreamingContext(sc, 5)
# Create a socket stream on target ip:port and count the
# words in input stream of \n delimited text (eg. generated by 'nc')
lines = ssc.socketTextStream(host, int(port))
words = lines.flatMap(lambda line: line.split(" "))
words.foreachRDD(process)
# 标准用法，没有output（foreachRDD）都是懒加载，只会记录后删除，并不运行；
ssc.start()
ssc.awaitTermination()
网友评论