图 8-2。结构化流式处理使用微批处理模型

第 1 步：定义输入源

与批处理查询一样，第一步是从流式源定义 DataFrame。但是，在读取批处理数据源时，我们需要spark.read创建一个. 具有与 大部分相同的方法，因此您可以以类似的方式使用它。下面是从要通过套接字连接接收的文本数据流创建 DataFrame 的示例：DataFrameReaderspark.readStreamDataStreamReaderDataStreamReaderDataFrameReader

# In Python
spark = SparkSession...
lines = (spark
  .readStream.format("socket")
  .option("host", "localhost")
  .option("port", 9999)
  .load())

// In Scala 
val spark = SparkSession...
val lines = spark
  .readStream.format("socket")
  .option("host", "localhost")
  .option("port", 9999)
  .load()

此代码将linesDataFrame 生成为从 localhost:9999 读取的以换行符分隔的文本数据的无界表。请注意，与带有 的批处理源类似spark.read，这不会立即开始读取流数据；它仅设置显式启动流式查询后读取数据所需的配置。

除了套接字之外，Apache Spark 本身还支持从 Apache Kafka 读取数据流以及所有DataFrameReader支持的各种基于文件的格式（Parquet、ORC、JSON 等）。本章稍后将讨论这些来源及其支持的选项的详细信息。此外，流式查询可以定义多个输入源，包括流式和批处理，可以使用联合和连接等 DataFrame 操作进行组合（本章稍后也会讨论）。

第 2 步：转换数据

现在我们可以应用通常的 DataFrame 操作，例如将行拆分为单个单词，然后对它们进行计数，如下面的代码所示：

# In Python
from pyspark.sql.functions import *
words = lines.select(explode(split(col("value"), "\\s")).alias("word"))
counts = words.groupBy("word").count()

// In Scala
import org.apache.spark.sql.functions._
val words = lines.select(explode(split(col("value"), "\\s")).as("word"))
val counts = words.groupBy("word").count()

counts是一个流式 DataFrame（即，无界流式数据上的 DataFrame），它表示一旦启动流式查询并且正在连续处理流式输入数据时将计算的运行字数。

请注意，如果是批处理 DataFrame ，这些转换lines流式 DataFrame 的操作将以完全相同的方式工作。lines一般来说，大多数可以应用于批处理 DataFrame 的 DataFrame 操作也可以应用于流式 DataFrame。要了解结构化流式处理中支持哪些操作，您必须认识两大类数据转换:

无状态转换

select(), filter(),等操作map()不需要前行的任何信息来处理下一行；每一行都可以自己处理。这些操作中缺乏先前的“状态”使它们成为无状态的。无状态操作可以应用于批处理和流数据帧。

有状态的转换

相比之下，像这样的聚合操作count()需要维护状态以跨多行组合数据。更具体地说，任何涉及分组、加入或聚合的 DataFrame 操作都是有状态的转换。虽然结构化流中支持其中许多操作，但不支持它们的一些组合，因为以增量方式计算它们要么计算困难要么不可行。

结构化流支持的有状态操作以及如何在运行时管理它们的状态将在本章后面讨论。

第 3 步：定义输出接收器和输出模式

转换数据后，我们可以定义如何写入处理后的输出数据DataFrame.writeStream（而不是DataFrame.write，用于批处理数据）。这将创建一个DataStreamWriterwhich，类似于DataFrameWriter，具有指定以下内容的其他方法：

• 输出写入细节（在哪里以及如何写入输出）

• 处理细节（如何处理数据以及如何从故障中恢复）

让我们从输出写入细节开始（我们将在下一步中重点关注处理细节）。例如，以下代码段显示了如何将 final 写入counts控制台：

# In Python
writer = counts.writeStream.format("console").outputMode("complete")

// In Scala
val writer = counts.writeStream.format("console").outputMode("complete")

在这里，我们指定"console"了输出流接收器和"complete"输出模式。流式查询的输出模式指定在处理新输入数据后要写出更新输出的哪一部分。在这个例子中，当处理大量新输入数据并更新字数时，我们可以选择将迄今为止看到的所有单词的计数（即完整模式）或仅打印到控制台在最后一块输入数据中更新。这是由指定的输出模式决定的，可以是以下之一（正如我们在“结构化流的编程模型”中已经看到的那样：

追加模式

这是默认模式，只有自上次触发后添加到结果表/DataFrame（例如counts表）的新行才会输出到接收器。从语义上讲，这种模式保证输出的任何行在将来都不会被查询更改或更新。因此，仅那些永远不会修改先前输出数据的查询（例如，无状态查询）支持追加模式。相比之下，我们的字数查询可以更新之前生成的字数；因此，它不支持附加模式。

完成模式

在此模式下，结果表/DataFrame 的所有行将在每次触发结束时输出。这由结果表可能比输入数据小得多的查询支持，因此可以保留在内存中。例如，我们的字数查询支持完整模式，因为计数数据可能远小于输入数据。

更新模式

在这种模式下，只有自上次触发后更新的结果表/DataFrame 的行将在每次触发结束时输出。这与追加模式相反，因为输出行可能会被查询修改并在将来再次输出。大多数查询支持更新模式。

除了将输出写入控制台之外，Structured Streaming 本身还支持对文件和 Apache Kafka 的流式写入。此外，您可以使用foreachBatch()和foreach()API 方法写入任意位置。事实上，您可以使用foreachBatch()现有的批处理数据源来编写流输出（但您将失去完全一次的保证）。这些接收器的详细信息及其支持的选项将在本章后面讨论。

第 4 步：指定处理细节

开始查询之前的最后一步是指定如何处理数据的详细信息。继续我们的字数统计示例，我们将指定处理细节，如下所示：

