Spark Runner : Support for Streaming side-inputs for Spark Runner #34560
Conversation
|
It is important to note one key aspect before proceeding. In Spark Runner, PCollectionView inherently utilizes Broadcast variables. As a result, it is not considered a checkpoint target. |
|
Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment |
|
There are something flaky tests. I'll fix it. |
twosom
force-pushed
the
spark-runner-streaming-side-inputs
branch
4 times, most recently
from
f13a840 to
60220a9
Compare
twosom
force-pushed
the
spark-runner-streaming-side-inputs
branch
from
8f9951c to
ea50052
Compare
|
The existing test |
twosom
force-pushed
the
spark-runner-streaming-side-inputs
branch
from
6c4f756 to
c78bf3d
Compare
twosom
force-pushed
the
spark-runner-streaming-side-inputs
branch
from
c78bf3d to
2072db6
Compare
|
Assigning reviewers. If you would like to opt out of this review, comment R: @chamikaramj added as fallback since no labels match configuration Available commands:
The PR bot will only process comments in the main thread (not review comments). |
|
Reminder, please take a look at this pr: @chamikaramj |
|
Hello, could you please review this PR? |
| @@ -50,6 +53,18 @@ public static List<PTransformOverride> getDefaultOverrides(boolean streaming) { | |||
| PTransformOverride.of( | |||
| PTransformMatchers.urnEqualTo(PTransformTranslation.SPLITTABLE_PROCESS_KEYED_URN), | |||
| new SplittableParDoNaiveBounded.OverrideFactory())); | |||
| } else { | |||
| // For streaming pipelines, this override is applied only when the PTransform has the same URN | |||
There was a problem hiding this comment.
I understand this comment goes to the first item of "PTransformOverride.of("
| @@ -28,7 +28,7 @@ | |||
| import org.apache.spark.api.java.JavaSparkContext; | |||
| import org.slf4j.Logger; | |||
| import org.slf4j.LoggerFactory; | |||
| import scala.Tuple2; | |||
| import scala.Tuple3; | |||
There was a problem hiding this comment.
any chance this will grow to Tuple4, ..., in the future? Shall we create a POJO class instead (similar to SideInputBroadcast) and make the value of the Map this POJO class
| * implementations based on the execution mode (streaming or batch) to optimize side input access | ||
| * patterns. | ||
| */ | ||
| public class SideInputReaderUtils { |
There was a problem hiding this comment.
From Beam naming convention it sounds this is a "SideInputReaderFactory". And the static method can be named after "create"
| Iterable<WindowedValue<?>> availableSideInputs = | ||
| (Iterable<WindowedValue<?>>) windowedBroadcastHelper.getValue().getValue(); | ||
| Iterable<?> sideInputForWindow = | ||
| (Iterable<WindowedValue<?>>) sideInputBroadcast.getValue(); |
There was a problem hiding this comment.
Thanks. This improved the readability
Please add a meaningful description for your change here
fixes #18136
This PR implements streaming side-inputs support in the Spark runner.
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.See the Contributor Guide for more tips on how to make review process smoother.
To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md
GitHub Actions Tests Status (on master branch)
See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.