Migration Guide: Structured Streaming

Note that this migration guide describes the items specific to Structured Streaming. Many items of SQL migration can be applied when migrating Structured Streaming to higher versions. Please refer Migration Guide: SQL, Datasets and DataFrame.

Upgrading from Structured Streaming 4.3 to 4.4

• Since Spark 4.4, streaming queries using EXCEPT with a streaming left input are rejected because their optimizer rewrites introduce streaming aggregations after the initial unsupported-operation checks. Existing queries can still restart from their persistent checkpoints using the compatibility value recorded in the offset log.

• Since Spark 4.4, DataStreamWriter.toTable writes to an existing V1 table with the partitioning of the table. A query whose partitionBy does not match the partitioning of the table, including the order of the columns, fails to start. A query that already committed files to the table in another layout, for example without partitionBy to a partitioned table, fails to start with STREAMING_FILE_SINK_PARTITIONING_MISMATCH. As a result, such existing queries fail to restart. To recover, write to a new table location with a new checkpoint location, or rewrite the committed files with the partition columns of the table.

Upgrading from Structured Streaming 4.2 to 4.3

• Since Spark 4.3, in Spark Connect the DataFrame passed to a Python foreachBatch function is bound to the streaming query’s own session, a clone of the session that started the query, matching classic foreachBatch. This runs the batch under the configuration the streaming engine pins on the clone rather than the root session’s. As a result, the batch DataFrame can no longer be combined with the root session inside the function: a temporary view created from it is not visible to the root session, referencing it from spark.sql on the root session fails with DATAFRAME_NOT_FOUND, and joining it with a DataFrame captured from the root session fails with SESSION_NOT_SAME. Run such work through the batch DataFrame’s own session (batch_df.sparkSession) instead, and pass results out through a table or a global temporary view rather than a session-local temporary view. To restore the previous behavior, set spark.sql.connect.streaming.foreachBatch.useClonedSession to false.

• Since Spark 4.3, a streaming query with a file stream sink validates its partition columns at start like batch file writes. The query fails with INVALID_PARTITION_COLUMN_DATA_TYPE if a partition column has a type such as ARRAY, MAP, STRUCT, or a string with a non-default collation like UTF8_LCASE, with COLUMN_ALREADY_EXISTS if the partition columns are duplicated, and with ALL_PARTITION_COLUMNS_NOT_ALLOWED if all columns are partition columns. As a result, existing queries with such partition columns fail to restart. For the first case, change the type of the partition column without renaming it to keep the directory layout. For the other cases, change the partition columns and use a new checkpoint location and output path.

Upgrading from Structured Streaming 4.1 to 4.2

• Since Spark 4.2, restarting a streaming query from a checkpoint whose metadata file is missing while the offset or commit logs contain data fails with STREAMING_CHECKPOINT_MISSING_METADATA_FILE, instead of silently generating a new query ID (which can duplicate data in exactly-once sinks). Restore the metadata file or use a new checkpoint location. To restore the previous behavior, set spark.sql.streaming.checkpoint.verifyMetadataExists.enabled to false. (See SPARK-55058 for more details.)

Upgrading from Structured Streaming 4.0 to 4.1

• Since Spark 4.1, AQE is supported for stateless workloads, and it could affect the behavior of the query after upgrade (especially since AQE is turned on by default). In general, it helps to achieve better performance including resolution of skewed partition, but you can turn off AQE via changing spark.sql.adaptive.streaming.stateless.enabled to false to restore the behavior if you see regression.

Upgrading from Structured Streaming 3.5 to 4.0

• Since Spark 4.0, Spark falls back to single batch execution if any source in the query does not support Trigger.AvailableNow. This is to avoid any possible correctness, duplication, and dataloss issue due to incompatibility between source and wrapper implementation. (See SPARK-45178 for more details.)
• Since Spark 4.0, new configuration spark.sql.streaming.ratioExtraSpaceAllowedInCheckpoint (default: 0.3) controls the amount of additional space allowed in the checkpoint directory to store stale version files for batch deletion inside maintenance task. This is to amortize the cost of listing in cloud store. Setting this to 0 defaults to the old behavior. (See SPARK-48931 for more details.)
• Since Spark 4.0, when relative path is used to output data in DataStreamWriter the resolution to absolute path is done in the Spark Driver and is not deferred to Spark Executor. This is to make Structured Streaming behavior similar to DataFrame API (DataFrameWriter). (See SPARK-50854 for more details.)

Upgrading from Structured Streaming 3.3 to 3.4

• Since Spark 3.4, Trigger.Once is deprecated, and users are encouraged to migrate from Trigger.Once to Trigger.AvailableNow. Please refer SPARK-39805 for more details.

• Since Spark 3.4, the default value of configuration for Kafka offset fetching (spark.sql.streaming.kafka.useDeprecatedOffsetFetching) is changed from true to false. The default no longer relies consumer group based scheduling, which affect the required ACL. For further details please see Structured Streaming Kafka Integration.

Upgrading from Structured Streaming 3.2 to 3.3

• Since Spark 3.3, all stateful operators require hash partitioning with exact grouping keys. In previous versions, all stateful operators except stream-stream join require loose partitioning criteria which opens the possibility on correctness issue. (See SPARK-38204 for more details.) To ensure backward compatibility, we retain the old behavior with the checkpoint built from older versions.

Upgrading from Structured Streaming 3.0 to 3.1

• In Spark 3.0 and before, for the queries that have stateful operation which can emit rows older than the current watermark plus allowed late record delay, which are “late rows” in downstream stateful operations and these rows can be discarded, Spark only prints a warning message. Since Spark 3.1, Spark will check for such queries with possible correctness issue and throw AnalysisException for it by default. For the users who understand the possible risk of correctness issue and still decide to run the query, please disable this check by setting the config spark.sql.streaming.statefulOperator.checkCorrectness.enabled to false.

• In Spark 3.0 and before Spark uses KafkaConsumer for offset fetching which could cause infinite wait in the driver. In Spark 3.1 a new configuration option added spark.sql.streaming.kafka.useDeprecatedOffsetFetching (default: true) which could be set to false allowing Spark to use new offset fetching mechanism using AdminClient. For further details please see Structured Streaming Kafka Integration.

Upgrading from Structured Streaming 2.4 to 3.0

• In Spark 3.0, Structured Streaming forces the source schema into nullable when file-based datasources such as text, json, csv, parquet and orc are used via spark.readStream(...). Previously, it respected the nullability in source schema; however, it caused issues tricky to debug with NPE. To restore the previous behavior, set spark.sql.streaming.fileSource.schema.forceNullable to false.

• Spark 3.0 fixes the correctness issue on Stream-stream outer join, which changes the schema of state. (See SPARK-26154 for more details). If you start your query from checkpoint constructed from Spark 2.x which uses stream-stream outer join, Spark 3.0 fails the query. To recalculate outputs, discard the checkpoint and replay previous inputs.

• In Spark 3.0, the deprecated class org.apache.spark.sql.streaming.ProcessingTime has been removed. Use org.apache.spark.sql.streaming.Trigger.ProcessingTime instead. Likewise, org.apache.spark.sql.execution.streaming.continuous.ContinuousTrigger has been removed in favor of Trigger.Continuous, and org.apache.spark.sql.execution.streaming.OneTimeTrigger has been hidden in favor of Trigger.Once.