Spark Structured Streaming 流查询

向右看齐 2023-06-13 11:08 2阅读 0赞

**目录**

启用流查询

Output Modes

Output Sinks

Foreach 和 ForeachBatch

触发器

管理流查询

监控流查询

读取指标交互

使用异步api以编程方式查询指标

使用Dropwizard

--------------------

# 启用流查询 #

一旦定义了最终结果 DataFrame/Dataset，剩下的就是开始流计算了。为此，必须使用通过 Dataset.writeStream() 返回的DataStreamWriter (Scala/Java/Python文档)。您必须在这个接口中指定一个或多个以下内容。

*  输出 Sink 的信息：数据格式、位置等
 *  输出模式：指定写入输出 Sink 的内容
 *  查询名称：可选项，指定查询的唯一名称
 *  触发时间：可选项，指定触发时间。如果没有指定，系统将在前面的处理完成后立即检查新数据的可用性。如果由于前一个处理未完成而错过了触发时间，则系统将立即触发处理。
 *  CheckPoint 位置：对于一些能够保证端到端容错的输出接收器，请指定系统将写入所有检查点信息的位置。这应该是 hdfs 兼容的容错文件系统中的一个目录。

## Output Modes ##

有几种输出模式：

*  **Append mode (default) - **这是默认模式，其中只有添加到结果表(因为最后一个触发器)的新行将输出到接收器。这只适用于那些添加到结果表的行永远不会改变的查询。因此，此模式保证每个行只输出一次(假设容错接收器)。例如，只有select、where、map、flatMap、filter、join等查询将支持Append模式。
 *  **Complete mode - **每次触发后，整个结果表将输出到接收器。这对于聚合查询是受支持的。
 *  **Update mode - **(从Spark 2.1.1起可用)只有结果表中自最后一个触发器以来更新的行将输出到接收器。

不同类型的流查询支持不同的输出模式。这是兼容性矩阵。

<table> 
 <tbody> 
  <tr> 
   <th>Query Type</th> 
   <th>&nbsp;</th> 
   <th>Supported Output Modes</th> 
   <th>Notes</th> 
  </tr> 
  <tr> 
   <td>Queries with aggregation</td> 
   <td>Aggregation on event-time with watermark</td> 
   <td>Append, Update, Complete</td> 
   <td> <p>append mode 使用 watermark 来删除旧的聚合状态。但是，一个窗口聚合的输出被延迟到“withWatermark()”中指定的后期阈值，因为根据模式语义，行只能在最后确定之后(即在交叉水印之后)添加到结果表。<br><br> Update mode 使用 watermark 删除旧的聚合状态。</p> <p>&nbsp;</p> <p>Complete mode 不支持删除旧的聚合状态， 因为根据定义，此模式保存结果表中的所有数据。</p> </td> 
  </tr> 
  <tr> 
   <td>Other aggregations</td> 
   <td>Complete, Update</td> 
   <td> <p>由于没有定义 watermark (仅在其他类别中定义)，所以不会删除旧的聚合状态。</p> <p><br> 不支持追加模式，因为聚合可能会更新，从而违反此模式的语义。</p> </td> 
  </tr> 
  <tr> 
   <td colspan="2">Queries with <code>mapGroupsWithState</code></td> 
   <td>Update</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>Queries with <code>flatMapGroupsWithState</code></td> 
   <td>Append operation mode</td> 
   <td>Append</td> 
   <td>Aggregations are allowed after <code>flatMapGroupsWithState</code>.</td> 
  </tr> 
  <tr> 
   <td>Update operation mode</td> 
   <td>Update</td> 
   <td>Aggregations not allowed after <code>flatMapGroupsWithState</code>.</td> 
  </tr> 
  <tr> 
   <td colspan="2">Queries with <code>joins</code></td> 
   <td>Append</td> 
   <td>Update and Complete mode not supported yet. See the <a href="http://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#support-matrix-for-joins-in-streaming-queries" rel="nofollow">support matrix in the Join Operations section</a> for more details on what types of joins are supported.</td> 
  </tr> 
  <tr> 
   <td colspan="2">Other queries</td> 
   <td>Append, Update</td> 
   <td>不支持 complete 模式，因为将所有未聚合的数据保存在结果表中是不可行的。</td> 
  </tr> 
 </tbody> 
