Spark 3.3: support rate limit in Spark Streaming

[singhpk234]

About the change

Supports rate limiting in SS via Admission Control API. Rate limiting is a required to have an uniform micro-batch sizes, At present spark Admission Control API exposes rate limiting on 2 things (File Read / Records Read).

Now we can have offset that read snapshot's partially (based on number of Files / Records limits specified) as well as now a Microbatch can span across mutiple snapshots where each snapshot could be partially read. Since we took into consideration already in past that a microbatch can read from [startFileIndex, endFileIndex) no changes to microbatch interface was made.

Please ref below for more about Admission Control APIs / Read Limit :

• [SPARK-30669][SS] Introduce AdmissionControl APIs for StructuredStreaming spark#27380
• https://issues.apache.org/jira/browse/SPARK-30669

---

Concepts used in this PR

(copied from : #2660)

Background on Spark 3 - MicroBatchStream

Spark expects micro batch streaming source to implement the interface MicroBatchStream

Spark driver - as part of MicroBatchExecution - instantiates the Iceberg's MicroBatchStream implementation & invokes the methods in this MicroBatchStream class in the below order:

• latestOffset(Offset startOffset, ReadLimit readLimit) - asks Iceberg streaming source to return what is the latestOffset availabe on the stream, given that previous offset was startOffset.
• initialOffset() - asks iceberg streaming source to return what is the first ever initialOffset when this stream started
• planInputPartitions(start, end) - spark runtime picks a start offset and end offset based on the stream position it is at - & asks iceberg streaming source to return a dataStructure InputPartition[] - which will later be distributed across spark executors
• createReaderFactory() - iceberg micro batch streaming source should implement this method - to educate spark executor - as to - how to read a given InputPartition
• there are other standard methods on the MicroBatchStream interface - which will be invoked by spark - like deserializeOffset, commit, stop - which are self explanatory.

Iceberg's Spark stream offset - StreamingOffset

Spark expects the streaming source - to implement Offset class. Offset - is a logical representation of position of the stream

• at which spark is reading.
  Iceberg's implementation (which already existed before this PR) is: StreamingOffset.

It stores 3 properties about the Iceberg table position:

• SnapshotId: the Iceberg table Snapshot Id
• position: file position in a given iceberg snapshot
• ScanAllFiles: whether to stream
  All files in that Snapshot - which includes - (a) files added in older Snapshots + (b) net NEW files added in the current

Snapshot
or to stream - only the files that are newly added in the current Snapshot.

---

Testing :
Added UT's to assert expected number of microbatches are made.

Closes : #2789

[singhpk234]

Hi all members,

I am attempting this feature of supporting rate-limiting when stream read iceberg table. I have put out an initial PR for the same, would appreciate your thoughts on same :) !!!

cc @rdblue, @aokolnychyi, @RussellSpitzer, @kbendick, @jackye1995, @rajarshisarkar

[singhpk234]

once read limit is respected via this PR:

will add support of Trigger.AvailableNow() introduced in spark 3.3 : https://issues.apache.org/jira/browse/SPARK-36533

[singhpk234]

cc @rdblue

[cccs-jc]

This is really cool. Currently, without being able limit the input rows, it's not possible to be down for too long..

[jackye1995]

the last ) should be ]?

[singhpk234]

added ) in place of ] to show the range endOffset is not inclusive.

[cccs-jc]

@singhpk234 it's looking good. Just a few more small changes are suggested. Do you think you'll have time to address them next week. Cheers

[singhpk234]

@singhpk234 it's looking good. Just a few more small changes are suggested. Do you think you'll have time to address them next week. Cheers

Definitely @cccs-jc ! Will address these feedback by end of this week as well as raise this against 3.3. Apologies have been keeping busy due to some internal deliverables

[jackye1995]

Looks good to me! Waiting for any other comment from @RussellSpitzer

[cccs-jc]

Looking forward to this feature @singhpk234 :-)

[jackye1995]

@RussellSpitzer do you have any further comment? If not I think this is good to go

[cccs-jc]

@RussellSpitzer good to go?

[jackye1995]

Looks like this has been inactive and I think it's ready to go.

I know we just added Spark 3.4, but given the fact that this PR has been hanging for a long time, I think it's probably better to get this in first. I will go ahead to merge it. @singhpk234 or @cccs-jc could you also add this patch to 3.4 afterwards?

[namrathamyske]

@singhpk234 any reason we don't use this from function parameter and marked this as TODO? I see that limit is not used in the whole function.

[RahmanQureshi]

@singhpk234 👋 Can you help me understand why the expected result here 1 and not 7? For example, the test "WithNumberOfRows_4" expects all of the test data to have been processed in 2 micro batches

[RahmanQureshi]

Ohh, I see! The next file has 2 records, and the rate limiting isn't sophisticated enough to break a file into multiple microbatches (which makes sense).

But I am curious why the first list of records gets broken into two files deterministically. The first file has simple-record-1, and the second file has simple-record-2 and simple-record-3.