Stream computing implementation details
Tablestore connects to the micro-batching mode of Spark Structured Streaming through a streaming Source that returns a UUID-based Offset for each batch and keeps channel checkpoints in a Meta table. The following sections explain why this design is required and how a single batch is executed, using the Spark DataSource v1 API as an example.
Micro-batching contract and the seek requirement
In micro-batching mode, Spark Structured Streaming drives a streaming Source through the following calls in each batch:
GetOffset — Retrieves the maximum Offset (EndOffset) that can be read in the current batch.
GetBatch — Retrieves and transforms the data between the StartOffset of the current batch, which is the EndOffset of the previous batch, and the EndOffset.
Custom computing logic — Runs the custom Spark computing logic on the data of the batch.
To satisfy this contract, the streaming API of the upstream database must provide a flexible and accurate seek feature that retrieves the start cursor or end cursor of each partition in real time. Spark uses this information to estimate the Offset in micro-batching mode.
In a distributed NoSQL database, a streaming API that is based on Change Data Capture (CDC) technology works as a continuous stream. Such an API usually does not provide a flexible seek interface, which makes integration with the micro-batching interface of Structured Streaming difficult. If you pull data in advance in the GetOffset phase, you can obtain the expected EndOffset. However, this requires an extra Resilient Distributed Dataset (RDD) computation that must be persisted to a cache, which significantly degrades the performance of the Source.
Design approach: placeholder Offsets and backfilled checkpoints
Because a CDC stream provides no flexible seek interface, the Tablestore Source cannot report a real data position when Spark requests the EndOffset of a batch. Instead of pulling data in advance, the Source returns a placeholder Offset and keeps the actual consumption positions outside the Offset:
Placeholder Offset — For each batch, the Source generates a random UUID string, wraps the UUID into an Offset, and returns that Offset as the EndOffset of the batch. The Offset itself carries no channel position.
Checkpoints in the Meta table — The channel checkpoints that belong to a UUID are stored in the Meta table. The first GetOffset call seeds them with the checkpoints held by the Tunnel server, and the RDD partitions of the batch fill in the latest checkpoints when the batch finishes.
Handover between batches — The EndOffset of one batch is the StartOffset of the next batch. The checkpoints filled in for that UUID are therefore available in the Meta table as the start position of the next batch.
Because the GetOffset phase reads no data, this design avoids the extra RDD computation and cache that pulling data in advance requires. The checkpoints of a batch reach the Tunnel server one batch later, during the Commit call of the following batch.
Core concepts
Channel — A channel is the unit of data consumption in Tablestore Tunnel. Each channel corresponds to a partition of the table.
Checkpoint — A checkpoint is the consumer offset of a channel. In this integration, checkpoints are held in two places: the Tunnel server, which stores the checkpoints that the client has committed, and the Meta table, which stores the checkpoints of each batch.
Offset, StartOffset, and EndOffset — An Offset marks a batch boundary in Spark Structured Streaming. The EndOffset is the maximum Offset that the current batch can read, and it becomes the StartOffset of the next batch. In this integration, an Offset wraps a UUID, and an Offset and a UUID can be converted to each other.
Meta table — The Meta table stores the mapping between UUIDs and checkpoints. Each UUID corresponds to one row, which holds the channel checkpoints of the batch that the UUID identifies.
RDD — An RDD is the most basic data abstraction in Spark. It represents an immutable, partitioned collection of elements that supports parallel computing.
Batch execution sequence
The following figure shows the UML sequence diagram of the solution that integrates Tablestore with Structured Streaming.
In the figure, MicroBatchExecutor is the micro-batching framework of Spark, Source is the abstract interface class of Structured Streaming, SourceRDD is the RDD abstract class of Spark, and TablestoreClient is the Tablestore client. Solid lines indicate detailed operations, and dashed lines indicate successful responses. The numbered steps that follow describe the same interaction in detail.
The sequence runs as a loop in which MicroBatchExecutor drives the Source through the GetOffset, Commit, and GetBatch calls. Each iteration of the loop corresponds to one batch execution in Spark Streaming and proceeds as follows:
GetOffset: determine the EndOffset of the current batch.
MicroBatchExecutor calls the GetOffset method in the Source to retrieve the maximum reachable Offset of the current batch, which is the EndOffset.
The Source generates a random UUID string. The UUID has a one-to-one mapping with the Offset.
If the GetOffset method in the Source is called for the first time, the Source retrieves the channel information of Tablestore. Otherwise, the system skips this step.
Retrieve the checkpoints of all current channels from the Tunnel server of Tablestore.
Persist the retrieved checkpoints of all channels to the Meta table to create a mapping between the UUID and the checkpoints.
Wrap the UUID into an Offset and return the Offset to MicroBatchExecutor as the EndOffset of the current batch.
Commit: persist the checkpoints of the previous batch to the Tunnel server. This step is performed only when the BatchId of the current batch is greater than 0. Otherwise, the system skips this step.
MicroBatchExecutor calls the Commit logic of the Source.
The Source persists the checkpoints that correspond to the EndOffset of the previous batch, which is the StartOffset of the current batch, to the Tunnel server.
In normal cases, the Commit logic does not need to process additional content. This operation serves the following purposes:
Progress visibility — After the checkpoints are persisted to the Tunnel server, the consumption progress can be displayed in real time.
Data ordering — To achieve data ordering, Tunnel maintains a parent-child partition relationship. The client must return the checkpoint of the fully consumed parent partition to the server before a child partition can be loaded.
GetBatch: build the data of the current batch.
MicroBatchExecutor calls the GetBatch operation in the Source based on the StartOffset of the current batch and the EndOffset returned by GetOffset. This call obtains the data of the current batch for the computing logic.
The Source retrieves the real-time checkpoints of the channel list of the Tunnel from the Meta table based on the UUID that corresponds to the StartOffset.
The Source periodically retrieves the checkpoints of the channels from the Tunnel server. Periodic retrieval is required because partition changes in a table may generate new partitions, such as child partitions.
The Source merges the real-time checkpoints retrieved from the Meta table with the checkpoints periodically retrieved from the Tunnel server. The merged result is the checkpoints of all current channels.
The Source constructs the SourceRDD based on the latest checkpoints, the UUID that corresponds to the EndOffset, and other information.
In the SourceRDD, each channel of the Tunnel is bound to one partition of the RDD. Each channel then transforms and processes data in parallel in a distributed manner on Spark execution nodes.
Custom computing logic: process the data of the batch.
Each RDD partition runs the custom Spark computing logic. The logic reads data from the channel that is bound to the partition and updates the checkpoint of that channel in memory.
An RDD partition finishes running in the current batch when a completion condition is met, for example, when a specified number of rows is read or when no new data is available. The partition then persists the latest checkpoint of its channel to the row that corresponds to the UUID in the Meta table. This action fills in the checkpoints for the UUID that corresponds to the EndOffset. After each batch ends, the checkpoints that correspond to the StartOffset of the new batch can be queried in the Meta table.
After the computing logic of all RDD partitions is complete, the Source returns the data within the Offset range of the current batch to MicroBatchExecutor.
BatchId increment: start the next iteration. After all computations of the current batch are complete, the BatchId is incremented and the loop starts again from step 1.