# In Python
checkpointDir = "..."
writer2 = (writer
  .trigger(processingTime="1 second")
  .option("checkpointLocation", checkpointDir))

// In Scala
import org.apache.spark.sql.streaming._
val checkpointDir = "..."
val writer2 = writer
  .trigger(Trigger.ProcessingTime("1 second"))
  .option("checkpointLocation", checkpointDir)

DataStreamWriter在这里，我们使用我们创建的 指定了两种类型的详细信息DataFrame.writeStream：

触发细节

这指示何时触发新可用流数据的发现和处理。有四个选项：

默认

如果未明确指定触发器，则默认情况下，流式查询以微批次执行数据，其中下一个微批次在前一个微批次完成后立即触发。

带触发间隔的处理时间

您可以显式指定ProcessingTime带有间隔的触发器，查询将以该固定间隔触发微批处理。

一次

在这种模式下，流式查询将只执行一个微批次——它会在一个批次中处理所有可用的新数据，然后自行停止。当您想控制外部调度程序的触发和处理时，这很有用，外部调度程序将使用任何自定义调度重新启动查询（例如，通过每天仅执行一次查询来控制成本）。

连续的

这是一种实验模式（从 Spark 3.0 开始），流式查询将连续处理数据，而不是微批处理。虽然只有一小部分 DataFrame 操作允许使用此模式，但它可以提供比微批量触发模式低得多的延迟（低至毫秒）。有关最新信息，请参阅最新的结构化流编程指南。

检查点位置

这是任何与 HDFS 兼容的文件系统中的一个目录，流式查询在其中保存其进度信息，即已成功处理的数据。失败时，此元数据用于在失败的查询中重新启动失败的查询。因此，设置此选项对于具有完全一次保证的故障恢复是必要的。

第五步：开始查询

指定所有内容后，最后一步是启动查询，您可以使用以下操作：

# In Python
streamingQuery = writer2.start()

// In Scala
val streamingQuery = writer2.start()

返回的 type 对象streamingQuery代表一个活动的查询，可以用来管理查询，我们将在本章后面介绍。

请注意，这start()是一种非阻塞方法，因此一旦在后台开始查询，它将立即返回。如果您希望主线程阻塞直到流式查询终止，您可以使用streamingQuery.awaitTermination(). 如果查询在后台因错误awaitTermination()而失败，也会因同样的异常而失败。

您可以使用 等待最多超时持续时间awaitTermination(timeoutMillis)，并且可以使用 显式停止查询streamingQuery.stop()。

把它们放在一起

总而言之，这里是通过套接字读取文本数据流、计算单词并将计数打印到控制台的完整代码：

# In Python
from pyspark.sql.functions import *
spark = SparkSession...
lines = (spark
  .readStream.format("socket")
  .option("host", "localhost")
  .option("port", 9999)
  .load())

words = lines.select(explode(split(col("value"), "\\s")).alias("word"))
counts = words.groupBy("word").count()
checkpointDir = "..."
streamingQuery = (counts
  .writeStream
  .format("console")
  .outputMode("complete")
  .trigger(processingTime="1 second")
  .option("checkpointLocation", checkpointDir)
  .start())
streamingQuery.awaitTermination()

// In Scala
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming._
val spark = SparkSession...
val lines = spark
  .readStream.format("socket")
  .option("host", "localhost")
  .option("port", 9999)
  .load()

val words = lines.select(explode(split(col("value"), "\\s")).as("word"))
val counts = words.groupBy("word").count()

val checkpointDir = "..."
val streamingQuery = counts.writeStream
  .format("console")
  .outputMode("complete")
  .trigger(Trigger.ProcessingTime("1 second"))
  .option("checkpointLocation", checkpointDir)
  .start()
streamingQuery.awaitTermination()

查询开始后，后台线程不断地从流式传输源读取新数据，对其进行处理，然后将其写入流式传输接收器。接下来，让我们快速了解一下它是如何执行的。

图 8-5。流式查询的增量执行

笔记

关于结构化流，您应该记住的一个关键点是，在它下面使用 Spark SQL 来执行数据。因此，Spark SQL 的超优化执行引擎的全部功能被用来最大化流处理吞吐量，提供关键的性能优势。

使用 StreamingQuery 查询当前状态

StreamingQuery您可以使用实例查询活动查询的当前运行状况。这里有两种方法：

// In Scala/Python
{
  "id" : "ce011fdc-8762-4dcb-84eb-a77333e28109",
  "runId" : "88e2ff94-ede0-45a8-b687-6316fbef529a",
  "name" : "MyQuery",
  "timestamp" : "2016-12-14T18:45:24.873Z",
  "numInputRows" : 10,
  "inputRowsPerSecond" : 120.0,
  "processedRowsPerSecond" : 200.0,
  "durationMs" : {
    "triggerExecution" : 3,
    "getOffset" : 2// In Scala/Python
{
  "id" : "ce011fdc-8762-4dcb-84eb-a77333e28109",
  "runId" : "88e2ff94-ede0-45a8-b687-6316fbef529a",
  "name" : "MyQuery",
  "timestamp" : "2016-12-14T18:45:24.873Z",
  "numInputRows" : 10,
  "inputRowsPerSecond" : 120.0,
  "processedRowsPerSecond" : 200.0,
  "durationMs" : {
    "triggerExecution" : 3,
    "getOffset" : 2
  },
  "stateOperators" : [ ],
  "sources" : [ {
    "description" : "KafkaSource[Subscribe[topic-0]]",
    "startOffset" : {
      "topic-0" : {
        "2" : 0,
        "1" : 1,
        "0" : 1
      }
    },
    "endOffset" : {
      "topic-0" : {
        "2" : 0,
        "1" : 134,
        "0" : 534
      }
    },
    "numInputRows" : 10,
    "inputRowsPerSecond" : 120.0,
    "processedRowsPerSecond" : 200.0
  } ],
  "sink" : {
    "description" : "MemorySink"
  }
}
  },
  "stateOperators" : [ ],
  "sources" : [ {
    "description" : "KafkaSource[Subscribe[topic-0]]",
    "startOffset" : {
      "topic-0" : {
        "2" : 0,
        "1" : 1,
        "0" : 1
      }
    },
    "endOffset" : {
      "topic-0" : {
        "2" : 0,
        "1" : 134,
        "0" : 534
      }
    },
    "numInputRows" : 10,
    "inputRowsPerSecond" : 120.0,
    "processedRowsPerSecond" : 200.0
  } ],
  "sink" : {
    "description" : "MemorySink"
  }
}

