# structured streaming功能与可靠性介绍
#### Structured Streaming支持的功能
1. 支持对流式数据的ETL操作。
2. 支持流式DataFrames或Datasets的schema推断和分区。
3. 流式DataFrames或Datasets上的操作：包括无类型，类似SQL的操作（比如select、where、groupBy），以及有类型的RDD操作（比如map、filter、flatMap）。
4. 支持基于Event Time的聚合计算，支持对迟到数据的处理。
5. 支持对流式数据的去除重复数据操作。
6. 支持状态计算。
7. 支持对流处理任务的监控。
8. 支持批流join，流流join。 当前JOIN操作支持列表如下：
   
   | 左表     | 右表     | 支持的Join类型   | 说明                                                 |
   |:---|:---|:---|:---|
   | Static | Static | 全部类型        | 即使在流处理中，不涉及流数据的join操作也能全部支持                        |
   | Stream | Static | Inner       | 支持，但是无状态                                           |
   | Stream | Static | Left Outer  | 支持，但是无状态                                           |
   | Stream | Static | Right Outer | 不支持                                                |
   | Stream | Static | Full Outer  | 不支持                                                |
   | Stream | Stream | Inner       | 支持，左右表可选择使用watermark或者时间范围进行状态清理                   |
   | Stream | Stream | Left Outer  | 有条件的支持，左表可选择使用watermark进行状态清理，右表必须使用watermark+时间范围 |
   | Stream | Stream | Right Outer | 有条件的支持，右表可选择使用watermark进行状态清理，左表必须使用watermark+时间范围 |
   | Stream | Stream | Full Outer  | 不支持                                                |
      
   
 
#### Structured Streaming不支持的功能
1. 不支持多个流聚合。
2. 不支持limit、first、take这些取N条Row的操作。
3. 不支持Distinct。
4. 只有当output mode为complete时才支持排序操作。
5. 有条件地支持流和静态数据集之间的外连接。
6. 不支持部分DataSet上立即运行查询并返回结果的操作：
   - count()：无法从流式Dataset返回单个计数，而是使用ds.groupBy().count()返回一个包含运行计数的streaming Dataset。
   
   - foreach()：使用ds.writeStream.foreach(...)代替。
   
   - show()：使用输出console sink代替。
    
 
#### Structured Streaming可靠性说明
Structured Streaming通过checkpoint和WAL机制，对可重放的sources，以及支持重复处理的幂等性sinks，可以提供端到端的exactly-once容错语义。
1. 用户可在程序中设置option("checkpointLocation", "checkpoint路径")启用checkpoint。 从checkpoint恢复时，应用程序或者配置可能发生变更，有部分变更会导致从checkpoint恢复失败，具体限制如下：
   1. 不允许source的个数或者类型发生变化。
   
   2. source的参数变化，这种情况是否能被支持，取决于source类型和查询语句，例如：
      - 速率控制相关参数的添加、删除和修改，此种情况能被支持，如：spark.readStream.format("kafka").option("subscribe", "topic")变更为spark.readStream.format("kafka").option("subscribe", "topic").option("maxOffsetsPerTrigger", ...)
      
      - 修改消费的topic/files可能会出现不可预知的问题，如：spark.readStream.format("kafka").option("subscribe", "topic")变更为spark.readStream.format("kafka").option("subscribe", "newTopic")
       
   
   3. sink的类型发生变化：允许特定的几个sink的组合，具体场景需要验证确认，例如：
      - File sink允许变更为kafka sink，kafka中只处理新数据。
      
      - kafka sink不允许变更为file sink。
      
      - kafka sink允许变更为foreach sink，反之亦然。
       
   
   4. sink的参数变化，这种情况是否能被支持，取决于sink类型和查询语句，例如：
      - 不允许file sink的输出路径发生变更。
      
      - 允许Kafka sink的输出topic发生变更。
      
      - 允许foreach sink中的自定义算子代码发生变更，但是变更结果取决于用户代码。
       
   
   5. Projection、filter和map-like操作变更，局部场景下能够支持，例如：
      - 支持Filter的添加和删除，如：sdf.selectExpr("a")变更为sdf.where(...).selectExpr("a").filter(...)
      
      - Output schema相同时，projections允许变更，如：sdf.selectExpr("stringColumn AS json").writeStream变更为sdf.select(to_json(...).as("json")).writeStream
      
      - Output schema不相同时，projections在部分条件下允许变更，如：sdf.selectExpr("a").writeStream变更为sdf.selectExpr("b").writeStream，只有当sink支持"a"到"b"的schema转换时才不会出错。
       
   
   6. 状态操作的变更，在部分场景下会导致状态恢复失败：
      - Streaming aggregation：如sdf.groupBy("a").agg(...)操作中，不允许分组键或聚合键的类型或者数量发生变化。
      
      - Streaming deduplication：如：sdf.dropDuplicates("a")操作中，不允许分组键或聚合键的类型或者数量发生变化。
      
      - Stream-stream join：如sdf1.join(sdf2, ...)操作中，关联键的schema不允许发生变化，join类型不允许发生变化，其他join条件的变更可能导致不确定性结果。
      
      - 任意状态计算：如sdf.groupByKey(...).mapGroupsWithState(...)或者sdf.groupByKey(...).flatMapGroupsWithState(...)操作中，用户自定义状态的schema或者超时类型都不允许发生变化；允许用户自定义state-mapping函数变化，但是变更结果取决于用户代码；如果需要支持schema变更，用户可以将状态数据编码/解码成二进制数据以支持schema迁移。
       
    

