Structured Streaming practice questions

Apply the distinct() transformation on the entire streaming DataFrame prior to writing the output to the Delta table.

Why it fails: Calling distinct() on an unbounded streaming DataFrame requires Spark to maintain an infinite state of all previously seen records to check for duplicates, which will eventually cause the streaming job to fail due to out-of-memory errors.

Use the dropDuplicates() method with a set of unique identifier columns without specifying a watermark.

Why it fails: Unbounded dropDuplicates() operations accumulate state indefinitely as new data arrives, mirroring the memory growth issues of distinct(), making it unsuitable for long-running production streaming applications processing continuous streams of data.

Enable Delta Lake change data feed and configure the sink to automatically deduplicate incoming records at the storage layer.

Why it fails: Delta Lake change data feed tracks row-level changes for downstream consumers rather than automatically deduplicating incoming streaming micro-batches, meaning duplicate records from upstream sources will still be persisted unless handled in the Spark logic.

Change the output mode from append to complete mode while retaining the current aggregation logic.

Why it fails: Complete mode rewrites the entire result table to the sink on every trigger. While it bypasses certain watermark requirements for append streams, it is not supported with the append output mode and does not solve the memory management problem for large stateful workloads.

Increase the spark.sql.shuffle.partitions configuration property to a larger value to handle the streaming state distribution.

Why it fails: Shuffle partitions control the parallelism of aggregation and join operations, but they do not provide the event-time threshold required by the Spark SQL engine to execute stateful streaming aggregations in append mode.

Switch the streaming sink format from delta to memory to execute local iterative calculations without external storage constraints.

Why it fails: The memory sink is designed primarily for debugging and testing purposes in interactive notebooks. Using it does not eliminate the fundamental requirement for a watermark when performing stateful streaming operations.

spark.databricks.delta.optimizeWrite.enabled

Why it fails: This specialized property optimizes write performance by automatically coalescing small files prior to writing into Delta storage tables. However, it does not manage or control the shuffle partition count generated during stateful stream aggregations.

spark.streaming.concurrentJobs

Why it fails: This configuration controls the maximum number of streaming jobs that can be executed concurrently within a single SparkSession instance. It manages scheduling priority rather than altering the shuffle partitioning topology for aggregation transformations.

spark.databricks.streaming.sql.maxFilesPerTrigger

Why it fails: This parameter restricts the maximum number of files consumed in a single micro-batch when reading from file-based data sources. While useful for rate-limiting file ingestion speeds, it has no impact on shuffle partition counts during stateful operations.

Update mode

Why it fails: Update mode only outputs rows that have been updated since the last trigger interval. It does not output rows that remain unchanged or the entire result set, which violates the requirement for complete updates of all historical aggregate values.

Append mode

Why it fails: Append mode only appends new rows to the result table and cannot be used with streaming aggregations unless watermarking is applied. Even with watermarking, it only outputs finalized rows and cannot rewrite existing aggregate results.

Overwrite mode

Why it fails: Overwrite mode is a batch-oriented save operation that replaces existing data entirely. It is not a valid output mode for Structured Streaming queries, as streaming pipelines rely on incremental micro-batch execution models.

The query was started using spark.read() instead of spark.readStream() for the file path.

Why it fails: Using spark.read() executes a static batch read that processes existing files once and terminates immediately. While it would not process new files, it is a fundamental syntax error rather than a silent failure of file detection.

The trigger interval was set to process data once using .trigger(Once=True) instead of a continuous time duration.

Why it fails: Trigger availableNow or Once processes all available files in micro-batches and then terminates the query. While it stops after processing, it does not ignore newly added files during its initial execution phase.

The output mode was set to Append instead of Complete when reading file streams.

Why it fails: Append mode is the default and correct output mode for file-based streaming sources that write to an append-only Delta table. Output modes govern writing behavior, not file discovery or ingestion mechanics.

Restart the streaming query after running an ALTER TABLE command manually to add the new nested field to the Delta table schema.

Why it fails: Manual intervention through explicit ALTER TABLE commands is slow and prone to human error in production environments. Structured Streaming requires automated mechanisms to handle rapid schema changes from high-throughput streaming sources without manual pipeline downtime.

Configure the Kafka source reader option 'spark.sql.streaming.schemaInference' to true to automatically update downstream consumers.