// In Scala/Python
{
  "message" : "Waiting for data to arrive",
  "isDataAvailable" : false,
  "isTriggerActive" : false
}

使用 Dropwizard Metrics 发布指标

Spark 通过一个名为Dropwizard Metrics的流行库支持报告指标。该库允许将指标发布到许多流行的监控框架（Ganglia、Graphite 等）。由于报告的数据量很大，因此默认情况下不会为结构化流查询启用这些指标。要启用它们，除了为 Spark 配置 Dropwizard Metrics之外，您还必须在开始查询之前将SparkSession配置显式设置spark.sql.streaming.metricsEnabled为。true

请注意，只有一部分可用信息StreamingQuery.lastProgress()通过 Dropwizard Metrics 发布。如果您想不断地向任意位置发布更多进度信息，则必须编写自定义侦听器，如下所述。

使用自定义 StreamingQueryListeners 发布指标

StreamingQueryListener是一个事件侦听器接口，您可以使用它注入任意逻辑来持续发布指标。此开发人员 API 仅在 Scala/Java 中可用。使用自定义侦听器有两个步骤：

1. 定义您的自定义侦听器。该StreamingQueryListener接口提供了三种方法，您的实现可以定义这些方法来获取与流式查询相关的三种类型的事件：开始、进度（即执行了触发器）和终止。这是一个例子：

   // In Scala
   import org.apache.spark.sql.streaming._
   val myListener = new StreamingQueryListener() {
     override def onQueryStarted(event: QueryStartedEvent): Unit = {
       println("Query started: " + event.id)
     }
     override def onQueryTerminated(event: QueryTerminatedEvent): Unit = {
       println("Query terminated: " + event.id)
     }
     override def onQueryProgress(event: QueryProgressEvent): Unit = {
       println("Query made progress: " + event.progress)
     }
   }

2. SparkSession在开始查询之前将您的侦听器添加到：

   // In Scala
   spark.streams.addListener(myListener)

   添加监听器后，在此运行的所有流式查询事件都SparkSession将开始调用监听器的方法.

从文件中读取

结构化流可以将写入目录的文件视为数据流。这是一个例子：

# In Python
from pyspark.sql.types import *
inputDirectoryOfJsonFiles =  ... 

fileSchema = (StructType()
  .add(StructField("key", IntegerType()))
  .add(StructField("value", IntegerType())))

inputDF = (spark
  .readStream
  .format("json")
  .schema(fileSchema)
  .load(inputDirectoryOfJsonFiles))

// In Scala
import org.apache.spark.sql.types._
val inputDirectoryOfJsonFiles =  ... 

val fileSchema = new StructType()
  .add("key", IntegerType)
  .add("value", IntegerType)

val inputDF = spark.readStream
  .format("json")
  .schema(fileSchema)
  .load(inputDirectoryOfJsonFiles)

返回的流数据帧将具有指定的模式。以下是使用文件时要记住的几个关键点：

• 所有文件必须具有相同的格式，并且应该具有相同的架构。例如，如果格式为"json"，则所有文件必须为 JSON 格式，每行一条 JSON 记录。每条 JSON 记录的架构必须与用 指定的架构相匹配readStream()。违反这些假设可能导致不正确的解析（例如，意外null值）或查询失败。

• 每个文件都必须以原子方式出现在目录列表中——也就是说，整个文件必须立即可供读取，并且一旦可用，就无法更新或修改文件。这是因为结构化流式处理将在引擎找到文件时（使用目录列表）处理该文件，并在内部将其标记为已处理。不会处理对该文件的任何更改。

• 当有多个新文件要处理但它只能在下一个微批处理中选择其中一些时（例如，由于速率限制），它将选择具有最早时间戳的文件。然而，在微批处理中，没有预定义的选定文件的读取顺序；所有这些都将被并行读取。

写入文件

结构化流式处理支持将流式查询输出写入与读取格式相同的文件。然而，它只支持追加模式，因为虽然在输出目录中写入新文件很容易（即，将数据追加到目录中），但很难修改现有数据文件（正如更新和完成模式所期望的那样） . 它还支持分区。这是一个例子：

# In Python
outputDir = ...
checkpointDir = ...
resultDF = ...

streamingQuery = (resultDF.writeStream
  .format("parquet")
  .option("path", outputDir)
  .option("checkpointLocation", checkpointDir)
  .start())

// In Scala
val outputDir = ...
val checkpointDir = ...
val resultDF = ...
 
val streamingQuery = resultDF
  .writeStream
  .format("parquet")
  .option("path", outputDir)
  .option("checkpointLocation", checkpointDir)
  .start()

您"path"可以直接将输出目录指定为start(outputDir).

要记住的几个关键点：

