[BEAM-14378] [CdapIO] SparkReceiverIO Read via SDF #17828

Amar3tto · 2022-06-03T10:06:19Z

Resolves #24960
Resolves #21680

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Add a link to the appropriate issue in your description, if applicable. This will automatically link the pull request to the issue.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

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.

asf-ci · 2022-06-03T10:06:21Z

Can one of the admins verify this patch?

asf-ci · 2022-06-03T10:06:21Z

Can one of the admins verify this patch?

asf-ci · 2022-06-03T10:06:21Z

Can one of the admins verify this patch?

codecov · 2022-06-03T10:31:47Z

Codecov Report

Merging #17828 (b84ab8c) into master (aa574f5) will decrease coverage by 0.17%.
The diff coverage is n/a.

❗ Current head b84ab8c differs from pull request most recent head 18ac5cc. Consider uploading reports for the commit 18ac5cc to get more accurate results

@@            Coverage Diff             @@
##           master   #17828      +/-   ##
==========================================
- Coverage   74.17%   74.00%   -0.18%     
==========================================
  Files         711      704       -7     
  Lines       93605    93250     -355     
==========================================
- Hits        69433    69007     -426     
- Misses      22895    22977      +82     
+ Partials     1277     1266      -11

Flag	Coverage Δ
python	`83.54% <ø> (-0.02%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...ython/apache_beam/io/gcp/bigquery_read_internal.py	`53.36% <0.00%> (-4.33%)`	⬇️
sdks/go/pkg/beam/io/filesystem/memfs/memory.go	`92.00% <0.00%> (-4.16%)`	⬇️
sdks/python/apache_beam/io/gcp/bigquery.py	`70.37% <0.00%> (-3.88%)`	⬇️
...ks/go/pkg/beam/runners/dataflow/dataflowlib/job.go	`21.55% <0.00%> (-1.13%)`	⬇️
sdks/go/pkg/beam/core/runtime/exec/input.go	`50.96% <0.00%> (-1.00%)`	⬇️
sdks/go/pkg/beam/core/graph/coder/row_encoder.go	`67.52% <0.00%> (-0.55%)`	⬇️
sdks/python/apache_beam/runners/direct/executor.py	`96.46% <0.00%> (-0.55%)`	⬇️
sdks/python/apache_beam/dataframe/io.py	`88.78% <0.00%> (-0.49%)`	⬇️
sdks/python/apache_beam/typehints/row_type.py	`96.55% <0.00%> (-0.47%)`	⬇️
sdks/python/apache_beam/pipeline.py	`92.42% <0.00%> (-0.42%)`	⬇️
... and 61 more

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

Amar3tto · 2022-06-16T08:00:07Z

Run Java PreCommit

Amar3tto · 2022-07-18T06:03:11Z

Run SQL_Java17 PreCommit

Amar3tto · 2022-07-20T18:09:24Z

Run Java PreCommit

Amar3tto · 2022-07-22T05:40:56Z

Run Java PreCommit

elizaveta-lomteva · 2022-07-26T15:25:39Z

@aromanenko-dev
Alexey, we've prepared the SparkReceiverIO connector Read interface. This PR is ready to be reviewed.
Thanks for your attention!

aromanenko-dev · 2022-08-01T17:11:11Z

Thanks!
I'm sorry for a delay with review, I'll try to take a look as soon as I can.

Amar3tto · 2022-08-10T16:22:55Z

@aromanenko-dev Kind ping on this.
Did you have time to review this PR?

aromanenko-dev · 2022-08-17T10:06:48Z

Please, avoid merge commits and use git rebase instead

aromanenko-dev

Thanks for contribution!

I did a first round of code review (I didn't touch SDF part), ptal.

Also, please rebase your feature branch against master and squash your commits.

aromanenko-dev · 2022-08-17T16:03:35Z

...ava/io/sparkreceiver/src/main/java/org/apache/beam/sdk/io/sparkreceiver/SparkReceiverIO.java

+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** Streaming sources for Spark {@link Receiver}. */


Please, add more details about this connector and examples of how to use it.

We expect to add more detailed documentation and README in a separate PR.

Is there any reason why not to add it along with this PR? Seems logical for me...

Added javadoc

...io/sparkreceiver/src/test/java/org/apache/beam/sdk/io/sparkreceiver/SparkReceiverIOTest.java

.../src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java

.../java/io/sparkreceiver/src/main/java/org/apache/beam/sdk/io/sparkreceiver/SparkConsumer.java

.../src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java

...ava/io/sparkreceiver/src/main/java/org/apache/beam/sdk/io/sparkreceiver/SparkReceiverIO.java

...io/sparkreceiver/src/test/java/org/apache/beam/sdk/io/sparkreceiver/SparkReceiverIOTest.java

aromanenko-dev · 2022-08-19T13:46:45Z

Thanks for addressing review comments. It looks ok in general for me.

I'll be on PTO for the next 2 weeks, so I'd kindly ask @chamikaramj or @lukecwik to take a look on SDF part.

chamikaramj · 2022-08-19T13:56:40Z

Ack. Will take a look. Thanks @aromanenko-dev

Amar3tto · 2022-08-20T05:38:47Z

Run Java PreCommit

chamikaramj

Thanks.

chamikaramj · 2022-08-22T13:53:17Z

.../src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java

+ objects -> {
+ V record = (V) objects[0];
+ recordsQueue.offer(record);
+ return null;


"return null" was intended here ?

Yes it was intended, because it is SerializableFunction<?, Void> and it should return something. The common way is to return null here.

chamikaramj · 2022-08-22T13:55:34Z

.../src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java

+ return null;
+ });
+ } catch (Exception e) {
+ LOG.error("Can not init Spark Receiver!", e);


We should be raising an exception here (or retry) instead of logging ?

chamikaramj · 2022-08-22T14:00:16Z

.../src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java

+ try {
+ TimeUnit.MILLISECONDS.sleep(START_POLL_TIMEOUT_MS);
+ } catch (InterruptedException e) {
+ LOG.error("SparkReceiver was interrupted before polling started", e);


chamikaramj · 2022-08-22T14:01:28Z

.../src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java

+ if (sparkReceiver != null) {
+ sparkReceiver.stop("SparkReceiver is stopped.");
+ }
+ recordsQueue.clear();


What happens to existing records in the queue ?

At this moment all records for the current restriction should be already read from the queue.
Here we are stopping the Receiver and clearing the queue because it is not needed anymore.

chamikaramj · 2022-08-23T16:25:43Z

.../src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java

+ checkStateNotNull(getOffsetFn, "Get offset fn can't be null!");
+ this.getOffsetFn = getOffsetFn;
+
+ SerializableFunction<V, Instant> getWatermarkFn = transform.getWatermarkFn();


This probably should be renamed to timestampFn since it basically provides the timestamp for a given record.

chamikaramj · 2022-08-24T13:37:56Z

...ava/io/sparkreceiver/src/main/java/org/apache/beam/sdk/io/sparkreceiver/SparkReceiverIO.java

+ }
+
+ /** A function to calculate watermark after a record. */
+ public Read<V> withWatermarkFn(SerializableFunction<V, Instant> watermarkFn) {


Probably should be renamed to getTimestampFn or similar (and also update the Javadoc).

chamikaramj · 2022-08-24T13:40:10Z

...ava/io/sparkreceiver/src/main/java/org/apache/beam/sdk/io/sparkreceiver/SparkReceiverIO.java

+ LOG.info("{} started reading", ReadFromSparkReceiverWithOffsetDoFn.class.getSimpleName());
+ return input
+ .apply(Impulse.create())
+ .apply(ParDo.of(new ReadFromSparkReceiverWithOffsetDoFn<>(sparkReceiverRead)));


Seems like we don't do any kind of splitting here which will limit the reading to a single worker. Is it possible to split data from SparkReceiver into multiple workers (for example, Kafka does this by creating splits for different partitions).

Could you please merge this implementation to start with, to unblock the work on PRs with Spark Recevier integration tests and integration with Cdap IO?
While we haven't found a way to split the Spark receivers, we continue to work in this direction

Yeah, this can be pushed to a future PR. Let's add a TODO though.

chamikaramj · 2022-08-24T13:47:34Z

...arkreceiver/src/test/java/org/apache/beam/sdk/io/sparkreceiver/CustomReceiverWithOffset.java

+
+ private void receive() {
+ Long currentOffset = startOffset;
+ while (!isStopped()) {


This operation of the test does not seem to be thread safe currently (source might read while a batch is being loaded).

chamikaramj

Thanks. Added a couple more comments which I think are critical issues. Splitting can be pushed to a future change.

chamikaramj · 2022-08-31T00:28:29Z

.../src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java

+ V record = sparkConsumer.poll();
+ if (record != null) {
+ Long offset = getOffsetFn.apply(record);
+ if (!tracker.tryClaim(offset)) {


When the runner resume an element from a checkpoint, how will the source start reading from the correct position ?

I think there should be logic here to seek to the correct position when "currentRestriction().getFrom()" is not zero.

Note that this has to be a seek (not read all elements from the beginning) for reading to be efficient.

Start reading from the correct position ("seek") implemented here:
sparkConsumer = new SparkConsumerWithOffset<>(tracker.currentRestriction().getFrom());

chamikaramj · 2022-08-31T00:31:58Z

.../src/main/java/org/apache/beam/sdk/io/sparkreceiver/ReadFromSparkReceiverWithOffsetDoFn.java

+ Long offset = getOffsetFn.apply(record);
+ if (!tracker.tryClaim(offset)) {
+ sparkConsumer.stop();
+ LOG.debug("Stop for restriction: {}", tracker.currentRestriction().toString());


I don't see any sort of acking here for messages that are already read. How does the SparkConsumer finalize messages so that it does not have to hold off to data forever ? Usually this is implemented via a bundle finalizer: https://beam.apache.org/documentation/programming-guide/#bundle-finalization

Discussed at the meeting that an additional ack is not needed, since Spark Consumer is created inside DoFn. As soon as element is processed it’s deleted from queue (not from the source).

chamikaramj · 2022-08-31T00:38:39Z

...ava/io/sparkreceiver/src/main/java/org/apache/beam/sdk/io/sparkreceiver/SparkReceiverIO.java

+ * .withSparkReceiverBuilder(receiverBuilder);
+ * }</pre>
+ */
+public class SparkReceiverIO {


Please add an experimental tag (see [1] for an example) since this is a new connector and the API can change.

[1]

beam/sdks/java/io/kinesis/src/main/java/org/apache/beam/sdk/io/kinesis/KinesisIO.java

Line 322 in f5435c0

@Experimental(Kind.SOURCE_SINK)

chamikaramj · 2022-08-31T00:41:18Z

...ava/io/sparkreceiver/src/main/java/org/apache/beam/sdk/io/sparkreceiver/SparkReceiverIO.java

+ LOG.info("{} started reading", ReadFromSparkReceiverWithOffsetDoFn.class.getSimpleName());
+ return input
+ .apply(Impulse.create())
+ .apply(ParDo.of(new ReadFromSparkReceiverWithOffsetDoFn<>(sparkReceiverRead)));


Yeah, this can be pushed to a future PR. Let's add a TODO though.

chamikaramj · 2022-09-13T00:49:23Z

Any updates ?

elizaveta-lomteva · 2022-09-19T14:29:48Z

@chamikaramj
All your comments have been answered. The PR is ready to be merged.
Thank you.

chamikaramj

Thanks. LGTM.

Abacn · 2022-09-30T16:22:06Z

New test flake seen on SparkReceiverIO test: #23449

github-actions bot added build io java labels Jun 3, 2022

elizaveta-lomteva mentioned this pull request Jul 14, 2022

[BEAM-14101] [CdapIO] Add ReceiverBuilder for SparkReceiverIO #17111

Merged

4 tasks

Amar3tto force-pushed the BEAM-14378-sdf branch from b84ab8c to be9c2fb Compare July 20, 2022 12:20

github-actions bot removed the build label Jul 20, 2022

Amar3tto force-pushed the BEAM-14378-sdf branch from 25ec729 to 1066c98 Compare July 21, 2022 10:18

Amar3tto marked this pull request as ready for review July 22, 2022 10:01

aromanenko-dev self-requested a review August 1, 2022 17:11

aromanenko-dev requested changes Aug 17, 2022

View reviewed changes

aromanenko-dev reviewed Aug 18, 2022

View reviewed changes

...io/sparkreceiver/src/test/java/org/apache/beam/sdk/io/sparkreceiver/SparkReceiverIOTest.java Show resolved Hide resolved

[BEAM-14378] Add SparkReceiverIO

e831176

Amar3tto force-pushed the BEAM-14378-sdf branch from 9642fef to e831176 Compare August 19, 2022 13:40

chamikaramj requested changes Aug 24, 2022

View reviewed changes

Fix comments

0088a2b

chamikaramj reviewed Aug 31, 2022

View reviewed changes

Amar3tto and others added 3 commits September 15, 2022 12:27

Resolve comments

4211097

Fix checkstyle

1fa0b9a

Add TestOutputDoFn's result set size assertion

18ac5cc

chamikaramj reviewed Sep 20, 2022

View reviewed changes

chamikaramj merged commit d578e3d into apache:master Sep 20, 2022

Amar3tto deleted the BEAM-14378-sdf branch September 21, 2022 11:57

[BEAM-14378] [CdapIO] SparkReceiverIO Read via SDF #17828

[BEAM-14378] [CdapIO] SparkReceiverIO Read via SDF #17828

Conversation

Amar3tto commented Jun 3, 2022 • edited Loading

GitHub Actions Tests Status (on master branch)

asf-ci commented Jun 3, 2022

asf-ci commented Jun 3, 2022

asf-ci commented Jun 3, 2022

codecov bot commented Jun 3, 2022 • edited Loading

Codecov Report

Amar3tto commented Jun 16, 2022

Amar3tto commented Jul 18, 2022

Amar3tto commented Jul 20, 2022

Amar3tto commented Jul 22, 2022

elizaveta-lomteva commented Jul 26, 2022

aromanenko-dev commented Aug 1, 2022

Amar3tto commented Aug 10, 2022

aromanenko-dev commented Aug 17, 2022

aromanenko-dev left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

aromanenko-dev commented Aug 19, 2022

chamikaramj commented Aug 19, 2022

Amar3tto commented Aug 20, 2022

chamikaramj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chamikaramj commented Sep 13, 2022

elizaveta-lomteva commented Sep 19, 2022

chamikaramj left a comment

Choose a reason for hiding this comment

Abacn commented Sep 30, 2022

Amar3tto commented Jun 3, 2022 •

edited

Loading

codecov bot commented Jun 3, 2022 •

edited

Loading