Why it fails: The spark.sql.streaming.schemaInference configuration is not a valid option for the Kafka data source in Spark. Kafka treats messages as binary byte arrays, so schema inference must be handled explicitly through schema definitions or Spark SQL functions.

Cast the incoming JSON payload into a predefined StringType column and store it raw, bypassing the schema validation checks.

Why it fails: Casting the entire payload to a string destroys structural queryability and pushes downstream parsing complexity to analytics consumers. It defeats the primary benefit of structured ingestion layers designed for immediate querying and transformation.

Delete the entire checkpoint directory and restart the streaming query with a new starting offset configuration.

Why it fails: Deleting the entire checkpoint directory destroys all state stores and historical offset metadata. This forces the streaming query to treat the source as brand new, which breaks stateful aggregations and causes data duplication or loss of state.

Modify the Kafka source options to use startingOffsets='earliest' while pointing to the existing checkpoint location.

Why it fails: When a checkpoint location is specified, Spark ignores source options like startingOffsets because it relies on the committed offsets stored inside the checkpoint directory to determine where to resume processing.

Switch the output mode from append to complete mode to force the pipeline to reprocess all historical data from scratch.

Why it fails: Complete mode requires the entire table to be outputted every micro-batch, which is unsupported for many sinks and does not solve offset recovery issues. Output modes dictate how state results are written, not how source offsets are tracked.

spark.readStream.option('processAll', 'true')

Why it fails: There is no built-in Spark Structured Streaming option named 'processAll'. Using non-existent options will result in a runtime exception during the query initialization phase. Developers must stick to documented configuration keys to ensure the Spark session correctly interprets the read request for the file-based streaming source integration.

spark.readStream.option('initialBatch', 'all')

Why it fails: While 'initialBatch' sounds intuitive, it is not a valid option for the Structured Streaming file source. Providing incorrect options will not trigger the desired behavior and may cause the query to fail. Only the officially supported options, such as includeExistingFiles, should be used for managing initial data ingestion.

spark.readStream.option('maxFilesPerTrigger', -1)

Why it fails: Setting maxFilesPerTrigger to -1 controls the throughput of the stream by allowing an unlimited number of files per batch, but it does not change the behavior of ignoring files already present. It only dictates how many files are processed per trigger interval, not which files are selected for ingestion.

Append

Why it fails: Append mode is generally not supported for queries involving aggregations because aggregations update results over time. Since Append mode only allows adding new rows, it cannot handle the updates required when previous aggregation results change due to late-arriving data. Exceptions are thrown if attempted without a watermark.

Overwrite

Why it fails: Overwrite is not a standard output mode for Structured Streaming queries. It is a concept often used in batch writes to replace existing data. Using Overwrite in a streaming context will result in a syntax error, as the streaming framework manages state incrementally rather than replacing entire datasets on disk.

Merge

Why it fails: Merge is not a valid output mode for Structured Streaming. While Delta Lake supports MERGE operations for batch updates, Structured Streaming output sinks require Append, Complete, or Update modes to correctly handle the continuous flow of data and state updates associated with streaming transformations across the Spark cluster.

Change trigger to 'AvailableNow(true)'.

Why it fails: AvailableNow is designed to process all currently available data and then shut down. While it provides low latency for the current batch, it does not keep the stream running continuously. It is intended for scheduled batch processing of streaming data, not for long-running 24/7 streaming application requirements.

Set trigger to 'None'.

Why it fails: Setting the trigger to None or passing an empty trigger will not work as expected in most Spark APIs. The trigger object requires a defined interval type. Using 'None' will likely result in a compilation error or a runtime error because the streaming engine requires a defined strategy for execution.

Set trigger to 'ProcessingTime('0 seconds')'.

Why it fails: Passing '0 seconds' to ProcessingTime is often interpreted as an invalid argument or can cause unpredictable behavior in the micro-batch scheduler. The correct way to achieve the fastest execution is to either omit the trigger or use a specific continuous processing mode if low latency is critical.

The query will execute but results will be delayed by 24 hours.

Why it fails: Spark does not automatically delay results for 24 hours. If the join requirements are not met, the query will fail immediately with an AnalysisException during the planning phase. It does not attempt to run with an invalid, infinite-state configuration because that would guarantee resource exhaustion and system failure.

The join will automatically default to an inner join with no state limit.

Why it fails: Spark does not default to an unbounded state join because it would lead to inevitable memory issues. The engine is architected to be robust, and it explicitly prohibits joins that lack the necessary watermarking or time-range constraints required for state management during the query plan compilation stage.