• 结构化流式处理通过维护已写入目录的数据文件的日志，在写入文件时实现端到端的一次性保证。此日志保存在子目录_spark_metadata中。对目录（而不是其子目录）的任何 Spark 查询都将自动使用日志来读取正确的数据文件集，从而保持完全一次保证（即，不会读取重复数据或部分文件）。请注意，其他处理引擎可能不知道此日志，因此可能无法提供相同的保证。

• 如果您在重新启动之间更改结果 DataFrame 的架构，则输出目录将包含多个架构中的数据。查询目录时必须协调这些模式。

Reading from Kafka

要从 Kafka 执行分布式读取，您必须使用选项来指定如何连接到源。假设您要订阅来自主题的数据"events"。以下是创建流式 DataFrame 的方法：

# In Python
inputDF = (spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "events")
  .load())

// In Scala
val inputDF = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "events")
  .load()

返回的 DataFrame 将具有表 8-1中描述的模式。

您还可以选择订阅多个主题、主题模式，甚至是主题的特定分区。此外，您可以选择是只读取订阅主题中的新数据还是处理这些主题中的所有可用数据。您甚至可以从批量查询中读取 Kafka 数据——也就是说，将 Kafka 主题视为表。有关更多详细信息，请参阅Kafka 集成指南。

写给 Kafka

为了写入 Kafka，Structured Streaming 期望结果 DataFrame 有一些特定名称和类型的列，如表 8-2中所述。

您可以在所有三种输出模式下写入 Kafka，但不建议使用完整模式，因为它会重复输出相同的记录。以下是将我们之前的字数查询的输出以更新模式写入 Kafka 的具体示例：

# In Python
counts = ... # DataFrame[word: string, count: long]
streamingQuery = (counts
  .selectExpr(
    "cast(word as string) as key", 
    "cast(count as string) as value")
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("topic", "wordCounts")
  .outputMode("update")
  .option("checkpointLocation", checkpointDir)
  .start())

// In Scala
val counts = ... // DataFrame[word: string, count: long]
val streamingQuery = counts
  .selectExpr(
    "cast(word as string) as key", 
    "cast(count as string) as value")
  .writeStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("topic", "wordCounts")
  .outputMode("update")
  .option("checkpointLocation", checkpointDir)
  .start()

有关更多详细信息，请参阅Kafka 集成指南.

写入任何存储系统

有两种操作允许您将流式查询的输出写入任意存储系统：foreachBatch()和foreach(). 它们的用例略有不同：虽然foreach()允许在每一行上自定义写入逻辑，但foreachBatch()允许在每个微批次的输出上进行任意操作和自定义逻辑。让我们更详细地探讨它们的用法。

# In Python
hostAddr = "<ip address>"
keyspaceName = "<keyspace>"
tableName = "<tableName>"

spark.conf.set("spark.cassandra.connection.host", hostAddr)

def writeCountsToCassandra(updatedCountsDF, batchId):
    # Use Cassandra batch data source to write the updated counts
    (updatedCountsDF
      .write
      .format("org.apache.spark.sql.cassandra")
      .mode("append")
      .options(table=tableName, keyspace=keyspaceName)
      .save())
      
streamingQuery = (counts
  .writeStream
  .foreachBatch(writeCountsToCassandra)
  .outputMode("update")
  .option("checkpointLocation", checkpointDir)
  .start())

// In Scala
import org.apache.spark.sql.DataFrame

val hostAddr = "<ip address>"
val keyspaceName = "<keyspace>"
val tableName = "<tableName>"

spark.conf.set("spark.cassandra.connection.host", hostAddr)

def writeCountsToCassandra(updatedCountsDF: DataFrame, batchId: Long) {
    // Use Cassandra batch data source to write the updated counts
    updatedCountsDF
      .write
      .format("org.apache.spark.sql.cassandra")
      .options(Map("table" -> tableName, "keyspace" -> keyspaceName))
      .mode("append")
      .save()
    }

val streamingQuery = counts
  .writeStream
  .foreachBatch(writeCountsToCassandra _)
  .outputMode("update")
  .option("checkpointLocation", checkpointDir)
  .start()

# In Python
def writeCountsToMultipleLocations(updatedCountsDF, batchId):
  updatedCountsDF.persist()
  updatedCountsDF.write.format(...).save()  # Location 1
  updatedCountsDF.write.format(...).save()  # Location 2
  updatedCountsDF.unpersist()

// In Scala
def writeCountsToMultipleLocations(
  updatedCountsDF: DataFrame, 
  batchId: Long) {
    updatedCountsDF.persist()
    updatedCountsDF.write.format(...).save()  // Location 1
    updatedCountsDF.write.format(...).save()  // Location 2
    updatedCountsDF.unpersist()
 }

# In Python
# Variation 1: Using function
def process_row(row):
    # Write row to storage
    pass

query = streamingDF.writeStream.foreach(process_row).start()  

# Variation 2: Using the ForeachWriter class
class ForeachWriter:
  def open(self, partitionId, epochId):
    # Open connection to data store
    # Return True if write should continue
    # This method is optional in Python 
    # If not specified, the write will continue automatically
    return True

  def process(self, row):
    # Write string to data store using opened connection
    # This method is NOT optional in Python
    pass

  def close(self, error):
    # Close the connection. This method is optional in Python
    pass

resultDF.writeStream.foreach(ForeachWriter()).start()

// In Scala
import org.apache.spark.sql.ForeachWriter
val foreachWriter = new ForeachWriter[String] {  // typed with Strings

    def open(partitionId: Long, epochId: Long): Boolean = {
      // Open connection to data store
      // Return true if write should continue
    }

    def process(record: String): Unit = {
      // Write string to data store using opened connection
    }

    def close(errorOrNull: Throwable): Unit = {
      // Close the connection
    }
 }

