Structured Streaming Functions and Reliability

Functions Supported by Structured Streaming

1. ETL operations on streaming data
2. Schema inference and partitioning of streaming DataFrames or Datasets
3. Operations on the streaming DataFrames or Datasets, including SQL-like operations without types (such as select, where, and groupBy) and RDD operations with types (such as map, filter, and flatMap)
4. Aggregation calculation based on Event Time and processing of delay data
5. Deduplication of streaming data
6. Status computing
7. Stream processing task monitoring
8. Batch-stream join and stream join

   The following table lists the supported join operations.

   Left Table Status	Right Table Status	Supported Join Type	Description
   Static	Static	All types	In stream processing, join operations with no streaming data involved are supported.
   Stream	Static	Inner	Supported, but stateless.
   		Left Outer	Supported, but stateless.
   		Right Outer	Not supported.
   		Full Outer	Not supported.
   Stream	Stream	Inner	Supported. The watermark or time range can be used to clear status of the left and right tables.
   		Left Outer	Conditionally supported. The watermark can be used to clear status of the left table, and watermark and time range must be used for the right table.
   		Right Outer	Conditionally supported. The watermark can be used to clear status of the right table, and watermark and time range must be used for the left table.
   		Full Outer	Not supported.

Functions Not Supported by Structured Streaming

1. Multi-stream aggregation is not supported.
2. Operations such as limit, first, and take are not supported.
3. Distinct data type is not supported.
4. Sorting is supported only when output mode is complete.
5. External connections between streaming and static datasets are supported under certain conditions.
6. Immediate query and result return on some datasets are not supported.
   • count(): Instead of returning a single count from streaming datasets, it uses ds.groupBy().count() to return a streaming dataset containing the running count.
   • foreach(): ds.writeStream.foreach(...) is used to replace it.
   • show(): The output console sink is used to replace it.

Structured Streaming Reliability

Based on the checkpoint and WAL mechanisms, Structured Streaming provides end-to-end exactly-once error tolerance semantics for the sources that can be replayed and the idempotent sinks that support repeated processing.

1. You can enable the checkpoint function by setting option ("checkpointLocation", "checkpoint path") in the program.

   When data is restored from checkpoint, the application or configuration may change. Some changes may result in the failure in restoring data from the checkpoint. The restrictions are as follows:

   1. The number or type of sources cannot be changed.
   2. Whether the source parameter can change depends on the source type and query statement. Typical examples are listed below:
      • You can add, delete, or modify parameters related to rate control. For instance, you can change spark.readStream.format("kafka").option("subscribe", "topic") to spark.readStream.format("kafka").option("subscribe", "topic").option("maxOffsetsPerTrigger", ...).
      • Modifying the topic/file of the consumption can lead to unexpected problems. For instance, changing spark.readStream.format("kafka").option("subscribe", "topic") to spark.readStream.format("kafka").option("subscribe", "newTopic").
   3. The sink type can be changed, and specific sinks can be combined, but it is important to verify the specific scenarios. Here are some notes to keep in mind:
      • File sink can be changed to kafka sink. Kafka processes only new data.
      • Kafka sink cannot be changed to file sink.
      • Kafka sink can be changed to foreach sink, and vice versa.
   4. Whether or not you can change the sink parameter depends on the sink type and query statement. Here are some important notes to consider:
      • The output path of file sink cannot be changed.
      • The output topic of Kafka sink can be changed.
      • The custom operator code in foreach sink can be changed, but the change result depends on the user code.
   5. In certain cases, you can modify operations such as projection, filter, and map-like. Here are some key points to keep in mind:
      • You can add or delete filters in your code. For instance, you can change sdf.selectExpr("a") to sdf.where(...).selectExpr("a").filter(...).
      • If the output schema remains the same, you can modify the projections. For instance, you can change sdf.selectExpr("stringColumn AS json").writeStream to sdf.select(to_json(...).as("json")).writeStream.
      • You can modify the projections in certain conditions if the output schema is different. For instance, changing sdf.selectExpr("a").writeStream to sdf.selectExpr("b").writeStream will not result in an error only if the sink can convert the schema from a to b.
   6. In some scenarios, the status restoration will fail after status is changed.
      • Streaming aggregation: You cannot modify the type or quantity of the grouping key or the aggregation key in the sdf.groupBy("a").agg(...) operation.
      • Streaming deduplication: You cannot modify the type or quantity of the grouping key or the aggregation key in the sdf.dropDuplicates("a") operation.
      • Stream-stream join: You cannot change the schema of the association key or the join type in the sdf1.join(sdf2, ...) operation. Changing other join conditions may result in uncertain outcomes.
      • Any status computing: You cannot modify the schema or timeout type of the user-defined status in operations such as sdf.groupByKey(...).mapGroupsWithState(...) and sdf.groupByKey(...).flatMapGroupsWithState(...). While users can change the state-mapping function, the outcome depends on their code. If schema changes are necessary, users can encode or decode the status data into binary data to enable schema migration.

2. Fault tolerance information of sources

   Sources	Supported Options	Fault Tolerance	Description
   File source	path: file path, which is mandatory. maxFilesPerTrigger: maximum number of files in each trigger. The default value is infinity. latestFirst: whether to process limited number of new files. The default value is false. fileNameOnly: whether the file name is used as the new file for verification instead of using the complete path. The default value is false.	Supported	Paths with wildcard are supported, but multiple paths separated by commas (,) are not supported. Files must be placed in a given directory in atomic mode, which can be implemented through file movement in most file systems.
   Socket Source	host: IP address of the connected node, which is mandatory. port: connected port, which is mandatory.	Not supported	-
   Rate Source	rowsPerSecond: number of rows generated per second. The default value is 1. rampUpTime: rising time before the speed specified by rowsPerSecond is reached. numPartitions: concurrency degree for generating data rows.	Supported	-
   Kafka Source	For details, see https://spark.apache.org/docs/3.1.1/structured-streaming-kafka-integration.html.	Supported	-

3. Fault tolerance information of sinks

   Sinks	Supported Output Mode	Supported Options	Fault Tolerance	Description
   File Sink	Append	Path: Specified file format. This parameter is mandatory. For details, see APIs in DataFrameWriter.	exactly-once	Data can be written to partitioned tables. Time-based partition is better.
   Kafka Sink	Append, Update, Complete	For details, see https://spark.apache.org/docs/3.1.1/structured-streaming-kafka-integration.html.	at-least-once	For details, see https://spark.apache.org/docs/3.1.1/structured-streaming-kafka-integration.html.
   Foreach Sink	Append, Update, Complete	None	Depends on ForeachWriter.	For details, see https://spark.apache.org/docs/3.1.1/structured-streaming-programming-guide.html#using-foreach.
   ForeachBatch Sink	Append, Update, Complete	None	Depends on operators.	For details, see https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#using-foreach-and-foreachbatch.
   Console Sink	Append, Update, Complete	numRows: number of rows printed in each round. The default value is 20. truncate: whether to clear the output when the output is too long. The default value is true.	Not supported	-
   Memory Sink	Append, Complete	None	Not supported. In complete mode, the entire table is rebuilt after the query is restarted.	-