Implementing Python Bounded Source Reader DoFn #13154

pabloem · 2020-10-20T21:06:35Z

This is valuable for BigQuery repeatedly firing side input. This PR is intended to be used here: #13170

This makes the SDF Bounded Source reader available to use. A small change in functionality:

If no source is provided to the initial restriction in the constructor, then the element is expected to be a source, and it's added to the initial restriction at creation.

r: @boyuanzz

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).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, 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.

Post-Commit Tests Status (on master branch)

Lang	Dataflow	Samza	Twister2
Go	---	---	---
Java
Python		---	---
XLang	---	---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---		---	---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

pabloem · 2020-10-20T22:29:48Z

Run Python 3.8 PostCommit

pabloem · 2020-10-24T17:43:39Z

Run Python 3.8 PostCommit

pabloem · 2020-10-24T17:43:54Z

Run PythonDocker PreCommit

boyuanzz

I'm curious do we have a plan to build actual SDF for BQ instead of still relying on BoundedSource implementation?

boyuanzz · 2020-10-28T00:36:51Z

sdks/python/apache_beam/io/iobase.py

@@ -1618,3 +1628,48 @@ def display_data(self):
 'source': DisplayDataItem(self.source.__class__, label='Read Source'),
 'source_dd': self.source
 }
+
+
+class SDFBoundedSourceReader(PTransform):


It seems like the major difference between SDFBoundedSourceWrapper and SDFBoundedSourceReader is that SDFBoundedSourceWrapper takes the source as construction param where SDFBoundedSourceReader takes the source as input element. We could change the implementation of SDFBoundedSourceWrapper as well.

I've done this - but I've still allowed the source to come in via the constructor as well as as an input. The intention of doing this is to keep the display data for simple Read transforms where the source is known at construction time.

I see. I thought we still keep _SDFBoundedSourceWrapper . Thanks for the clarification!

I'm thinking whether it would be better for SDFBoundedSourceReader to take data_to_display as constructor instead of source directly if any. What do you think?

pabloem · 2020-10-28T18:29:58Z

I'm curious do we have a plan to build actual SDF for BQ instead of still relying on BoundedSource implementation?

In this case, we will have a simple DoFn that starts the read from BQ, but it eventually returns multiple Avro file sources that can be read individually. This is different from what we had before, where all of the BQ reading logic was part of a BoundedSource. In fact, the _CustomBigQuerySource will be removed eventually.

pabloem · 2020-10-28T18:30:08Z

Run Python 3.8 PostCommit

boyuanzz · 2020-10-28T21:03:48Z

sdks/python/apache_beam/io/iobase.py

@@ -1618,3 +1628,48 @@ def display_data(self):
 'source': DisplayDataItem(self.source.__class__, label='Read Source'),
 'source_dd': self.source
 }
+
+
+class SDFBoundedSourceReader(PTransform):


I see. I thought we still keep _SDFBoundedSourceWrapper . Thanks for the clarification!

boyuanzz · 2020-10-28T21:04:23Z

sdks/python/apache_beam/io/iobase.py

+ initializes restriction based on input element that is expected to be of
+ BoundedSource type.
+ """
+ def __init__(self, source: BoundedSource = None, desired_chunk_size=None):


We should be able to remote source here?

boyuanzz · 2020-10-28T21:15:04Z

sdks/python/apache_beam/io/iobase.py

@@ -1618,3 +1628,48 @@ def display_data(self):
 'source': DisplayDataItem(self.source.__class__, label='Read Source'),
 'source_dd': self.source
 }
+
+
+class SDFBoundedSourceReader(PTransform):


I'm thinking whether it would be better for SDFBoundedSourceReader to take data_to_display as constructor instead of source directly if any. What do you think?

boyuanzz

Thanks! Changes look good to me except some minor comments.

boyuanzz · 2020-10-29T17:39:28Z

sdks/python/apache_beam/io/iobase.py

+ """
+ A `RestrictionProvider` that is used by SDF for `BoundedSource`.
+
+ If source is provided, uses it for initializing restriction. Otherwise


It seems like we also need to update pydoc here as well.

boyuanzz · 2020-10-29T17:56:10Z

sdks/python/apache_beam/io/iobase.py

+ self._desired_chunk_size = desired_chunk_size
+
+ def _check_source(self, src):
+ if src is not None and not isinstance(src, BoundedSource):


The src cannot be None, right?

boyuanzz · 2020-10-29T18:07:20Z

I'm curious do we have a plan to build actual SDF for BQ instead of still relying on BoundedSource implementation?