resultDSofStrings.writeStream.foreach(foreachWriter).start()

从任何存储系统读取

不幸的是，从 Spark 3.0 开始，用于构建自定义流式传输源和接收器的 API 仍处于试验阶段。Spark 3.0 中的 DataSourceV2 计划引入了流 API，但它们尚未被声明为稳定的。因此，没有官方方法可以从任意存储系统中读取.

笔记

一些逻辑操作根本上要么不切实际，要么增量计算非常昂贵，因此结构化流不支持它们。例如，任何尝试使用类似cube()or的操作启动流式查询rollup()都会抛出UnsupportedOperationException.

分布式和容错状态管理

回想第1章和第2章，在集群中运行的 Spark 应用程序有一个驱动程序和一个或多个执行程序。在驱动程序中运行的 Spark 调度程序将您的高级操作分解为更小的任务并将它们放入任务队列中，当资源可用时，执行程序从队列中拉出任务以执行它们。流式查询中的每个微批处理本质上执行一组这样的任务，这些任务从流式源读取新数据并将更新的输出写入流式传输下沉。对于有状态的流处理查询，除了写入接收器之外，每个微批处理任务都会生成中间状态数据，这些数据将被下一个微批处理使用。这种状态数据的生成是完全分区和分布式的（因为所有的读取、写入和处理都在 Spark 中），并且缓存在执行器内存中以供高效使用。这在图 8-6中进行了说明，它显示了在我们的原始流式字数查询中如何管理状态。

每个微批次读取一组新单词，在执行程序中将它们打乱以对它们进行分组，计算微批次中的计数，最后将它们添加到运行计数中以产生新计数。这些新计数既是下一个微批次的输出又是状态，因此它们被缓存在执行器的内存中。下一个微批量数据在 executor 之间按照与之前完全相同的方式进行分组，这样每个单词总是由同一个 executor 处理，因此可以在本地读取和更新其运行计数。

但是，仅将此状态保留在内存中是不够的，因为任何故障（执行程序或整个应用程序）都会导致内存中状态丢失。为了避免丢失，我们将键/值状态更新作为更改日志同步保存在用户提供的检查点位置。这些更改与每个批次中处理的偏移范围共同进行版本化，并且可以通过读取检查点日志来自动重建所需的状态版本。如果发生任何故障，结构化流能够通过重新处理相同的输入数据以及它在该微批次之前的相同状态来重新执行失败的微批次，从而产生与如果没有失败。这对于确保端到端的一次性保证至关重要。

总而言之，对于所有有状态的操作，Structured Streaming 通过以分布式方式自动保存和恢复状态来确保操作的正确性。根据有状态操作，您可能需要做的就是调整状态清理策略，以便可以自动从缓存状态中删除旧键和值。这就是我们接下来要讨论的。

有状态操作的类型

流状态的本质是保留过去数据的摘要。有时需要从状态中清理旧摘要，以便为新摘要腾出空间。根据这是如何完成的，我们可以区分两种类型的有状态操作：

托管状态操作

它们根据特定于操作的“旧”定义自动识别和清理旧状态。您可以调整定义为旧的内容以控制资源使用（例如，用于存储状态的执行程序内存）。属于这一类的操作是那些:

• 流式聚合

• 流-流连接

• 流式重复数据删除

非托管有状态操作

这些操作让您可以定义自己的自定义状态清理逻辑。此类别中的操作是：

• MapGroupsWithState

• FlatMapGroupsWithState

这些操作允许您定义任意有状态操作（会话化等）。

以下各节将详细讨论这些操作中的每一个。

笔记

您不能在流式 DataFrames 上使用直接聚合操作，例如DataFrame.count()和。Dataset.reduce()这是因为，对于静态 DataFrame，这些操作会立即返回最终计算的聚合，而对于流式 DataFrame，聚合必须不断更新。因此，您必须始终在流式 DataFrame 上使用DataFrame.groupBy()or进行聚合。Dataset.groupByKey()

图 8-7。将事件时间映射到翻滚窗口

图 8-8。将事件时间映射到多个重叠窗口

图 8-9。每五分钟触发后更新结果表中的计数

处理带有水印的延迟数据

水印被定义为事件时间的移动阈值，该阈值落后于处理数据中查询看到的最大事件时间。尾随间隙（称为水印延迟）定义了引擎将等待延迟数据到达的时间。通过知道给定组不再有数据到达的点，引擎可以自动完成某些组的聚合并将它们从状态中删除。这限制了引擎为计算查询结果而必须维护的状态总量。

例如，假设您知道您的传感器数据延迟不会超过 10 分钟。然后可以如下设置水印：

# In Python
(sensorReadings
  .withWatermark("eventTime", "10 minutes")
  .groupBy("sensorId", window("eventTime", "10 minutes", "5 minutes"))
  .mean("value"))

// In Scala
sensorReadings
  .withWatermark("eventTime", "10 minutes")
  .groupBy("sensorId", window("eventTime", "10 minutes", "5 minute"))
  .mean("value")

请注意，您必须在用于定义窗口的时间戳列withWatermark()之前和之前调用。groupBy()执行该查询时，Structured Streaming 将持续跟踪该eventTime列的最大观察值，并相应地更新水印，过滤“太迟”的数据，并清除旧状态。也就是说，任何迟到 10 分钟以上的数据都将被忽略，所有比最新（按事件时间）输入数据早 10 分钟以上的时间窗口都将从状态中清除。为了阐明这个查询将如何执行，请考虑图 8-10中的时间线，该时间线显示了如何处理选择的输入记录。

图 8-10。说明引擎如何跨事件跟踪最大事件时间、更新水印并相应地处理延迟数据