</table>

## Output Sinks ##

有几种类型的内置输出接收器。

*  **File sink** \- 将输出存储到一个目录中。

writeStream
        .format("parquet")        // can be "orc", "json", "csv", etc.
        .option("path", "path/to/destination/dir")
        .start()

*  **Kafka sink** \- 在Kafka中存储一个或多个主题的输出。

writeStream
        .format("kafka")
        .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
        .option("topic", "updates")
        .start()

*  **Foreach sink** \- 对输出中的记录执行任意计算。

writeStream
        .foreach(...)
        .start()

*  **Console sink (for debugging)** \- 每当有触发器时，就将输出打印到控制台/标准输出。支持 Append 和 Complete 输出模式。这应该用于在低数据量上进行调试，因为在每次触发之后，整个输出都会被收集并存储在 Driver 的内存中。

writeStream
        .format("console")
        .start()

*  **Memory sink (for debugging)** \- 输出作为内存中的表存储在内存中。支持 Append 和 Complete 输出模式。当整个输出被收集并存储在 Driver 的内存中时，这应该用于低数据量上的调试目的。因此，请谨慎使用它。

writeStream
        .format("memory")
        .queryName("tableName")
        .start()

有些接收器不是容错的，因为它们不能保证输出的持久性，只用于调试。这是 Spark 中所有 Sink 的细节。

<table> 
 <tbody> 
  <tr> 
   <th>Sink</th> 
   <th>Supported Output Modes</th> 
   <th>Options</th> 
   <th>Fault-tolerant</th> 
   <th>Notes</th> 
  </tr> 
  <tr> 
   <td><strong>File Sink</strong></td> 
   <td>Append</td> 
   <td><code>path</code>: path to the output directory, must be specified.<br><br> For file-format-specific options, see the related methods in DataFrameWriter (<a href="http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.DataFrameWriter" rel="nofollow">Scala</a>/<a href="http://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/DataFrameWriter.html" rel="nofollow">Java</a>/<a href="http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrameWriter" rel="nofollow">Python</a>/<a href="http://spark.apache.org/docs/latest/api/R/write.stream.html" rel="nofollow">R</a>). E.g. for "parquet" format options see <code>DataFrameWriter.parquet()</code></td> 
   <td>Yes (exactly-once)</td> 
   <td>支持写入到分区表。一般以时间作为分区字段</td> 
  </tr> 
  <tr> 
   <td><strong>Kafka Sink</strong></td> 
   <td>Append, Update, Complete</td> 
   <td>See the <a href="http://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html" rel="nofollow">Kafka Integration Guide</a></td> 
   <td>Yes (at-least-once)</td> 
   <td>More details in the <a href="http://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html" rel="nofollow">Kafka Integration Guide</a></td> 
  </tr> 
  <tr> 
   <td><strong>Foreach Sink</strong></td> 
   <td>Append, Update, Complete</td> 
   <td>None</td> 
   <td>Yes (at-least-once)</td> 
   <td>More details in the <a href="http://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#using-foreach-and-foreachbatch" rel="nofollow">next section</a></td> 
  </tr> 
  <tr> 
   <td><strong>ForeachBatch Sink</strong></td> 
   <td>Append, Update, Complete</td> 
   <td>None</td> 
   <td>Depends on the implementation</td> 
   <td>More details in the <a href="http://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#using-foreach-and-foreachbatch" rel="nofollow">next section</a></td> 
  </tr> 
  <tr> 
   <td><strong>Console Sink</strong></td> 
   <td>Append, Update, Complete</td> 
   <td><code>numRows</code>: Number of rows to print every trigger (default: 20)<br><code>truncate</code>: Whether to truncate the output if too long (default: true)</td> 
   <td>No</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td><strong>Memory Sink</strong></td> 
   <td>Append, Complete</td> 
   <td>None</td> 
   <td>No. But in Complete Mode, restarted query will recreate the full table.</td> 
   <td>表名是查询名称</td> 
  </tr> 
 </tbody> 
</table>

