Apache Spark Cache And Streaming Metadata Notes

Apache Spark had a busy week, with 108 commits on master and several changes that matter to data platform teams more than application authors. The main thread is practical: cache layout, Kafka metadata calls, Real Time Mode checkpoint cost, and catalog metadata that needs to be correct when tools read JSON output.

The largest change is the addition of Apache Arrow as a native cache format for in memory Dataset caching. This is not a default change. Spark still uses DefaultCachedBatchSerializer unless a session sets spark.sql.cache.serializer to org.apache.spark.sql.execution.columnar.ArrowCachedBatchSerializer.

That distinction matters for operators. The serializer is a static SQL setting, so it has to be present when the SparkSession is created. It is not a per table toggle. The new docs in docs/sql-arrow-cache-format.md also make a useful point that can prevent bad assumptions: the cached bytes are an internal Arrow RecordBatch payload, not a complete Arrow IPC stream meant for outside readers.

The patch is large because it includes the serializer, cached batch representation, Kryo registration, benchmarks, and a broad suite around primitive values, nested types, nulls, collation, NaN bounds, projection, filter pruning, and compression. The new setting in SQLConf.scala for spark.sql.execution.arrow.cache.prefetch.enabled is off by default. Treat it as a separate test axis, not as part of merely turning on the Arrow cache.

For production ETL jobs, the right read is conservative. This gives teams a way to test Arrow backed cache paths where repeated scans dominate cost. It also adds a new failure surface around unsupported types, memory pressure, compression choice, and cache migration inside long lived JVMs. Benchmark claims in the docs are useful, but cluster shape and data types will decide whether this beats the default serializer.

Structured Streaming Kafka sources gained a small option with a clear operational target. The Kafka offset reader now supports caching partition metadata through partition.metadata.cache.ttl.ms.

The old behavior called DescribeTopics every micro batch through partition assignment discovery. On large topics with short triggers, that is avoidable broker traffic. With a positive TTL, the driver can reuse the assigned TopicPartition set for offset fetches inside that window. The default is -1, so existing jobs do not change.

There is a tradeoff in the option itself. New partitions are not discovered until the TTL expires. That is fine for stable topic layouts and bad for pipelines where partition expansion is part of normal scaling. The validation added in KafkaSourceProvider.scala accepts -1 or a positive millisecond value, and rejects zero or other negative values.

The practical use case is clear:

.option("partition.metadata.cache.ttl.ms", "60000")

Use that only after deciding how long delayed partition discovery is acceptable. A one minute TTL may remove a lot of broker metadata load in high frequency jobs, but it also means partition changes are not instant from the stream point of view.

Real Time Mode picked up a checkpoint path change that fits its latency goal. Async progress tracking is now supported for stateless Real Time Mode queries, and the docs no longer describe the combination as unsupported.

The useful part is not just that asyncProgressTrackingEnabled=true is accepted. The change makes async progress tracking the default for stateless Real Time Mode queries, gated by an internal SQL config, while the writer option still wins when set. The patch also splits Real Time Mode offset handling out of MicroBatchExecution so the async subclass can reuse it. That is mostly internal refactor, but it reduces the chance that the two execution paths drift.

The update in docs/streaming/real-time-mode.md states the important operator behavior: offset and commit logs are written outside the record processing path, and every batch is checkpointed in Real Time Mode. If async log writes fail, the code is meant to fail fast rather than hide checkpoint damage behind continued processing.

The boundary still matters. This is for stateless Real Time Mode. Stateful support remains out of scope in the current docs, so low latency jobs with aggregation, dedupe, stream joins, or transformWithState still need different planning.

The most recent commit makes SHOW TABLE EXTENDED AS JSON line up with V2 metadata. The JSON output now uses table summary types instead of hard coding every extended entry as TABLE. That matters if a catalog contains views and external relation types and some inventory job consumes the JSON form rather than the tabular command output.

The changed command path is ShowTablesJsonCommand.scala. For a relation catalog, Spark now pulls relation summaries and uses tableType(). For a plain table catalog, it uses table summaries when extended output is requested. That is the right place to fix it because downstream scanners should not have to infer object type from names or follow up calls.

There were also correctness additions around semi structured data. The new try_variant_array_append function adds a null returning counterpart to variant_array_append when a path segment hits the wrong type or the target is not an array. The Python entry point landed in python/pyspark/sql/functions/builtin.py, with matching SQL and Scala registration.

For ingestion code, this is a useful shape. Bad JSON shape can become a nullable expression result instead of a failed query, while malformed paths and size limit violations still throw. In the same week, XML schema inference stopped skipping empty values to avoid data loss during inference. That is the kind of small fix that can change downstream schema expectations for feeds with sparse XML fields.

  • Test Arrow cache with the exact data types and compression settings used by real jobs. The new serializer is promising, but it is a session wide cache decision.
  • Set partition.metadata.cache.ttl.ms only for Kafka streams where delayed partition discovery is acceptable. Broker relief and discovery latency move in opposite directions.
  • Watch Real Time Mode changes if you operate low latency stateless streams. Async checkpoint writes reduce path cost, but checkpoint storage failures should now surface quickly.