此图显示了根据处理时间（x 轴）和事件时间（y 轴）处理的记录的二维图。记录以五分钟的微批次进行处理，并用圆圈标记。底部的表格显示了每个微批次完成后结果表的状态。

每条记录在其左侧的所有记录之后被接收和处理。考虑两条记录[12:15, id1]（在 12:17 左右处理）和[12:13, id3]（在 12:18 左右处理）。记录id3被认为是迟到的（因此标记为纯红色），因为它是在记录之前由传感器生成的，id1但在后者之后处理。但是，在处理时间范围12:15–12:20的微批处理中，使用的水印是 12:04，它是根据直到前一个微批处理看到的最大事件时间（即 12:14减去 10 分钟的水印延迟）。因此，迟到的记录[12:13, id3]不算太晚，算成功了。相比之下，在下一个微批次中，记录[12:04, id1]与12:11的新水印相比，被认为为时已晚，被丢弃。

您可以根据应用程序的要求设置水印延迟 - 此参数的较大值允许数据稍后到达，但以增加状态大小（即内存使用）为代价，反之亦然。

支持的输出模式

与不涉及时间的流式聚合不同，具有时间窗口的聚合可以使用所有三种输出模式。但是，您需要注意关于状态清理的其他含义，具体取决于模式：

更新模式

在这种模式下，每个微批次将只输出聚合更新的行。此模式可用于所有类型的聚合。特别是对于时间窗口聚合，水印将确保定期清理状态。这是使用流式聚合运行查询的最有用和最有效的模式。但是，您不能使用此模式将聚合写入仅附加的流接收器，例如任何基于文件的格式，如 Parquet 和 ORC（除非您使用 Delta Lake，我们将在下一章讨论）。

完成模式

在这种模式下，每个微批次都将输出所有更新的聚合，无论它们的年龄或是否包含更改。虽然此模式可用于所有类型的聚合，但对于时间窗口聚合，使用完整模式意味着即使指定了水印也不会清除状态。输出所有聚合需要所有过去的状态，因此即使定义了水印，也必须保留聚合数据。谨慎地在时间窗口聚合上使用此模式，因为这可能会导致状态大小和内存使用量的无限增加。

追加模式

此模式只能用于事件时间窗口上的聚合并启用水印. 回想一下，附加模式不允许更改以前的输出结果。对于任何没有水印的聚合，每个聚合都可以用任何未来的数据进行更新，因此这些不能以附加模式输出。只有在事件时间窗口上的聚合上启用水印时，查询才会知道聚合何时不会进一步更新。因此，仅当水印确保聚合不会再次更新时，追加模式才输出每个键及其最终聚合值，而不是输出更新的行。这种模式的优点是它允许您将聚合写入仅附加的流接收器（例如，文件）。.

图 8-11。使用流-流连接的广告获利

带有可选水印的内部连接

假设我们已将impressionsDataFrame 重新定义为流式 DataFrame。要获得匹配的展示流及其相应的点击，我们可以使用之前用于静态连接和流静态连接的相同代码：

# In Python
# Streaming DataFrame [adId: String, impressionTime: Timestamp, ...]
impressions = spark.readStream. ... 

# Streaming DataFrame[adId: String, clickTime: Timestamp, ...]
clicks = spark.readStream. ...
matched = impressions.join(clicks, "adId")

// In Scala
// Streaming DataFrame [adId: String, impressionTime: Timestamp, ...] 
val impressions = spark.readStream. ...

// Streaming DataFrame[adId: String, clickTime: Timestamp, ...] 
val clicks = spark.readStream. ...
val matched = impressions.join(clicks, "adId")

即使代码相同，执行方式也完全不同。当这个查询被执行时，处理引擎将把它识别为流-流连接而不是流-静态连接。引擎会将所有点击和展示作为状态进行缓冲，并在收到的点击与缓冲的展示匹配时立即生成匹配的展示和点击（反之亦然，具体取决于哪个先收到）。让我们使用图 8-12中的事件时间线示例来可视化这个内部连接是如何工作的。

图 8-12。点击次数、展示次数及其联合输出的说明性时间表

在图 8-12中，蓝点表示在不同微批次中接收到的印象和点击事件的事件时间（由灰色虚线分隔）。出于此说明的目的，假设每个事件实际上是在与事件时间相同的挂钟时间收到的。请注意加入相关事件的不同场景。adId带有= ⧮的两个事件都在同一个微批次中接收，因此它们的联合输出是由该微批次生成的。然而，对于adId= ⧉印象是在 12:04 收到的，比相应的点击在 12:13 早得多。结构化流式传输将首先在 12:04 接收展示，并将其缓冲在状态中。对于每次收到的点击，引擎都会尝试将其与所有缓冲的展示一起加入（反之亦然）。最终，在 12:13 左右运行的后续微批处理中，引擎接收到adId= ⧉的点击并生成连接输出。

然而，在这个查询中，我们没有给出引擎应该缓冲多长时间来找到匹配的事件。因此，引擎可能会永远缓冲一个事件并累积无限量的流状态。要限制由流-流连接维护的流状态，您需要了解有关您的用例的以下信息：

• 两个事件在各自来源产生之间的最大时间范围是多少？在我们的用例的上下文中，我们假设点击可以在相应的展示之后的 0 秒到 1 小时内发生。

• 事件在源和处理引擎之间的传输过程中可以延迟的最大持续时间是多少？例如，来自浏览器的广告点击可能会由于间歇性连接而延迟，并且比预期晚得多，并且出现故障。假设展示次数和点击次数最多可以分别延迟两个小时和三个小时。

