[BEAM-5650]: Modify BoundedToUnboundedSourceAdapter to configure its reader to read more than 1 bounded source by jhalaria · Pull Request #6952 · apache/beam

jhalaria · 2018-11-05T23:16:40Z

@iemejia - Please review. Thank-you.

Noticed an issue while reading around 1000 files from S3.
Started getting connection timeouts as the max number of open connections is set to 50.
Inside https://github.com/apache/beam/blob/master/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/UnboundedSourceWrapper.java#L250 , we open all the readers simultaneously. Thats the intended behavior for an unbounded source. But when it comes to reading a bounded source (eg. Doing reads from S3), opening all connections at the same time leads to connections not available when the number of files we are trying to open is greater than the maxHttpConnections possible which by default is set to 50 (ClientConfiguration.DEFAULT_MAX_CONNECTIONS) and there isn't a way to override it [We should have the ability to override this anyways. Will create a separate PR for that.].
The Change essentially gives the BoundedToUnboundedSourceAdapter an ArrayDeque so that one reader can read more than 1 BoundedSource. Each Reader finishes reading from a BoundedSource and then goes to the next one.
In case of check-pointing, if a reader has elements that aren't read yet from a BoundedSource, we create a checkpoint with the remaining elements of the BoundedSource and the elements remaining in the ArrayDequeue.

Post-Commit Tests Status (on master branch)

Lang	Apex	Dataflow	Flink	Gearpump	Samza	Spark
Go	---	---	---	---	---	---
Java
Python	---			---	---	---

jhalaria · 2018-11-05T23:18:07Z

.../src/main/java/org/apache/beam/runners/core/construction/UnboundedReadFromBoundedSource.java

-    private BoundedSource<T> boundedSource;
+    //TODO: There must be a better way to figure out the maxOpenConnections available. Use that number here
+    //Mention this as part of PR to get feedback.
+    private static final int READER_QUEUE_SIZE = 10;


@iemejia - Whats a good way to set the value here?

Shouldn't this be part of the splitting behavior of BoundedSource?

…d more than 1 bounded source

jhalaria · 2018-11-08T06:02:29Z

@iemejia - Whenever you get a chance, please review. Thanks.

iemejia · 2018-11-08T14:44:09Z

Ok some general comments:

I updated the JIRA ticket to make it more clear now that we know that the issue is related to the way Flink translation works (by using runners-core UnboundedReadFromBoundedSource).
For the comment on tuning ClientConfiguration on S3 you can already do that via S3Options.setClientConfiguration.

