Returns the row-level changes (Change Data Capture) from the specified table as a streaming DataFrame. Currently this API is only supported for Data Source V2 tables whose catalog implements TableCatalog.loadChangelog().

Use .option() to specify the starting version/timestamp and processing options:

spark.readStream
  .option("startingVersion", "10")
  .changes("my_table")

Streaming reads support all of the same post-processing as batch reads -- computeUpdates, deduplicationMode = dropCarryovers, and deduplicationMode = netChanges. The streaming netChanges path holds per-row-identity state in the state store and emits the SPIP collapse output once the global watermark advances past the last _commit_timestamp observed for that row identity. Row identities only touched in the latest observed commit are therefore not emitted until a later commit (with strictly greater _commit_timestamp) advances the watermark past them, or the source terminates.

Streaming netChanges differs from batch netChanges in scope. Batch netChanges collapses all changes for a row identity over the entire requested version range. Streaming netChanges is incremental: it collapses changes that fall within a single watermark window for a row identity (i.e. up to the timer firing that emits its current net result). After a row identity's net result has been emitted, subsequent commits on the same identity start a fresh window and produce additional output rows -- streaming cannot retract previously emitted results to match the batch range-scoped collapse. For a query that observes id=1 inserted at v1 and deleted at v3 with another commit at v2 in between, batch netChanges over [v1..v3] cancels to no row, while streaming emits an insert (when v2 advances the watermark past v1) followed later by a delete (when end-of-stream or another commit advances the watermark past v3).

Because the streaming netChanges path uses transformWithState, the state store provider must be RocksDB. Set spark.sql.streaming.stateStore.providerClass to org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider before starting a streaming netChanges query; the default HDFS-backed provider is rejected at query start.

When the requested options engage post-processing (carry-over removal, update detection, or netChanges), the rewrite injects an internal EventTimeWatermark on _commit_timestamp and a stateful streaming operator (an aggregate for the row-level passes, a transformWithState for netChanges). Two implications follow:

• A commit's events are emitted in the next micro-batch after the commit is read (append-mode aggregate eviction is eventTime <= watermark, and the watermark advances to the max _commit_timestamp observed in the previous batch). A stream that reads its last commit and stops will keep that commit's events in state until a subsequent (no-data) micro-batch fires.
• The query is constrained to Append output mode; Update and Complete are rejected at writer-start time with STREAMING_OUTPUT_MODE.UNSUPPORTED_OPERATION. The internal watermark metadata is stripped from the user-visible _commit_timestamp output, so downstream user-supplied watermarks on other columns do not interact with it via the global multi-watermark policy.

Specifies the input data source format.

Loads input in as a DataFrame, for data streams that read from some path.

Loads input data stream in as a DataFrame, for data streams that don't require a path (e.g. external key-value stores).

Specifies a name for the streaming source. This name is used to identify the source in checkpoint metadata and enables stable checkpoint locations for source evolution.

Adds an input option for the underlying data source.

(Scala-specific) Adds input options for the underlying data source.

Specifies the input schema. Some data sources (e.g. JSON) can infer the input schema automatically from data. By specifying the schema here, the underlying data source can skip the schema inference step, and thus speed up data loading.

Define a Streaming DataFrame on a Table. The DataSource corresponding to the table should support streaming mode.

Loads a CSV file stream and returns the result as a DataFrame.

This function will go through the input once to determine the input schema if inferSchema is enabled. To avoid going through the entire data once, disable inferSchema option or specify the schema explicitly using schema.

You can set the following option(s):

• maxFilesPerTrigger (default: no max limit): sets the maximum number of new files to be considered in every trigger.
• maxBytesPerTrigger (default: no max limit): sets the maximum total size of new files to be considered in every trigger.

You can find the CSV-specific options for reading CSV file stream in <a href="https://spark.apache.org/docs/latest/sql-data-sources-csv.html#data-source-option"> Data Source Option in the version you use.

Loads a JSON file stream and returns the results as a DataFrame.

JSON Lines (newline-delimited JSON) is supported by default. For JSON (one record per file), set the multiLine option to true.

This function goes through the input once to determine the input schema. If you know the schema in advance, use the version that specifies the schema to avoid the extra scan.

You can set the following option(s):

• maxFilesPerTrigger (default: no max limit): sets the maximum number of new files to be considered in every trigger.
• maxBytesPerTrigger (default: no max limit): sets the maximum total size of new files to be considered in every trigger.

You can find the JSON-specific options for reading JSON file stream in <a href="https://spark.apache.org/docs/latest/sql-data-sources-json.html#data-source-option"> Data Source Option in the version you use.

Adds an input option for the underlying data source.

Adds an input option for the underlying data source.

Adds an input option for the underlying data source.

(Java-specific) Adds input options for the underlying data source.

Loads a ORC file stream, returning the result as a DataFrame.

You can set the following option(s):

• maxFilesPerTrigger (default: no max limit): sets the maximum number of new files to be considered in every trigger.
• maxBytesPerTrigger (default: no max limit): sets the maximum total size of new files to be considered in every trigger.

ORC-specific option(s) for reading ORC file stream can be found in Data Source Option in the version you use.

Loads a Parquet file stream, returning the result as a DataFrame.

You can set the following option(s):

• maxFilesPerTrigger (default: no max limit): sets the maximum number of new files to be considered in every trigger.
• maxBytesPerTrigger (default: no max limit): sets the maximum total size of new files to be considered in every trigger.

Parquet-specific option(s) for reading Parquet file stream can be found in Data Source Option in the version you use.

Specifies the schema by using the input DDL-formatted string. Some data sources (e.g. JSON) can infer the input schema automatically from data. By specifying the schema here, the underlying data source can skip the schema inference step, and thus speed up data loading.

Loads text files and returns a DataFrame whose schema starts with a string column named "value", and followed by partitioned columns if there are any. The text files must be encoded as UTF-8.

By default, each line in the text files is a new row in the resulting DataFrame. For example:

// Scala:
spark.readStream.text("/path/to/directory/")

// Java:
spark.readStream().text("/path/to/directory/")

You can set the following option(s):

• maxFilesPerTrigger (default: no max limit): sets the maximum number of new files to be considered in every trigger.
• maxBytesPerTrigger (default: no max limit): sets the maximum total size of new files to be considered in every trigger.

You can find the text-specific options for reading text files in <a href="https://spark.apache.org/docs/latest/sql-data-sources-text.html#data-source-option"> Data Source Option in the version you use.

Loads text file(s) and returns a Dataset of String. The underlying schema of the Dataset contains a single string column named "value". The text files must be encoded as UTF-8.

If the directory structure of the text files contains partitioning information, those are ignored in the resulting Dataset. To include partitioning information as columns, use text.

By default, each line in the text file is a new element in the resulting Dataset. For example:

// Scala:
spark.readStream.textFile("/path/to/spark/README.md")

// Java:
spark.readStream().textFile("/path/to/spark/README.md")

You can set the text-specific options as specified in DataStreamReader.text.

Loads a XML file stream and returns the result as a DataFrame.

This function will go through the input once to determine the input schema if inferSchema is enabled. To avoid going through the entire data once, disable inferSchema option or specify the schema explicitly using schema.

You can set the following option(s):

• maxFilesPerTrigger (default: no max limit): sets the maximum number of new files to be considered in every trigger.
• maxBytesPerTrigger (default: no max limit): sets the maximum total size of new files to be considered in every trigger.

You can find the XML-specific options for reading XML file stream in <a href="https://spark.apache.org/docs/latest/sql-data-sources-xml.html#data-source-option"> Data Source Option in the version you use.