这些延迟限制和事件时间约束可以使用水印和时间范围条件在 DataFrame 操作中进行编码。换句话说，您必须在连接中执行以下附加步骤以确保状态清理：

1. 在两个输入上定义水印延迟，以便引擎知道输入可以延迟多长时间（类似于流式聚合）。

2. 定义跨两个输入的事件时间约束，以便引擎可以确定何时不需要一个输入的旧行（即不满足时间约束）与另一个输入匹配。可以通过以下方式之一定义此约束：

   1. 时间范围连接条件（例如，连接条件 = ）"leftTime BETWEEN rightTime AND rightTime + INTERVAL 1 HOUR"

   2. 加入事件时间窗口（例如，加入条件 = ）"leftTimeWindow = rightTimeWindow"

在我们的广告用例中，我们的内部连接代码会变得有点复杂：

# In Python
# Define watermarks
impressionsWithWatermark = (impressions
  .selectExpr("adId AS impressionAdId", "impressionTime")
  .withWatermark("impressionTime", "2 hours"))

clicksWithWatermark = (clicks
  .selectExpr("adId AS clickAdId", "clickTime")
  .withWatermark("clickTime", "3 hours"))

# Inner join with time range conditions
(impressionsWithWatermark.join(clicksWithWatermark,
  expr(""" 
    clickAdId = impressionAdId AND 
    clickTime BETWEEN impressionTime AND impressionTime + interval 1 hour""")))

// In Scala
// Define watermarks
val impressionsWithWatermark = impressions
  .selectExpr("adId AS impressionAdId", "impressionTime")
  .withWatermark("impressionTime", "2 hours ")

val clicksWithWatermark = clicks
  .selectExpr("adId AS clickAdId", "clickTime")
  .withWatermark("clickTime", "3 hours")

// Inner join with time range conditions
impressionsWithWatermark.join(clicksWithWatermark,
  expr(""" 
    clickAdId = impressionAdId AND 
    clickTime BETWEEN impressionTime AND impressionTime + interval 1 hour"""))

使用每个事件的这些时间限制，处理引擎可以自动计算需要缓冲多长时间才能生成正确的结果，以及何时可以从状态中删除事件。例如，它将评估以下内容（如图 8-13 所示）：

• 展示最多需要缓冲四个小时（在事件时间中），因为延迟三个小时的点击可能与四个小时前的展示相匹配（即延迟三个小时 + 展示之间最多延迟一小时）点击）。

• 相反，点击最多需要缓冲两个小时（在事件时间），因为延迟两个小时的展示可能与两个小时前收到的点击相匹配。

图 8-13。结构化流式处理使用水印延迟和时间范围条件自动计算状态清理的阈值

关于内部连接，需要记住几个关键点：

• 对于内部连接，指定水印和事件时间约束都是可选的。换句话说，冒着潜在无界状态的风险，您可以选择不指定它们。只有当两者都被指定时，你才会得到状态清理。

• 与聚合加水印提供的保证类似，水印延迟两小时保证引擎永远不会丢弃或不匹配任何延迟少于两小时的数据，但延迟超过两小时的数据可能会或可能不会被处理.

带水印的外部连接

先前的内部连接将仅输出已收到两个事件的广告。换句话说，根本不会报告未获得点击的广告。相反，您可能希望报告所有广告展示，无论是否包含相关的点击数据，以便稍后进行其他分析（例如，点击率）。这给我们带来了流-流外连接。您需要做的就是指定外部连接类型：

# In Python
# Left outer join with time range conditions
(impressionsWithWatermark.join(clicksWithWatermark,
  expr(""" 
    clickAdId = impressionAdId AND 
    clickTime BETWEEN impressionTime AND impressionTime + interval 1 hour"""),
  "leftOuter"))  # only change: set the outer join type

// In Scala
// Left outer join with time range conditions
impressionsWithWatermark.join(clicksWithWatermark,
  expr(""" 
    clickAdId = impressionAdId AND 
    clickTime BETWEEN impressionTime AND impressionTime + interval 1 hour"""),
  "leftOuter")  // Only change: set the outer join type

正如外连接所期望的那样，这个查询将开始为每次展示生成输出，无论有没有（即使用NULL）点击数据。但是，关于外连接还有几点需要注意：

• 与内连接不同，水印延迟和事件时间约束对于外连接不是可选的。这是因为为了生成NULL结果，引擎必须知道将来什么时候事件不会与其他任何事件匹配。为了正确的外部连接结果和状态清理，必须指定水印和事件时间约束。

• 因此，外部NULL结果将延迟生成，因为引擎必须等待一段时间以确保既没有也不会有任何匹配。该延迟是引擎为每个事件计算的最大缓冲时间（相对于事件时间），如上一节所述（即，展示时间为 4 小时，点击时间为 2 小时）。

笔记

您必须提供几个附加参数。所有类型 ( K, V, S, U) 都必须可由 Spark SQL 的编码器编码。因此，在 中mapGroupsWithState()，您必须在 Scala 中隐式或在 Java 中显式提供类型化编码S器U。有关详细信息，请参阅第 6 章中的“数据集编码器”。

处理时间超时

处理时间超时基于运行流式查询的机器的系统时间（也称为挂钟时间），定义如下：如果密钥最后在系统时间戳 接收数据T，并且当前时间戳大于，然后将使用新的空数据迭代器再次调用该函数。(T + <timeout duration>)

让我们通过更新我们的用户示例来研究如何使用超时，以删除基于一小时不活动的用户状态。我们将进行三个更改：

• 在mapGroupsWithState()中，我们将超时指定为GroupStateTimeout.ProcessingTimeTimeout。

• 在状态更新函数中，在用新数据更新状态之前，我们必须检查状态是否超时。因此，我们将更新或删除状态。

