@InterfaceStability.Evolving public interface ContinuousReader extends org.apache.spark.sql.execution.streaming.BaseStreamingSource, DataSourceReader
DataSourceReader
. Data source readers can implement this
interface to allow reading in a continuous processing mode stream.
Implementations must ensure each partition reader is a ContinuousInputPartitionReader
.
Note: This class currently extends BaseStreamingSource
to maintain compatibility with
DataSource V1 APIs. This extension will be removed once we get rid of V1 completely.Modifier and Type | Method and Description |
---|---|
void |
commit(Offset end)
Informs the source that Spark has completed processing all data for offsets less than or
equal to `end` and will only request offsets greater than `end` in the future.
|
Offset |
deserializeOffset(String json)
Deserialize a JSON string into an Offset of the implementation-defined offset type.
|
Offset |
getStartOffset()
Return the specified or inferred start offset for this reader.
|
Offset |
mergeOffsets(PartitionOffset[] offsets)
Merge partitioned offsets coming from
ContinuousInputPartitionReader instances
for each partition to a single global offset. |
default boolean |
needsReconfiguration()
The execution engine will call this method in every epoch to determine if new input
partitions need to be generated, which may be required if for example the underlying
source system has had partitions added or removed.
|
void |
setStartOffset(java.util.Optional<Offset> start)
Set the desired start offset for partitions created from this reader.
|
planInputPartitions, readSchema
Offset mergeOffsets(PartitionOffset[] offsets)
ContinuousInputPartitionReader
instances
for each partition to a single global offset.Offset deserializeOffset(String json)
IllegalArgumentException
- if the JSON does not encode a valid offset for this readervoid setStartOffset(java.util.Optional<Offset> start)
Offset getStartOffset()
IllegalStateException
- if setStartOffset has not been calleddefault boolean needsReconfiguration()
void commit(Offset end)