ATHARVA262005 commented on code in PR #37411:
URL: https://github.com/apache/beam/pull/37411#discussion_r2724606068
##########
sdks/java/io/sparkreceiver/3/src/main/java/org/apache/beam/sdk/io/sparkreceiver/SparkReceiverIO.java:
##########
@@ -151,12 +162,21 @@ public Read<V> withStartPollTimeoutSec(Long
startPollTimeoutSec) {
return toBuilder().setStartPollTimeoutSec(startPollTimeoutSec).build();
}
- /** Inclusive start offset from which the reading should be started. */
public Read<V> withStartOffset(Long startOffset) {
checkArgument(startOffset != null, "Start offset can not be null");
return toBuilder().setStartOffset(startOffset).build();
}
+ /**
+ * A number of workers to read from Spark {@link Receiver}.
+ *
+ * <p>If this value is not set, or set to 1, the reading will be performed
on a single worker.
Review Comment:
Good catch. I have refactored the implementation to support proper sharding.
1. I added [setShard(int shardId, int
numShards)](cci:1://file:///d:/open%20souce/beam/sdks/java/io/sparkreceiver/3/src/test/java/org/apache/beam/sdk/io/sparkreceiver/CustomReceiverWithOffset.java:57:2-61:3)
to the
[HasOffset](cci:2://file:///d:/open%20souce/beam/sdks/java/io/sparkreceiver/3/src/main/java/org/apache/beam/sdk/io/sparkreceiver/HasOffset.java:23:0-44:1)
interface.
2. The DoFn now passes the unique shard ID to the Receiver via
[setShard](cci:1://file:///d:/open%20souce/beam/sdks/java/io/sparkreceiver/3/src/test/java/org/apache/beam/sdk/io/sparkreceiver/CustomReceiverWithOffset.java:57:2-61:3).
3. I updated the documentation to clarify that the receiver is expected to
handle partitioning based on these parameters.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