In this case, we will have a simple DoFn that starts the read from BQ, but it eventually returns multiple Avro file sources that can be read individually. This is different from what we had before, where all of the BQ reading logic was part of a BoundedSource. In fact, the _CustomBigQuerySource will be removed eventually.

I see. It seems like you will use SDFBoundedSourceReader in your BQ readAll transform. I think it would be nice to not build anything new directly on top of BoundedSource since overall we want deprecate BoundedSource in the feature. The bounded sdf wrapper is for helping us to do the migration smoothly. What do you think?

pabloem · 2020-10-29T18:32:04Z

Run Python 3.8 PostCommit

pabloem · 2020-10-29T18:32:58Z

I'm curious do we have a plan to build actual SDF for BQ instead of still relying on BoundedSource implementation?

In this case, we will have a simple DoFn that starts the read from BQ, but it eventually returns multiple Avro file sources that can be read individually. This is different from what we had before, where all of the BQ reading logic was part of a BoundedSource. In fact, the _CustomBigQuerySource will be removed eventually.

I see. It seems like you will use SDFBoundedSourceReader in your BQ readAll transform. I think it would be nice to not build anything new directly on top of BoundedSource since overall we want deprecate BoundedSource in the feature. The bounded sdf wrapper is for helping us to do the migration smoothly. What do you think?

I think that's reasonable. If any improvements are made to ReadAllFromBQ, we can make sure that they are done without relying on BoundedSource then.

pabloem · 2020-10-29T18:53:16Z

Run Portable_Python PreCommit

pabloem · 2020-11-03T22:14:35Z

Run Python 3.8 PostCommit

Implementing Python Bounded Source Reader DoFn

0ab8ea2

pabloem force-pushed the bsreadersdf branch from a8a5b08 to 0ab8ea2 Compare October 20, 2020 21:52

Adding annotations

c82f2d1

pabloem closed this Oct 23, 2020

pabloem reopened this Oct 23, 2020

pabloem mentioned this pull request Oct 24, 2020

Refactoring BigQuery Read utilities into internal file #13164

Merged

4 tasks

pabloem requested a review from boyuanzz October 26, 2020 17:23

boyuanzz reviewed Oct 28, 2020

View reviewed changes

Wrapping BoundedSource out of a single implementation

7297e0a

boyuanzz reviewed Oct 28, 2020

View reviewed changes

pabloem added 2 commits October 28, 2020 17:48

Fixup

db7aafd

Fix formatter

d0a1509

boyuanzz approved these changes Oct 29, 2020

View reviewed changes

fixup

c16e3b3

pabloem added 3 commits November 3, 2020 11:50

fixup

4934da1

fix formatting

ac09ae0

fixup

050cd4c

pabloem merged commit c3cf904 into apache:master Nov 3, 2020

pabloem deleted the bsreadersdf branch November 3, 2020 23:28

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Implementing Python Bounded Source Reader DoFn #13154

Implementing Python Bounded Source Reader DoFn #13154

pabloem commented Oct 20, 2020 •

edited

Loading

pabloem commented Oct 20, 2020

pabloem commented Oct 24, 2020

pabloem commented Oct 24, 2020

boyuanzz left a comment

boyuanzz Oct 28, 2020

pabloem Oct 28, 2020

boyuanzz Oct 28, 2020

boyuanzz Oct 28, 2020

pabloem commented Oct 28, 2020

pabloem commented Oct 28, 2020

boyuanzz Oct 28, 2020

boyuanzz Oct 28, 2020

boyuanzz Oct 28, 2020

boyuanzz left a comment

boyuanzz Oct 29, 2020

boyuanzz Oct 29, 2020

boyuanzz commented Oct 29, 2020

pabloem commented Oct 29, 2020

pabloem commented Oct 29, 2020

pabloem commented Oct 29, 2020

pabloem commented Nov 3, 2020

Implementing Python Bounded Source Reader DoFn #13154

Implementing Python Bounded Source Reader DoFn #13154

Conversation

pabloem commented Oct 20, 2020 • edited Loading

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

GitHub Actions Tests Status (on master branch)

pabloem commented Oct 20, 2020

pabloem commented Oct 24, 2020

pabloem commented Oct 24, 2020

boyuanzz 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

pabloem commented Oct 28, 2020

pabloem commented Oct 28, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

boyuanzz left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

boyuanzz commented Oct 29, 2020

pabloem commented Oct 29, 2020

pabloem commented Oct 29, 2020

pabloem commented Oct 29, 2020

pabloem commented Nov 3, 2020

pabloem commented Oct 20, 2020 •

edited

Loading