> 注意，您必须调用 start()来实际启动查询的执行。这将返回一个 StreamingQuery 对象，该对象是连续运行在 Execution 中。

// ========== DF with no aggregations ==========
    val noAggDF = deviceDataDf.select("device").where("signal > 10")   
    
    // Print new data to console
    noAggDF
      .writeStream
      .format("console")
      .start()
    
    // Write new data to Parquet files
    noAggDF
      .writeStream
      .format("parquet")
      .option("checkpointLocation", "path/to/checkpoint/dir")
      .option("path", "path/to/destination/dir")
      .start()
    
    // ========== DF with aggregation ==========
    val aggDF = df.groupBy("device").count()
    
    // Print updated aggregations to console
    aggDF
      .writeStream
      .outputMode("complete")
      .format("console")
      .start()
    
    // Have all the aggregates in an in-memory table
    aggDF
      .writeStream
      .queryName("aggregates")    // this query name will be the table name
      .outputMode("complete")
      .format("memory")
      .start()
    
    spark.sql("select * from aggregates").show()   // interactively query in-memory table

### Foreach 和 ForeachBatch  ###

foreach 和 foreachBatch 操作允许您对流查询的输出应用任意操作和写入逻辑。它们的用例略有不同—— foreach 允许在每一行上执行自定义写逻辑，而 foreachBatch 允许在每个微批的输出上执行任意操作和自定义逻辑。

**ForeachBatch**

foreachBatch(…)允许您指定在流查询的每个微批的输出数据上执行的函数。从 Spark 2.4开始，Scala、Java 和 Python 都支持这个特性。它有两个参数：一个 DataFrame 或 DataSet，该数据集具有微批处理的输出数据和微批处理的唯一 ID。

streamingDF.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
      // Transform and write batchDF 
    }.start()

使用 foreachBatch，您可以执行以下操作。

*  **Reuse existing batch data sources** \- 对于许多存储系统，可能还没有可用的流接收，但可能已经存在用于批处理查询的数据写入器。使用foreachBatch，您可以在每个微批的输出上使用批数据写入器。
 *  **Write to multiple locations** \- 如果希望将流查询的输出写入多个位置，则只需多次写入输出DataFrame/Dataset。但是，每次写操作都会导致重新计算输出数据(包括可能的重新读取输入数据)。为了避免重复计算，您应该缓存输出的DataFrame/Dataset，将其写入多个位置，然后释放它。如下：

streamingDF.writeStream.foreachBatch { (batchDF: DataFrame, batchId: Long) =>
      batchDF.persist()
      batchDF.write.format(...).save(...)  // location 1
      batchDF.write.format(...).save(...)  // location 2
      batchDF.unpersist()
    }

*  **Apply additional DataFrame operations** \- 流数据流中不支持许多 DataFrame 和 Dataset 操作，因为在这些情况下 Spark 不支持生成增量计划。使用 foreachBatch，您可以对每个微批处理输出应用这些操作。但是，您必须自己考虑执行该操作的端到端语义。

>  *  默认情况下，foreachBatch只提供至少一次写保证。但是，您可以使用提供给函数的batchId来复制输出并获得一次准确的保证。
>  *  foreachBatch 不能使用连续处理模式，因为它基本上依赖于流查询的微批处理执行。如果以连续模式写入数据，则使用 foreach。

**Foreach**

如果 foreachBatch 不是一个选项(例如，对应的批数据写入器不存在，或者是连续处理模式)，那么您可以使用 foreach 来表达您的自定义写入器逻辑。具体来说，您可以通过将数据写入逻辑划分为三种方法来表达：open、process 和 close。从 Spark 2.4 开始，foreach 就支持 Scala、Java 和 Python。

streamingDatasetOfString.writeStream.foreach(
      new ForeachWriter[String] {
    
        def open(partitionId: Long, version: Long): Boolean = {
          // Open connection
        }
    
        def process(record: String): Unit = {
          // Write string to connection
        }
    
        def close(errorOrNull: Throwable): Unit = {
          // Close the connection
        }
      }
    ).start()

当流查询启动时，Spark以如下方式调用函数或对象的方法：