2. Source的容错性支持列表
   
   | Sources       | 支持的Options                                                                                                                                                                                                                                | 容错支持 | 说明                                                                                                |
   |:---|:---|:---|:---|
   | File source   | path：必填，文件路径 maxFilesPerTrigger：每次trigger最大文件数（默认无限大） latestFirst：是否有限处理新文件（默认值: false） fileNameOnly：是否以文件名作为新文件校验，而不是使用完整路径进行判断（默认值: false） | 支持   | 支持通配符路径，但不支持以逗号分隔的多个路径。 文件必须以原子方式放置在给定的目录中，这在大多数文件系统中可以通过文件移动操作实现。 |
   | Socket Source | host：连接的节点ip，必填 port：连接的端口，必填                                                                                                                                                                              | 不支持  | -                                                                                                 |
   | Rate Source   | rowsPerSecond：每秒产生的行数，默认值1 rampUpTime：在达到rowsPerSecond速度之前的上升时间 numPartitions：生成数据行的并行度                                                                                    | 支持   | -                                                                                                 |
   | Kafka Source  | 参见<https://archive.apache.org/dist/spark/docs/3.3.1/structured-streaming-kafka-integration.html>                                                                                                                                          | 支持   | -                                                                                                 |
      
   
3. Sink的容错性支持列表
   
   | Sinks             | 支持的output模式              | 支持Options                                                                                         | 容错性                              | 说明                                                                                                                  |
   |:---|:---|:---|:---|:---|
   | File Sink         | Append                   | Path：必须指定 指定的文件格式，参见DataFrameWriter中的相关接口                           | exactly-once                     | 支持写入分区表，按时间分区用处较大                                                                                                   |
   | Kafka Sink        | Append, Update, Complete | 参见：<https://archive.apache.org/dist/spark/docs/3.3.1/structured-streaming-kafka-integration.html> | at-least-once                    | 参见<https://archive.apache.org/dist/spark/docs/3.3.1/structured-streaming-kafka-integration.html>                    |
   | Foreach Sink      | Append, Update, Complete | None                                                                                              | 依赖于ForeachWriter实现               | 参见<https://archive.apache.org/dist/spark/docs/3.3.1/structured-streaming-programming-guide.html#using-foreach>      |
   | ForeachBatch Sink | Append, Update, Complete | None                                                                                              | 依赖于算子实现                          | 参见<https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#using-foreach-and-foreachbatch> |
   | Console Sink      | Append, Update, Complete | numRows：每轮打印的行数，默认20 truncate：输出太长时是否清空，默认true                     | 不支持容错                            | -                                                                                                                   |
   | Memory Sink       | Append, Complete         | None                                                                                              | 不支持容错，在complete模式下，重启query会重建整个表 | -                                                                                                                   |