五、Structured Streaming的流式DataFrames/Datasets的操作

素颜马尾好姑娘i 2024-04-18 15:14 46阅读 0赞

**目录**

1、创建

2、输入源

3、操作：选择（Selection）、投射（Projection）和聚合（Aggregation）

4、Join操作

（1）Stream-static Joins

（2）Stream-steam Joins

6、不支持的操作

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

## 1、创建 ##

val sqLContext = SparkSession.builder().appName(" event-time-window_App").getOrCreate()

## 2、输入源 ##

<table> 
 <tbody> 
  <tr> 
   <th>Source</th> 
   <th>Options</th> 
   <th>Fault-tolerant</th> 
   <th>Notes</th> 
  </tr> 
  <tr> 
   <td><strong>File source</strong></td> 
   <td><code>path</code>: path to the input directory, and common to all file formats.<br><code>maxFilesPerTrigger</code>: maximum number of new files to be considered in every trigger (default: no max)<br><code>latestFirst</code>: whether to process the latest new files first, useful when there is a large backlog of files (default: false)<br><code>fileNameOnly</code>: whether to check new files based on only the filename instead of on the full path (default: false). With this set to `true`, the following files would be considered as the same file, because their filenames, "dataset.txt", are the same:<br> "file:///dataset.txt"<br> "s3://a/dataset.txt"<br> "s3n://a/b/dataset.txt"<br> "s3a://a/b/c/dataset.txt"<br><br><br> For file-format-specific options, see the related methods in <code>DataStreamReader</code> (<a href="http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.streaming.DataStreamReader" rel="nofollow">Scala</a>/<a href="http://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/streaming/DataStreamReader.html" rel="nofollow">Java</a>/<a href="http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.streaming.DataStreamReader" rel="nofollow">Python</a>/<a href="http://spark.apache.org/docs/latest/api/R/read.stream.html" rel="nofollow">R</a>). E.g. for "parquet" format options see <code>DataStreamReader.parquet()</code>.<br><br> In addition, there are session configurations that affect certain file-formats. See the <a href="http://spark.apache.org/docs/latest/sql-programming-guide.html" rel="nofollow">SQL Programming Guide</a> for more details. E.g., for "parquet", see <a href="http://spark.apache.org/docs/latest/sql-data-sources-parquet.html#configuration" rel="nofollow">Parquet configuration</a> section.</td> 
   <td>Yes</td> 
   <td>Supports glob paths, but does not support multiple comma-separated paths/globs.</td> 
  </tr> 
  <tr> 
   <td><strong>Socket Source</strong></td> 
   <td><code>host</code>: host to connect to, must be specified<br><code>port</code>: port to connect to, must be specified</td> 
   <td>No</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td><strong>Rate Source</strong></td> 
   <td><code>rowsPerSecond</code> (e.g. 100, default: 1): How many rows should be generated per second.<br><br><code>rampUpTime</code> (e.g. 5s, default: 0s): How long to ramp up before the generating speed becomes <code>rowsPerSecond</code>. Using finer granularities than seconds will be truncated to integer seconds.<br><br><code>numPartitions</code> (e.g. 10, default: Spark's default parallelism): The partition number for the generated rows.<br><br> The source will try its best to reach <code>rowsPerSecond</code>, but the query may be resource constrained, and <code>numPartitions</code> can be tweaked to help reach the desired speed.</td> 
   <td>Yes</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td><strong>Kafka Source</strong></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</td> 
   <td>&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

## 3、操作：选择（Selection）、投射（Projection）和聚合（Aggregation） ##

例子

object HelloApp {
      def main(args: Array[String]): Unit = {
        Logger.getLogger("org").setLevel(Level.ERROR)
        val sqLContext = SparkSession.builder().appName(" HelloApp").master("local[*]")
          .getOrCreate()
        val df = sqLContext.readStream.format("socket")
          .option("host","192.168.199.135")
          .option("port",9999).load()//nc -lk 9999
        import sqLContext.implicits._
        val words = df.as[String].flatMap(_.split(" "))
        // 对 "value" 列做 count，得到多行二列的 Dataset/DataFrame；即 result table
        val wordCounts = words.groupBy("value").count()
        //操作流动的DataFrame（选择和投射）
        val operationDF=wordCounts.select("value").where("count>5")
        val query  = wordCounts.writeStream// 打算写出 wordCounts 这个 Dataset/DataFrame
    //      .outputMode("complete") // 打算写出 wordCounts 的全量数据
          .outputMode("update") // 打算写出 wordCounts 的全量数据
          .format("console")// 打算写出到控制台
          .start()// 新起一个线程开始真正不停写出
        query.awaitTermination()// 当前用户主线程挂住，等待新起来的写出线程结束
      }
    }

DataFrames/Datasets 的操作和Spark SQL的操作类似

## 4、Join操作 ##

### （1）Stream-static Joins ###

从Spark 2.0，结构化流已经支持流和静态DataFrame/DataSet的连接（内连接和外连接）。

**stream-static joins是无状态的，所以不需要状态存储。某些类型的stream-static outer joins不支持。**

### （2）Stream-steam Joins ###

自Spark 2.3版本起，我们已经加入了对Stream-Stream Joins的支持。对于两个数据流做join的挑战在于：**在任何时候，两边数据集的视图都是不完整的，**这使得在输入之间查找匹配变得更加困难。从一个输入流接收到的任何行都可以与来自另一个输入流的将来的任何尚未接收到的行匹配。**因此，我们对两端输入流的过往数据进行缓存，这样我们就可以用它来匹配以后的输入数据，从而产生join的结果。**此外，和流聚合一样，我们使用**watermark机制自动的处理迟到数据以及限制state的存储。**

**a、.Inner Joins（内连接）—— watermark可选**

任何类型的列以及join条件都是支持的。然而，**随着流的运行，其state会无限的增大。为了避免这种无限增大的state，必须定义额外的join条件，使得过于旧的输入数据无法与将来的输入数据进行匹配，因此可以在state中清除它。**换句话说，必须在join中做如下附加步骤：

*  在**两端的输入数据上定义watermark**，以便引擎知道其延迟的程度；
 *  在**两端输入数据上定义event-time约束**，这样引擎就**可以计算出一个输入的旧行何时不需要与另一个输入进行匹配。**

import org.apache.spark.sql.functions.expr
    
    val impressions = spark.readStream. ...
    val clicks = spark.readStream. ...
    
    // Apply watermarks on event-time columns
    val impressionsWithWatermark = impressions.withWatermark("impressionTime", "2 hours")
    val clicksWithWatermark = clicks.withWatermark("clickTime", "3 hours")
    
    // Join with event-time constraints
    impressionsWithWatermark.join(
      clicksWithWatermark,
      expr("""
        clickAdId = impressionAdId AND
        clickTime >= impressionTime AND
        clickTime <= impressionTime + interval 1 hour
        """)
    )

**b、Outer Joins（外连接）—— watermark必须有**

对于**内连接来说，watermark+event-time约束是可选的，对于左外和右外连接来说，它们是必须指定的。**这是因为对于outer join中产生的NULL记录，引擎必须知道输入行在将来的什么时候不再匹配。一个outer join查询看起来与前面的示例非常相似，只是会有一个额外的参数将其指定为outer join。

impressionsWithWatermark.join(
      clicksWithWatermark,
      expr("""
        clickAdId = impressionAdId AND
        clickTime >= impressionTime AND
        clickTime <= impressionTime + interval 1 hour
        """),
      joinType = "leftOuter"      // can be "inner", "leftOuter", "rightOuter"
     )

**c、注意**

*  outer join的NULL记录会被延迟产生，延迟的时间取决于设定的watermark延时以及时间范围条件。这是因为**引擎必须等待以确保现在和将来都不会再有匹配。**
 *  如果join的两个输入流中的任意一个在一段时间内没有接收到数据，output join的输出将会被延迟。

**d、structured streaming支持的join**

<table> 
 <tbody> 
  <tr> 
   <th>Left Input</th> 
   <th>Right Input</th> 
   <th>Join Type</th> 
   <th>&nbsp;</th> 
  </tr> 
  <tr> 
   <td>Static</td> 
   <td>Static</td> 
   <td>All types</td> 
   <td>Supported, since its not on streaming data even though it can be present in a streaming query</td> 
  </tr> 
  <tr> 
   <td>Stream</td> 
   <td>Static</td> 
   <td>Inner</td> 
   <td>Supported, not stateful</td> 
  </tr> 
  <tr> 
   <td>Left Outer</td> 
   <td>Supported, not stateful</td> 
  </tr> 
  <tr> 
   <td>Right Outer</td> 
   <td>Not supported</td> 
  </tr> 
  <tr> 
   <td>Full Outer</td> 
   <td>Not supported</td> 
  </tr> 
  <tr> 
   <td>Static</td> 
   <td>Stream</td> 
   <td>Inner</td> 
   <td>Supported, not stateful</td> 
  </tr> 
  <tr> 
   <td>Left Outer</td> 
   <td>Not supported</td> 
  </tr> 
  <tr> 
   <td>Right Outer</td> 
   <td>Supported, not stateful</td> 
  </tr> 
  <tr> 
   <td>Full Outer</td> 
   <td>Not supported</td> 
  </tr> 
  <tr> 
   <td>Stream</td> 
   <td>Stream</td> 
   <td>Inner</td> 
   <td>Supported, optionally specify watermark on both sides + time constraints for state cleanup</td> 
  </tr> 
  <tr> 
   <td>Left Outer</td> 
   <td>Conditionally supported, must specify watermark on right + time constraints for correct results, optionally specify watermark on left for all state cleanup</td> 
  </tr> 
  <tr> 
   <td>Right Outer</td> 
   <td>Conditionally supported, must specify watermark on left + time constraints for correct results, optionally specify watermark on right for all state cleanup</td> 
  </tr> 
  <tr> 
   <td>Full Outer</td> 
   <td>Not supported</td> 
  </tr> 
 </tbody> 
</table>

5、处理多个水印的策略

一个流查询可以有多个联合或连接在一起的输入流。每个输入流都可以有不同的延迟数据阈值，对于有状态操作，这些阈值是需要被接纳的，在每个输入流上都可以使用withWatermarks(“eventTime”，delay)指定这些阈值。例如，一个inputStream1和inputStream2之间具有流-流连接的查询。

inputStream1.withWatermark(“eventTime1”, “1 hour”) .join( 
    inputStream2.withWatermark(“eventTime2”, “2 hours”), joinCondition)

在执行查询时，**结构化流单独跟踪每个输入流中的最大事件时间，**根据相应的时延计算水印。S**park还会选择一个全局水印，默认情况下，选择最小的那个作为全局水印，**因为它可以确保，如果其中一个流落后于其他流(例如，其中一个流由于上游故障而停止接收数据)，不会因为太晚而意外地删除数据。换句话说，全局水印将以最慢的流的速度安全地移动，查询输出将相应地延迟。

然而，在某些情况下，我们可能希望获得更快的结果，哪怕这意味着从最慢的流删除数据。从Spark 2.4开始，可以通过设置SQL配置 **spark.sql.streaming.multipleWatermarkPolicy=max(默认值是min)** 来设置最大值来作为全局水印。这使得全局水印以最快的流的速度移动。这种情况，来自较慢流的数据将可能会被删除。因此，需要根据自己的需求谨慎地使用这种配置。

## 6、不支持的操作 ##

有一些DataFrame/Dataset操作是流Dataframe /Dataset不支持的。其中一些如下:

*  流Dataset中还不支持多个流聚合(即流DF上的聚合链)。
 *  流Dataset不支持限制和获取前N行。
 *  不支持流Dataset上的Distinct操作。
 *  只有在聚合之后并在完全输出模式下，流DataSet才支持排序操作。
 *  流式Join操作中有些外部连接操作不能支持，具体参见上篇。

此外，还有一些Dataset方法不能在流Dataset上起作用。它们是立即运行查询并返回结果的操作，这在流Dataset中没有意义。相反，这些功能可以通过显式地启动流查询来实现。

*  count() -不能从流式Dataset中返回单个计数。相反，使用ds.groupBy().count()将返回一个流数据集，其中包含一个正在运行的count。
 *  foreach() -需要改为使用ds.writeStream.foreach(…)。
 *  show()——使用控制台接收器(console sink)代替(参见下一节)。

**如果在流式Dataset/Dataframe上尝试这些操作的话，程序将抛出 *AnalysisException*，**比如“流Dataset/Dataframe不支持XYZ操作”。虽然其中一些可能在Spark的未来版本中得到支持，但还有一些从根本上难以在流数据上有效实现。例如，不支持对输入流进行排序，因为它需要跟踪流中接收的所有数据，而流中的数据是动态增加的。因此，从根本上说，这很难有效地执行。