*  此对象的单个副本负责查询中单个任务生成的所有数据。换句话说，一个实例负责处理以分布式方式生成的数据的一个分区。
 *  此对象必须是可序列化的，因为每个任务将获得所提供对象的一个新的序列化-反序列化副本。因此，强烈建议对写入数据进行任何初始化(例如。在调用 open()方法之后完成(打开连接或启动事务)，这意味着任务已经准备好生成数据。
 *  方法的生命周期如下：对于带有 partition\_id 的每个分区：对于带有 epoch\_id 的流数据的每批 epoch：方法 open(partitionId, epochId)被调用。如果 open(…)返回 true，那么对于分区和 batch/epoch 中的每一行，都将调用方法 process(row)。方法 close(error)是在处理行时看到错误(如果有)时调用的。
 *  close()方法(如果存在)在 open()方法存在并成功返回时调用(与返回值无关)，除非 JVM 或 Python 进程中途崩溃。

> Spark 不保证相同的输出(partitionId, epochId)，所以不能实现重复数据删除(partitionId, epochId)。例如，source 由于某些原因提供了不同数量的分区，Spark optimization 改变了分区数量，等等。如果你需要重复数据删除，试试 foreachBatch。

## 触发器 ##

流查询的触发器设置定义了流数据处理的时间，该查询是作为具有固定批处理间隔的微批处理查询执行，还是作为连续处理查询执行。下面是支持的不同类型的触发器。

<table> 
 <tbody> 
  <tr> 
   <th>Trigger Type</th> 
   <th>Description</th> 
  </tr> 
  <tr> 
   <td><em>unspecified (default)</em></td> 
   <td>如果没有显式指定触发器设置，那么默认情况下，查询将在微批处理模式下执行，在微批处理模式下，前一个微批处理完成后，将立即生成微批处理。</td> 
  </tr> 
  <tr> 
   <td><strong>Fixed interval micro-batches</strong></td> 
   <td>查询将以micro-batch模式执行，其中micro-batch将在用户指定的时间间隔启动。 
    <ul> 
     <li>如果前一个微批在间隔内完成，那么引擎将等待，直到间隔结束后再启动下一个微批。</li> 
     <li>如果前一个微批的完成时间比间隔时间长，那么下一个微批将在前一个微批完成后立即开始。</li> 
     <li>如果没有新的数据可用，那么就不会启动微批。</li> 
    </ul></td> 
  </tr> 
  <tr> 
   <td><strong>One-time micro-batch</strong></td> 
   <td>查询将仅执行*一个*微批处理来处理所有可用数据，然后自行停止。这在您希望定期启动集群、处理自上一周期以来可用的所有内容，然后关闭集群的场景中非常有用。在某些情况下，这可能导致显著的成本节约。</td> 
  </tr> 
  <tr> 
   <td><strong>Continuous with fixed checkpoint interval</strong><br><em>(测试阶段)</em></td> 
   <td>查询将以新的低延迟、连续处理模式执行。</td> 
  </tr> 
 </tbody> 
</table>

下面是一些代码示例。

import org.apache.spark.sql.streaming.Trigger
    
    // Default trigger (尽快运行微批处理)
    df.writeStream
      .format("console")
      .start()
    
    // 2秒的触发器
    df.writeStream
      .format("console")
      .trigger(Trigger.ProcessingTime("2 seconds"))
      .start()
    
    // 一次性触发
    df.writeStream
      .format("console")
      .trigger(Trigger.Once())
      .start()
    
    // 具有一秒检查点间隔的连续触发器
    df.writeStream
      .format("console")
      .trigger(Trigger.Continuous("1 second"))
      .start()

# 管理流查询 #

启动查询时创建的 StreamingQuery 对象可用于监视和管理查询。

val query = df.writeStream.format("console").start()   // 获取查询对象
    
    query.id          // 获取从检查点数据重新启动时持久运行的查询的唯一标识符
    
    query.runId       // 获取这次查询运行的惟一id，该id将在每次启动/重启时生成
    
    query.name        // 获取自动生成的或用户指定的名称
    
    query.explain()   // 打印查询的详细说明
    
    query.stop()      // 停止查询
    
    query.awaitTermination()   // 阻塞，知道查询被停止或错误
    
    query.exception       // 如果查询已被错误终止，则为异常
    
    query.recentProgress  // 此查询的最新进展更新的数组
    
    query.lastProgress    // 此流查询的最新进展更新

您可以在单个 SparkSession 中启动任意数量的查询。它们将同时运行，共享集群资源。您可以使用 sparkSession.streams() 来获得可用于管理当前活动查询的 StreamingQueryManager。

val spark: SparkSession = ...
    
    spark.streams.active    // 获取当前活动的流查询列表
    spark.streams.get(id)   // 通过查询对象的惟一id获取查询对象
    
    spark.streams.awaitAnyTermination()   // 阻塞，直到其中任何一个终止

# 监控流查询 #

有多种方法可以监视活动的流查询。您可以使用 Spark 的 Dropwizard 指标支持将指标推送到外部系统，或者以编程方式访问它们。

## 读取指标交互 ##

可以使用 streamingQuery.lastProgress()和 streamingQuery.status()直接获得活动查询的当前状态和指标。lastProgress()在 Scala 和 Java 中返回一个 StreamingQueryProgress 对象，在 Python 中返回一个具有相同字段的字典。它包含关于流的最后一个触发器中所取得的进展的所有信息—处理了什么数据、处理速率、延迟等等。还有 streamingQuery.recentProgress 返回最近几个进展的数组。

此外，streamingQuery.status()在 Scala 和 Java 中返回一个 StreamingQueryStatus 对象，在 Python 中返回一个具有相同字段的字典。它提供了关于查询正在立即执行的操作的信息——触发器是否活动、数据是否正在处理等等。

这里有几个例子。

val query: StreamingQuery = ...
    
    println(query.lastProgress)
    
    /* Will print something like the following.
    
    {
      "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
      },
      "eventTime" : {
        "watermark" : "2016-12-14T18:45:24.873Z"
      },
      "stateOperators" : [ ],
      "sources" : [ {
        "description" : "KafkaSource[Subscribe[topic-0]]",
        "startOffset" : {
          "topic-0" : {
            "2" : 0,
            "4" : 1,
            "1" : 1,
            "3" : 1,
            "0" : 1
          }
        },
        "endOffset" : {
          "topic-0" : {
            "2" : 0,
            "4" : 115,
            "1" : 134,
            "3" : 21,
            "0" : 534
          }
        },
        "numInputRows" : 10,
        "inputRowsPerSecond" : 120.0,
        "processedRowsPerSecond" : 200.0
      } ],
      "sink" : {
        "description" : "MemorySink"
      }
    }
    */
    
    
    println(query.status)
    
    /*  Will print something like the following.
    {
      "message" : "Waiting for data to arrive",
      "isDataAvailable" : false,
      "isTriggerActive" : false
    }
    */

## 使用异步api以编程方式查询指标 ##

您还可以通过附加 StreamingQueryListener (Scala/Java文档)来异步监控与 SparkSession 相关的所有查询。一旦您将定制的 StreamingQueryListener 对象与 sparkSession.streams.attachListener()连接起来，您将在查询开始和停止时以及在活动查询中取得进展时获得回调。举个例子，

val spark: SparkSession = ...
    
    spark.streams.addListener(new StreamingQueryListener() {
        override def onQueryStarted(queryStarted: QueryStartedEvent): Unit = {
            println("Query started: " + queryStarted.id)
        }
        override def onQueryTerminated(queryTerminated: QueryTerminatedEvent): Unit = {
            println("Query terminated: " + queryTerminated.id)
        }
        override def onQueryProgress(queryProgress: QueryProgressEvent): Unit = {
            println("Query made progress: " + queryProgress.progress)
        }
    })

## 使用Dropwizard ##

Spark支持使用Dropwizard库报告指标。为了能够报告结构化流查询的指标，您必须在 SparkSession 中启用 `spark.sql.streaming.metricsEnabled 配置。`

spark.conf.set("spark.sql.streaming.metricsEnabled", "true")
    // or
    spark.sql("SET spark.sql.streaming.metricsEnabled=true")

在启用此配置之后，所有在 SparkSession 中启动的查询都将通过 Dropwizard 向任何已配置的接收端报告指标(例如 Ganglia, Graphite, JMX 等)。