In general I think the solution makes sense, however I have a bit of doubt if the point where you are fixing it is the correct one, I would like to preserve the current 1-1 relation of unbounded-bounded if possible. (Note that if we can't we will need to update the current documentation because with these changes the semantics has clearly changed).

Thinking a bit this is similar to the Connection/Thread Pool classic problem and I don't know if we may create something similar where the limited 'tasks' are such readers, or if we can just delegate this into native Flink and it will take care. Since I am not a Flink expert I asked @mxm to help us take a look, maybe he knows a better way to proceed.

mxm

Thank you for working on this problem @jhalaria. We definitely need to address this issue. Some observations from reading the code:

After the BoundedSource is split, it will be processed in bundles of READER_QUEUE_SIZE. This breaks the current guarantee that all readers will be processed at the same time. Some applications might actually assume this behavior.
There is an unnecessary change with how the class is instantiated, i.e. with an ArrayDequeue
The constructor assumes an implicit size 1, otherwise the the splitting / size estimation won't work correctly.

Rather than encapsulating this throttling behavior in UnboundedReadFromBoundedSource, wouldn't it make sense to delegate this a BoundedSource? We could create a ThrottledBoundedSource wrapper which would handle the throttling you implemented. This would be completely transparent to UnboundedReadFromBoundedSource.

The question remains how we decide during translation when to implement the throttling. I'd say, this could be a PipelineOption which you can set, as we have with HadoopFileSystemOptions.

mxm · 2018-11-09T11:30:14Z

runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ApexPipelineTranslator.java

+      ArrayDeque<BoundedSource<T>> unboundedSources = new ArrayDeque<>();
+      unboundedSources.add(transform.getSource());
+      UnboundedReadFromBoundedSource.BoundedToUnboundedSourceAdapter unboundedSource =
+          new UnboundedReadFromBoundedSource.BoundedToUnboundedSourceAdapter(unboundedSources);


Would prefer if there was no change here. Seems arbitrary to add a one-item Dequeue here.

mxm · 2018-11-09T11:30:58Z

.../src/main/java/org/apache/beam/runners/core/construction/UnboundedReadFromBoundedSource.java

  public PCollection<T> expand(PBegin input) {
-    return input.getPipeline().apply(Read.from(new BoundedToUnboundedSourceAdapter<>(source)));
+    final ArrayDeque<BoundedSource<T>> dequeue = new ArrayDeque<>(Arrays.asList(source));
+    return input.getPipeline().apply(Read.from(new BoundedToUnboundedSourceAdapter<>(dequeue)));


Please revert, wouldn't make any change here. The normal use case of wrapping a single source should still be supported.

mxm · 2018-11-09T11:31:53Z

.../src/main/java/org/apache/beam/runners/core/construction/UnboundedReadFromBoundedSource.java

-
-    public BoundedToUnboundedSourceAdapter(BoundedSource<T> boundedSource) {
-      this.boundedSource = boundedSource;
-    }


I think we should keep this constructor. See above.

mxm · 2018-11-09T11:46:19Z

.../src/main/java/org/apache/beam/runners/core/construction/UnboundedReadFromBoundedSource.java

-    private BoundedSource<T> boundedSource;
+    //TODO: There must be a better way to figure out the maxOpenConnections available. Use that number here
+    //Mention this as part of PR to get feedback.
+    private static final int READER_QUEUE_SIZE = 10;


Shouldn't this be part of the splitting behavior of BoundedSource?

mxm · 2018-11-09T11:56:37Z

.../src/main/java/org/apache/beam/runners/core/construction/UnboundedReadFromBoundedSource.java

        List<? extends BoundedSource<T>> splits = boundedSource.split(desiredBundleSize, options);
-        return splits
+        final List<? extends List<? extends BoundedSource<T>>> partition =
+            Lists.partition(splits, READER_QUEUE_SIZE);


This re-partitioning seems arbitrary / optimized for your case only.

mxm · 2018-11-09T11:57:49Z

.../src/main/java/org/apache/beam/runners/core/construction/UnboundedReadFromBoundedSource.java

+         */
+        final BoundedSource<T> boundedSource = boundedSources.peek();
+        long estimatedSize = boundedSource.getEstimatedSizeBytes(options);
+        long desiredBundleSize = estimatedSize / desiredNumSplits;


This will only estimate the bundle size based on the first boundedsource. Doesn't seem right.

mxm · 2018-11-09T12:01:49Z

.../src/main/java/org/apache/beam/runners/core/construction/UnboundedReadFromBoundedSource.java

+    public BoundedToUnboundedSourceAdapter(ArrayDeque<BoundedSource<T>> boundedSources) {
+      this.boundedSources = boundedSources;
+      final BoundedSource<T> source = boundedSources.peek();
+      this.checkpointCoder = new CheckpointCoder<>(source.getDefaultOutputCoder());


This will only work if the coder are the same for all sources.

mxm · 2018-11-09T12:15:10Z

.../src/main/java/org/apache/beam/runners/core/construction/UnboundedReadFromBoundedSource.java

+        return helper(onlyFinishReadingCurrentSource);
+      }
+
+      private boolean helper(boolean onlyFinishReadingCurrentSource) throws IOException {


Change this to a more descriptive name?

jhalaria · 2018-11-09T15:08:57Z

Thank-you @mxm and @iemejia for looking into this. I will incorporate your feedback.

robertwb · 2018-12-10T13:33:09Z

Any updates on this PR?

iemejia · 2019-01-08T13:19:57Z

Oups seems this needs a rebase too now @jhalaria

jhalaria · 2019-01-11T22:12:39Z

@robertwb - I was out on vacation. Haven't had a chance to resume work on it yet. Will let you know once I have something.

stale · 2019-03-12T22:57:47Z

This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@beam.apache.org list. Thank you for your contributions.

stale · 2019-03-19T23:07:04Z

This pull request has been closed due to lack of activity. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.

jhalaria commented Nov 5, 2018

View reviewed changes

jhalaria force-pushed the UnboundedReadFromBoundedSourceEnhanced branch 9 times, most recently from b38b23b to b0fd857 Compare November 6, 2018 22:59

Modify BoundedToUnboundedSourceAdapter to configure its reader to req…

83e321a

…d more than 1 bounded source

jhalaria force-pushed the UnboundedReadFromBoundedSourceEnhanced branch from b0fd857 to 83e321a Compare November 6, 2018 23:29

iemejia requested a review from mxm November 8, 2018 14:55

mxm reviewed Nov 9, 2018

View reviewed changes

stale bot added the stale label Mar 12, 2019

stale bot closed this Mar 19, 2019

Conversation

jhalaria commented Nov 5, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Post-Commit Tests Status (on master branch)

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jhalaria commented Nov 8, 2018

Uh oh!

iemejia commented Nov 8, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mxm left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jhalaria commented Nov 9, 2018

Uh oh!

robertwb commented Dec 10, 2018

Uh oh!

iemejia commented Jan 8, 2019

Uh oh!

jhalaria commented Jan 11, 2019

Uh oh!

stale bot commented Mar 12, 2019

Uh oh!

stale bot commented Mar 19, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

jhalaria commented Nov 5, 2018 •

edited

Loading

iemejia commented Nov 8, 2018 •

edited

Loading