• 此外，每次我们用新数据更新状态时，我们都会设置超时时间。

这是更新的代码：

// In Scala
def updateUserStatus(
    userId: String, 
    newActions: Iterator[UserAction],
    state: GroupState[UserStatus]): UserStatus = {

  if (!state.hasTimedOut) {       // Was not called due to timeout
    val userStatus = state.getOption.getOrElse {
      new UserStatus(userId, false)
    }
    newActions.foreach { action => userStatus.updateWith(action) }
    state.update(userStatus) 
    state.setTimeoutDuration("1 hour") // Set timeout duration
    return userStatus
    
  } else {
    val userStatus = state.get()
    state.remove()                  // Remove state when timed out
    return userStatus.asInactive()  // Return inactive user's status
  }
}
 
val latestStatuses = userActions
  .groupByKey(userAction => userAction.userId) 
  .mapGroupsWithState(
    GroupStateTimeout.ProcessingTimeTimeout)(
    updateUserStatus _)

此查询将自动清理查询超过一个小时未处理任何数据的用户的状态。但是，关于超时，有几点需要注意：

• 上次调用该函数设置的超时时间会在再次调用该函数时自动取消，无论是新接收到的数据还是超时。因此，无论何时调用该函数，都需要显式设置超时持续时间或时间戳以启用超时。

• 由于超时是在微批处理期间处理的，因此它们的执行时间是不精确的，并且在很大程度上取决于触发间隔和微批处理时间。因此，不建议使用超时来进行精确的时序控制。

• 虽然处理时间超时很容易推理，但它们对减速和停机时间并不稳健。如果流式查询的宕机时间超过一小时，那么重启后，所有处于该状态的键都会超时，因为每个键接收到数据后已经过去了一个多小时。如果查询处理数据的速度比到达源的速度慢（例如，如果数据到达并在 Kafka 中缓冲），也会发生类似的大规模超时。例如，如果超时为 5 分钟，则处理速率的突然下降（或数据到达率的峰值）导致 5 分钟的延迟可能会产生虚假超时。为了避免此类问题，我们可以使用事件时间超时，我们将在下面讨论。

事件时间超时

事件时间超时不是基于系统时钟时间，而是基于数据中的事件时间（类似于基于时间的聚合）和在事件时间上定义的水印。如果一个键配置了特定的超时时间戳T（即，不是持续时间），那么T如果自上次调用函数以来没有接收到该键的新数据，则该键将在水印超过时超时。回想一下，水印是一个移动阈值，它滞后于处理数据时看到的最大事件时间。因此，与系统时间不同，水印在时间上以与处理数据相同的速率向前移动。这意味着（与处理时间超时不同）查询处理中的任何减速或停机时间都不会导致虚假超时。

让我们修改我们的示例以使用事件时间超时。除了我们已经为使用处理时间超时所做的更改之外，我们还将进行以下更改：

• 在输入数据集上定义水印（假设类UserAction有一个eventTimestamp字段）。回想一下，水印阈值表示输入数据可能延迟和乱序的可接受时间量。

• 更新mapGroupsWithState()使用EventTimeTimeout。

• 更新函数以设置将发生超时的阈值时间戳。请注意，事件时间超时不允许设置超时持续时间，例如处理时间超时。我们稍后会讨论这个原因。在这个例子中，我们将这个超时时间计算为当前水印加上一小时。

这是更新的示例：

// In Scala
def updateUserStatus(
    userId: String, 
    newActions: Iterator[UserAction],
    state: GroupState[UserStatus]):UserStatus = {

  if (!state.hasTimedOut) {  // Was not called due to timeout
    val userStatus = if (state.getOption.getOrElse {
      new UserStatus()
    }
    newActions.foreach { action => userStatus.updateWith(action) }
    state.update(userStatus)

    // Set the timeout timestamp to the current watermark + 1 hour
    state.setTimeoutTimestamp(state.getCurrentWatermarkMs, "1 hour") 
    return userStatus
  } else {
    val userStatus = state.get()
    state.remove()
    return userStatus.asInactive() }
}

val latestStatuses = userActions
  .withWatermark("eventTimestamp", "10 minutes") 
  .groupByKey(userAction => userAction.userId) 
  .mapGroupsWithState(
    GroupStateTimeout.EventTimeTimeout)(
    updateUserStatus _)

此查询对于由重启和处理延迟引起的虚假超时将更加健壮。

以下是有关事件时间超时的几点注意事项：

• 与前面的处理时间超时示例不同，我们使用GroupState.setTimeoutTimestamp()了GroupState.setTimeoutDuration(). 这是因为对于处理时间超时，持续时间足以计算发生超时的确切未来时间戳（即当前系统时间 + 指定持续时间），但对于事件时间超时，情况并非如此。不同的应用程序可能希望使用不同的策略来计算阈值时间戳。在这个例子中，我们只是根据当前的 watermark 来计算它，但不同的应用程序可能会选择根据为该键看到的最大事件时间时间戳（作为状态的一部分被跟踪并保存）来计算该键的超时时间戳。

• 超时时间戳必须设置为大于当前水印的值。这是因为当时间戳越过水印时预计会发生超时，因此将时间戳设置为已经大于当前水印的值是不合逻辑的。

在我们从超时开始之前，要记住的最后一件事是，您可以使用这些超时机制进行比固定持续时间超时更具创造性的处理。例如，您可以通过将最后一个任务执行时间戳保存在状态中并使用它来设置处理时间超时持续时间来在状态上实现一个近似周期性的任务（例如，每个小时），如此代码片段所示:

// In Scala
timeoutDurationMs = lastTaskTimstampMs + periodIntervalMs - 
groupState.getCurrentProcessingTimeMs()