Only the inner join is supported in stream-stream joins.

Why it fails: Spark supports various join types, including inner and outer joins, for stream-stream scenarios. The restriction is not on the join type itself but on the necessity of including event-time constraints. This ensures that even outer joins remain performant and memory-efficient by allowing the state store to be cleaned periodically.

It is only available in Python.

Why it fails: The foreachBatch method is available across all supported Spark languages, including Scala, Java, and Python. It is a fundamental feature of the streaming API, not limited by language, and provides a powerful interface for executing arbitrary batch logic on each micro-batch produced by the streaming engine during execution.

It ignores the checkpointing mechanism.

Why it fails: ForeachBatch respects the checkpointing mechanism; Spark still manages the offsets. If the function inside fails, the batch will be retried based on the checkpointed state. Therefore, it does not ignore checkpointing; rather, it relies on it to ensure that the batch processing remains consistent and reliable throughout the lifecycle.

Spark automatically updates the sink schema.

Why it fails: Spark does not automatically modify the sink schema to match the incoming data. Doing so would be dangerous for downstream consumers that expect a specific structure. The developer must explicitly handle schema evolution, typically by using features like Delta Lake's schema evolution options or manual schema migration techniques.

Spark ignores the extra fields and processes the rest.

Why it fails: Spark does not simply ignore schema mismatches. Depending on the source and configuration, it will either throw an error or populate the fields with nulls if the schema is incompatible. Relying on silent truncation is not a default behavior, and you cannot assume the process will continue successfully without configuration.

The stream automatically restarts with the new schema.

Why it fails: There is no automatic mechanism for a stream to detect a schema change and restart with a new configuration. A schema change is a structural shift that requires human intervention or explicit logic to handle the evolution, preventing unpredictable behavior in the pipeline which could compromise downstream data integrity.

Querying the system log files via SSH.

Why it fails: SSH access to driver nodes is generally restricted or not recommended in Databricks environments. Monitoring should be done through managed interfaces like the Spark UI, Query Progress metrics, or platform-native logging tools, not by manually inspecting log files on individual nodes, which is neither efficient nor scalable for production.

Manually counting rows in the sink table.

Why it fails: Counting rows in the sink is an extremely expensive operation that does not provide insights into the streaming engine's health. It is not a valid monitoring technique, as it doesn't indicate latency, state issues, or processing bottlenecks, and it can significantly degrade the performance of the storage layer itself.

Restarting the cluster every hour.

Why it fails: Restarting the cluster is a maintenance action, not a monitoring strategy. It would interrupt the streaming query and force it to re-process data from the checkpoint, which is completely counter-productive to maintaining low-latency, 24/7 streaming operations and does not help in understanding the performance metrics of the query.

The record is processed and stored in the state store.

Why it fails: Records falling behind the watermark threshold are strictly ignored to maintain state consistency. Processing them would violate the defined latency constraints and potentially cause memory pressure by forcing the engine to maintain state for data that is considered 'too late' for downstream business logic requirements.

The record is sent to the dead-letter queue.

Why it fails: Structured Streaming does not automatically route dropped records to a dead-letter queue. Unless a custom implementation is built using foreachBatch or specific sink configurations, the engine silently drops these records. This behavior is designed to prioritize performance and deterministic streaming output over data lineage for late records.

The record triggers an immediate watermark update.

Why it fails: Watermarks are advanced based on the maximum observed event time in the stream, not by individual late records. An event that is already behind the threshold cannot influence the progression of the watermark, as the watermark only moves forward when the window of potential data arrival passes.

It supports stateful streaming operations.

Why it fails: Stateful operations like mapGroupsWithState cannot be performed inside foreachBatch. The function receives a static DataFrame representing a micro-batch, not a streaming DataFrame. Consequently, the stream's stateful context is not available for these batch-level transformations, limiting it to stateless operations within the provided dataframe scope.

It automatically handles checkpointing for all logic.

Why it fails: While the streaming query itself uses checkpoints to track progress, the code inside foreachBatch is executed by the user. If an operation fails inside the function, the checkpoint will not automatically handle partial writes or data inconsistencies. Developers must ensure their batch-level logic is idempotent and robust.

It requires the query to be in Complete mode.

