Apache Spark 流式处理概述

此定义是静态的，且在运行该应用程序前没有处理任何数据。

创建 StreamingContext

从指向群集的 SparkContext 创建 StreamingContext。 创建 StreamingContext 时，指定批处理的大小（以秒为单位），例如：

import org.apache.spark._
import org.apache.spark.streaming._
val ssc = new StreamingContext(sc, Seconds(1))
创建 DStream
使用 StreamingContext 实例，为输入源创建输入 DStream。 在这种情况下，应用程序会监视默认附加存储中的新文件的外观。
val lines = ssc.textFileStream("/uploads/Test/")
通过对 DStream 应用转换来实现处理。 此应用程序一次从文件接收一行文本，将每行文本拆分成字。 然后使用映射化简模式来计算每个字出现的次数。
val words = lines.flatMap(_.split(" "))
val pairs = words.map(word => (word, 1))
val wordCounts = pairs.reduceByKey(_ + _)
通过应用输出操作，将转换结果推送到目标系统。 在本例中，整个计算中的每个运行的结果都将打印到控制台输出中。
wordCounts.print()
运行应用程序
启动流式处理应用程序并运行，直到收到终止信号。
ssc.start()
ssc.awaitTermination()
有关 Spark 流 API 的详细信息，请参阅 Apache Spark 流式处理编程指南。
下面的示例应用程序是自包含的，因此可以在 Jupyter Notebook 内运行它。 此示例将在类 DummySource 中创建模拟数据源，该源每五秒输出计数器的值以及当前时间（以毫秒为单位）。 新 StreamingContext 对象的批处理间隔为 30 秒。 每创建一个批次时，流式处理应用程序都会检查生成的 RDD。 然后将 RDD 转换为 Spark DataFrame 并通过 DataFrame 创建一个临时表。
class DummySource extends org.apache.spark.streaming.receiver.Receiver[(Int, Long)](org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK_2) {
    /** Start the thread that simulates receiving data */
    def onStart() {
        new Thread("Dummy Source") { override def run() { receive() } }.start()
    def onStop() {  }
    /** Periodically generate a random number from 0 to 9, and the timestamp */
    private def receive() {
        var counter = 0  
        while(!isStopped()) {
            store(Iterator((counter, System.currentTimeMillis)))
            counter += 1
            Thread.sleep(5000)
// A batch is created every 30 seconds
val ssc = new org.apache.spark.streaming.StreamingContext(spark.sparkContext, org.apache.spark.streaming.Seconds(30))
// Set the active SQLContext so that we can access it statically within the foreachRDD
org.apache.spark.sql.SQLContext.setActive(spark.sqlContext)
// Create the stream
val stream = ssc.receiverStream(new DummySource())
// Process RDDs in the batch
stream.foreachRDD { rdd =>
    // Access the SQLContext and create a table called demo_numbers we can query
    val _sqlContext = org.apache.spark.sql.SQLContext.getOrCreate(rdd.sparkContext)
    _sqlContext.createDataFrame(rdd).toDF("value", "time")
        .registerTempTable("demo_numbers")
// Start the stream processing
ssc.start()
在启动上面的应用程序以后，等待约 30 秒钟。  然后，你可以定期查询 DataFrame 来查看批处理中存在的当前值集，例如，使用以下 SQL 查询：
%%sql
SELECT * FROM demo_numbers
生成的输出如下所示：
value
存在六个值，因为 DummySource 每隔 5 秒创建一个值，且应用程序每隔 30 秒发出一个批处理。
若要对某个时间段内的 DStream 执行聚合计算，例如获取最后 2 秒钟内的平均温度，可以使用 Spark 流式处理中包括的 sliding window 操作。 滑动窗口具有一个持续时间（窗口长度）和在期间计算窗口内容的时间间隔（即滑动间隔）。
滑动窗口可以重叠，例如，可以定义长度为两秒的窗口，每一秒滑动一次。 该操作意味着每次执行聚合计算时，窗口将包括上一个窗口最后一秒的数据。 以及下一秒的任何新数据。
下面的示例将更新使用 DummySource 的代码，将批处理收集到持续时间为一分钟且滑动为一分钟的窗口中。
class DummySource extends org.apache.spark.streaming.receiver.Receiver[(Int, Long)](org.apache.spark.storage.StorageLevel.MEMORY_AND_DISK_2) {
    /** Start the thread that simulates receiving data */
    def onStart() {
        new Thread("Dummy Source") { override def run() { receive() } }.start()
    def onStop() {  }
    /** Periodically generate a random number from 0 to 9, and the timestamp */
    private def receive() {
        var counter = 0  
        while(!isStopped()) {
            store(Iterator((counter, System.currentTimeMillis)))
            counter += 1
            Thread.sleep(5000)
// A batch is created every 30 seconds
val ssc = new org.apache.spark.streaming.StreamingContext(spark.sparkContext, org.apache.spark.streaming.Seconds(30))
// Set the active SQLContext so that we can access it statically within the foreachRDD
org.apache.spark.sql.SQLContext.setActive(spark.sqlContext)
// Create the stream
val stream = ssc.receiverStream(new DummySource())
// Process batches in 1 minute windows
stream.window(org.apache.spark.streaming.Minutes(1)).foreachRDD { rdd =>
    // Access the SQLContext and create a table called demo_numbers we can query
    val _sqlContext = org.apache.spark.sql.SQLContext.getOrCreate(rdd.sparkContext)
    _sqlContext.createDataFrame(rdd).toDF("value", "time")
    .registerTempTable("demo_numbers")
// Start the stream processing
ssc.start()
第一分钟后，会产生 12 个条目 - 窗口中收集到的两个批处理中各有 6 个条目。
value
Spark 流式传输 API 中可用的滑动窗口函数包括 window、countByWindow、reduceByWindow 和 countByValueAndWindow。 有关这些函数的详细信息，请参阅 Transformations on DStreams（DStreams 的转换）。
为提供复原和容错功能，Spark 流式处理依赖检查点，确保即使出现节点故障，流处理仍可以不间断地继续。 Spark 创建持久存储（Azure 存储或 Data Lake Storage）的检查点。 这些检查点存储流式处理应用程序元数据，例如应用程序定义的配置和操作。 以及已排队但尚未处理的所有批处理。 有时，检查点还包括将数据保存在 RDD 中以便更快速地基于由 Spark 托管的 RDD 中存在的内容重新生成数据状态。
部署 Spark 流式处理应用程序
通常将 Spark 流式处理应用程序本地生成到 JAR 文件中。 然后通过将 JAR 文件复制到默认的附加存储将其部署到 HDInsight 上的 Spark。 可以使用 POST 操作，通过群集上可用的 LIVY REST API 启动该应用程序。 POST 的正文包括提供 JAR 路径的 JSON 文档。 和其 main 方法定义并运行流式处理应用程序的类的名称以及可选的作业资源要求（例如执行器、内存和核心的数量）。 以及应用程序代码所需的任何配置设置。
此外，可以使用 GET 请求针对 LIVY 终结点检查所有应用程序的状态。 最后，可以通过针对 LIVY 终结点发出 DELETE 请求终止正在运行的应用程序。 有关 LIVY API 的详细信息，请参阅使用 Apache LIVY 执行远程作业
在 HDInsight 中创建 Apache Spark 群集
Apache Spark Streaming Programming Guide（Apache Spark 流式处理编程指南）
Apache Spark 结构化流的概述