Why it fails: foreachBatch is independent of output modes. It can be used regardless of whether the underlying query uses Append, Update, or Complete mode, as it simply intercepts the output DataFrame of each micro-batch. This makes it highly versatile for various streaming architectural requirements and custom processing needs.

It specifies where the actual output data is saved.

Why it fails: The checkpointLocation only stores metadata and offsets related to the streaming state. The actual output data is directed to the path specified in the writeStream configuration, such as a Delta table or parquet directory, and not to the checkpoint folder itself.

It defines the frequency of data writes to the sink.

Why it fails: The frequency of data writes is determined by the trigger interval, which is explicitly set to '5 seconds' in the configuration. The checkpointLocation has no control over the timing or frequency of output operations, as it is strictly used for tracking metadata and recovery information.

It forces the query to use the Complete output mode.

Why it fails: The checkpointLocation is required for all streaming queries with stateful operations, regardless of the output mode. It does not influence whether the output mode is Append, Update, or Complete; those are separate settings configured within the writeStream() operation of the Spark DataFrame API.

The function passed to `foreachBatch` is executed on the driver, so it can access driver-local variables and broadcast them.

Why it fails: The function passed to `foreachBatch` runs on the driver, but it cannot directly access driver-local variables unless they are serializable and broadcast. Actually, the function is executed on the driver, but any operations inside are distributed. However, accessing driver-local variables is possible but must be handled carefully. This statement is misleading because it implies unrestricted access, which is not true due to serialization constraints.

`foreachBatch` supports continuous processing mode, allowing sub-millisecond latency.

Why it fails: `foreachBatch` is not supported in continuous processing mode. Continuous processing is an experimental feature that supports only a subset of operations, and `foreachBatch` is not one of them. This option incorrectly associates `foreachBatch` with continuous processing, which has different latency characteristics and limitations.

`foreachBatch` guarantees exactly-once processing when writing to an external database without additional idempotent logic.

Why it fails: `foreachBatch` does not guarantee exactly-once by itself. If the external database does not support transactions or idempotent writes, duplicates can occur on retries. The engineer must implement idempotent logic, such as using a unique batch ID to deduplicate. The checkpoint ensures the batch is retried, but the sink must handle it.

`trigger(continuous='1 second')`

Why it fails: Continuous processing is an experimental feature that provides low-latency processing but does not automatically stop after processing available data. It runs continuously until stopped. It also has limited operation support. This trigger does not meet the requirement of stopping after processing all data.

`trigger(availableNow=True)`

Why it fails: `availableNow` trigger processes all available data in multiple micro-batches and then stops. While it also stops automatically, it is designed to process data in multiple batches to avoid large single batches. The question asks for processing all available data and then stopping; both `once` and `availableNow` do this, but `once` processes in a single batch. However, `availableNow` is the recommended replacement for `once` in newer versions. The question does not specify version, but `once` is the direct answer for a single batch. Actually, `availableNow` is more efficient for large datasets. The stem says 'process all available data and then stop automatically', which both do. But `once` is the classic answer. However, to avoid ambiguity, the correct answer is `once` because it is explicitly designed for one-time processing. In newer Spark, `availableNow` is preferred. But the exam may expect `once` for this scenario. Given the options, `once` is correct.

`trigger(processingTime='10 seconds')`

Why it fails: A processing time trigger runs the query every 10 seconds indefinitely until stopped manually. It does not automatically terminate after processing available data. This trigger is for continuous processing at fixed intervals, not for one-time batch-like execution.

Append mode

Why it fails: Append mode only emits new rows that are finalized and will never change. For an aggregation like `groupBy("accountId").count()`, a row for an account is only finalized when the watermark passes and no further updates are expected. Accounts with no new transactions in the current micro-batch would not be re-emitted, so you would not see their updated counts. Therefore, append mode does not satisfy the requirement.

Complete mode

Why it fails: Complete mode emits the entire result table after every micro-batch, including all accounts and their current counts. While this would show updated counts for all accounts, it also re-emits accounts that had no new transactions. The requirement is to emit updated counts as new data arrives, not to re-emit the full result set each time. Complete mode can be inefficient for large state and is not the intended mode for incremental updates.

Delta mode

Why it fails: Delta mode is not a valid output mode in Structured Streaming. The supported output modes are append, update, and complete. Delta Lake is a storage format that can be used as a sink, but it does not define an output mode. Choosing this option reflects a misunderstanding of the distinction between sink formats and output modes, and it would not compile or run as a streaming output mode.