You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2022/08/06 19:49:06 UTC
[GitHub] [beam] svetakvsundhar opened a new pull request, #22616: BEAM-11587 Wrapper PTransform
svetakvsundhar opened a new pull request, #22616:
URL: https://github.com/apache/beam/pull/22616
This PR adds a wrapper PTransform 'ReadGbq', that allows users to execute on PCollections, to ReadFromBQ, convert the schema into a Beam Schema, and plumb into a Deferred Dataframe.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Mention the appropriate issue in your description (for example: `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, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/get-started-contributing/#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1213588731
Run Python 3.8 PostCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a diff in pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on code in PR #22616:
URL: https://github.com/apache/beam/pull/22616#discussion_r942836844
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -58,6 +58,17 @@
_DEFAULT_BYTES_CHUNKSIZE = 1 << 20
+@frame_base.with_docs_from(pd)
+def read_gbq(table, *args, **kwargs):
+ """If you would like to use the 'DIRECT_READ' method ins ReadFromBigQuery,
+ please set use_bq_storage_api to True.
+ Otherwise, if you would like to use the 'EXPORT' method, please set
+ use_bq_storage_api to False, or leave it unspecified."""
Review Comment:
From the perspective of the users of this API, the mode is fully controlled by the `use_bqstorage_api` parameter. It just confuses things to reveal that it maps to mode={DIRECT_READ, EXPORT} under the hood.
If we want to reference the mode in the docstring, we should just have a `mode` parameter, and not bother with a `use_bqstorage_api` parameter. To be clear, I think that's an acceptable alternative, but I'd like to pick one or the other.
I have a slight preference to avoid using Beam terminology though - a goal of the DataFrame API is to provide an API that feels natural to developers who are already familiar with pandas, and not familiar with Beam.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1233266637
Run Python 3.8 PostCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1216916412
Run RAT PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a diff in pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on code in PR #22616:
URL: https://github.com/apache/beam/pull/22616#discussion_r957876647
##########
sdks/python/apache_beam/dataframe/io_it_test.py:
##########
@@ -0,0 +1,123 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Integration tests for Dataframe sources and sinks."""
+# pytype: skip-file
+
+import logging
+import unittest
+
+import pytest
+
+import apache_beam as beam
+import apache_beam.io.gcp.bigquery
+from apache_beam.io.gcp import bigquery_read_it_test
+from apache_beam.io.gcp import bigquery_schema_tools
+from apache_beam.io.gcp import bigquery_tools
+from apache_beam.testing.util import assert_that
+from apache_beam.testing.util import equal_to
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ReadUsingReadGbqTests(bigquery_read_it_test.BigQueryReadIntegrationTests):
+ @pytest.mark.it_postcommit
+ def test_ReadGbq(self):
+ from apache_beam.dataframe import convert
+ with beam.Pipeline(argv=self.args) as p:
+ actual_df = p | apache_beam.dataframe.io.read_gbq(
+ table="apache-beam-testing:beam_bigquery_io_test."
+ "dfsqltable_3c7d6fd5_16e0460dfd0",
+ use_bqstorage_api=False)
+ assert_that(
+ convert.to_pcollection(actual_df),
+ equal_to([(3, 'customer1', 'test'), (1, 'customer1', 'test'),
+ (2, 'customer2', 'test'), (4, 'customer2', 'test')]))
+
+ def test_ReadGbq_export_with_project(self):
+ from apache_beam.dataframe import convert
+ with beam.Pipeline(argv=self.args) as p:
+ actual_df = p | apache_beam.dataframe.io.read_gbq(
+ table="dfsqltable_3c7d6fd5_16e0460dfd0",
+ dataset="beam_bigquery_io_test",
+ project_id="apache-beam-testing",
+ use_bqstorage_api=False)
+ assert_that(
+ convert.to_pcollection(actual_df),
+ equal_to([(3, 'customer1', 'test'), (1, 'customer1', 'test'),
+ (2, 'customer2', 'test'), (4, 'customer2', 'test')]))
+
+ @pytest.mark.it_postcommit
+ def test_ReadGbq_direct_read(self):
+ from apache_beam.dataframe import convert
+ with beam.Pipeline(argv=self.args) as p:
+ actual_df = p | apache_beam.dataframe.io.\
+ read_gbq(
+ table=
+ "apache-beam-testing:beam_bigquery_io_test."
+ "dfsqltable_3c7d6fd5_16e0460dfd0",
+ use_bqstorage_api=True)
+ assert_that(
+ convert.to_pcollection(actual_df),
+ equal_to([(3, 'customer1', 'test'), (1, 'customer1', 'test'),
+ (2, 'customer2', 'test'), (4, 'customer2', 'test')]))
+
+ @pytest.mark.it_postcommit
+ def test_ReadGbq_direct_read_with_project(self):
+ from apache_beam.dataframe import convert
+ with beam.Pipeline(argv=self.args) as p:
+ actual_df = p | apache_beam.dataframe.io.read_gbq(
+ table="dfsqltable_3c7d6fd5_16e0460dfd0",
+ dataset="beam_bigquery_io_test",
+ project_id="apache-beam-testing",
+ use_bqstorage_api=True)
+ assert_that(
+ convert.to_pcollection(actual_df),
+ equal_to([(3, 'customer1', 'test'), (1, 'customer1', 'test'),
+ (2, 'customer2', 'test'), (4, 'customer2', 'test')]))
+
+ @pytest.mark.it_postcommit
+ def test_ReadGbq_with_computation(self):
+ the_table = bigquery_tools.BigQueryWrapper().get_table(
+ project_id="apache-beam-testing",
+ dataset_id="beam_bigquery_io_test",
+ table_id="dfsqltable_3c7d6fd5_16e0460dfd0")
+ table = the_table.schema
+ utype = bigquery_schema_tools. \
+ generate_user_type_from_bq_schema(table)
+ from apache_beam.dataframe import convert
+ with beam.Pipeline(argv=self.args) as p:
+ beam_df = p | apache_beam.dataframe.io.read_gbq(
+ table="dfsqltable_3c7d6fd5_16e0460dfd0",
+ dataset="beam_bigquery_io_test",
+ project_id="apache-beam-testing")
+ actual_df = beam_df.groupby('id').count()
+ assert_that(
+ convert.to_pcollection(actual_df, include_indexes=True),
+ equal_to([
+ utype(id=1, name=1, type=1),
+ utype(id=2, name=1, type=1),
+ utype(id=3, name=1, type=1),
+ utype(id=4, name=1, type=1)
Review Comment:
Can we also avoid using utype in this test?
##########
sdks/python/apache_beam/dataframe/io_it_test.py:
##########
@@ -0,0 +1,123 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Integration tests for Dataframe sources and sinks."""
+# pytype: skip-file
+
+import logging
+import unittest
+
+import pytest
+
+import apache_beam as beam
+import apache_beam.io.gcp.bigquery
+from apache_beam.io.gcp import bigquery_read_it_test
+from apache_beam.io.gcp import bigquery_schema_tools
+from apache_beam.io.gcp import bigquery_tools
+from apache_beam.testing.util import assert_that
+from apache_beam.testing.util import equal_to
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ReadUsingReadGbqTests(bigquery_read_it_test.BigQueryReadIntegrationTests):
Review Comment:
```suggestion
class ReadUsingReadGbqTests(unittest.TestCase):
```
Can we avoid inheriting from the BQ test here? It looks like maybe we do it to get access to `self.args`, but I don't think that should be necessary. If you use `with TestPipeline` instead of `with Pipeline` it should handle passing through test_pipeline_options.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1229269083
Run Python 3.8 PostCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit merged pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
TheNeuralBit merged PR #22616:
URL: https://github.com/apache/beam/pull/22616
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] github-actions[bot] commented on pull request #22616: BEAM-11587 Wrapper PTransform
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1207286067
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @pabloem for label python.
R: @johnjcasey for label io.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on a diff in pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on code in PR #22616:
URL: https://github.com/apache/beam/pull/22616#discussion_r954253870
##########
sdks/python/apache_beam/io/gcp/dataframe_io_it_test.py:
##########
@@ -0,0 +1,146 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Unit tests for Dataframe sources and sinks."""
+# pytype: skip-file
+
+import logging
+import unittest
+
+import pytest
+
+import apache_beam as beam
+import apache_beam.io.gcp.bigquery
+from apache_beam.io.gcp import bigquery_read_it_test
+from apache_beam.io.gcp import bigquery_schema_tools
+from apache_beam.io.gcp import bigquery_tools
+from apache_beam.testing.util import assert_that
+from apache_beam.testing.util import equal_to
+
+# Protect against environments where bigquery library is not available.
+# pylint: disable=wrong-import-order, wrong-import-position
+# pylint: enable=wrong-import-order, wrong-import-position
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ReadUsingReadGbqTests(bigquery_read_it_test.BigQueryReadIntegrationTests):
Review Comment:
this is to set up and tear down the requests to BQ.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on a diff in pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on code in PR #22616:
URL: https://github.com/apache/beam/pull/22616#discussion_r945818844
##########
sdks/python/apache_beam/io/gcp/dataframe_io_it_test.py:
##########
@@ -0,0 +1,146 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Unit tests for Dataframe sources and sinks."""
+# pytype: skip-file
+
+import logging
+import unittest
+
+import pytest
+
+import apache_beam as beam
+import apache_beam.io.gcp.bigquery
+from apache_beam.io.gcp import bigquery_read_it_test
+from apache_beam.io.gcp import bigquery_schema_tools
+from apache_beam.io.gcp import bigquery_tools
+from apache_beam.testing.util import assert_that
+from apache_beam.testing.util import equal_to
+
+# Protect against environments where bigquery library is not available.
+# pylint: disable=wrong-import-order, wrong-import-position
+# pylint: enable=wrong-import-order, wrong-import-position
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ReadUsingReadGbqTests(bigquery_read_it_test.BigQueryReadIntegrationTests):
+ @pytest.mark.it_postcommit
+ def test_ReadGbq(self):
+ from apache_beam.dataframe import convert
+ the_table = bigquery_tools.BigQueryWrapper().get_table(
+ project_id="apache-beam-testing",
+ dataset_id="beam_bigquery_io_test",
+ table_id="dfsqltable_3c7d6fd5_16e0460dfd0")
+ table = the_table.schema
+ utype = bigquery_schema_tools. \
+ generate_user_type_from_bq_schema(table)
+ with beam.Pipeline(argv=self.args) as p:
+ actual_df = p | apache_beam.dataframe.io.read_gbq(
+ table="apache-beam-testing:beam_bigquery_io_test."
+ "dfsqltable_3c7d6fd5_16e0460dfd0",
+ use_bqstorage_api=False)
+ assert_that(
+ convert.to_pcollection(actual_df),
+ equal_to([
+ utype(id=3, name='customer1', type='test'),
+ utype(id=1, name='customer1', type='test'),
+ utype(id=2, name='customer2', type='test'),
+ utype(id=4, name='customer2', type='test')
+ ]))
+
+ def test_ReadGbq_export_with_project(self):
+ from apache_beam.dataframe import convert
+ the_table = bigquery_tools.BigQueryWrapper().get_table(
+ project_id="apache-beam-testing",
+ dataset_id="beam_bigquery_io_test",
+ table_id="dfsqltable_3c7d6fd5_16e0460dfd0")
+ table = the_table.schema
+ utype = bigquery_schema_tools. \
+ generate_user_type_from_bq_schema(table)
+ with beam.Pipeline(argv=self.args) as p:
+ actual_df = p | apache_beam.dataframe.io.read_gbq(
+ table="dfsqltable_3c7d6fd5_16e0460dfd0",
+ dataset="beam_bigquery_io_test",
+ project_id="apache-beam-testing",
+ use_bqstorage_api=False)
+ assert_that(
+ convert.to_pcollection(actual_df),
+ equal_to([
+ utype(id=3, name='customer1', type='test'),
+ utype(id=1, name='customer1', type='test'),
+ utype(id=2, name='customer2', type='test'),
+ utype(id=4, name='customer2', type='test')
+ ]))
+
+ @pytest.mark.it_postcommit
+ def test_ReadGbq_direct_read(self):
+ from apache_beam.dataframe import convert
+ the_table = bigquery_tools.BigQueryWrapper().get_table(
+ project_id="apache-beam-testing",
+ dataset_id="beam_bigquery_io_test",
+ table_id="dfsqltable_3c7d6fd5_16e0460dfd0")
+ table = the_table.schema
+ utype = bigquery_schema_tools. \
+ generate_user_type_from_bq_schema(table)
+ with beam.Pipeline(argv=self.args) as p:
+ actual_df = p | apache_beam.dataframe.io.\
+ read_gbq(
+ table=
+ "apache-beam-testing:beam_bigquery_io_test."
+ "dfsqltable_3c7d6fd5_16e0460dfd0",
+ use_bqstorage_api=True)
+ assert_that(
+ convert.to_pcollection(actual_df),
+ equal_to([
+ utype(id=3, name='customer1', type='test'),
+ utype(id=1, name='customer1', type='test'),
+ utype(id=2, name='customer2', type='test'),
+ utype(id=4, name='customer2', type='test')
+ ]))
+
+ @pytest.mark.it_postcommit
+ def test_ReadGbq_direct_read_with_project(self):
+ from apache_beam.dataframe import convert
+ the_table = bigquery_tools.BigQueryWrapper().get_table(
+ project_id="apache-beam-testing",
+ dataset_id="beam_bigquery_io_test",
+ table_id="dfsqltable_3c7d6fd5_16e0460dfd0")
+ table = the_table.schema
+ utype = bigquery_schema_tools. \
+ generate_user_type_from_bq_schema(table)
+ with beam.Pipeline(argv=self.args) as p:
+ actual_df = p | apache_beam.dataframe.io.read_gbq(
+ table="dfsqltable_3c7d6fd5_16e0460dfd0",
+ dataset="beam_bigquery_io_test",
+ project_id="apache-beam-testing",
+ use_bqstorage_api=True)
+ assert_that(
+ convert.to_pcollection(actual_df),
+ equal_to([
+ utype(id=3, name='customer1', type='test'),
+ utype(id=1, name='customer1', type='test'),
+ utype(id=2, name='customer2', type='test'),
+ utype(id=4, name='customer2', type='test')
+ ]))
Review Comment:
sg. ill add an integration test here and make a separate PR for an example pipeline to add?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on a diff in pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on code in PR #22616:
URL: https://github.com/apache/beam/pull/22616#discussion_r942807437
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -58,6 +58,17 @@
_DEFAULT_BYTES_CHUNKSIZE = 1 << 20
+@frame_base.with_docs_from(pd)
Review Comment:
Sure-- for clarity here, should the new docstring be documenting the args to read_gbq, as well as the ReadGbq PTransform (e.g. some of the comments written in the ReadGbq PTransform documentation)?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1226190435
Run Python 3.8 PostCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1226526348
Thanks @TheNeuralBit ! Ive addressed changes, ptal. I've also verified that the new IT's are running.
<img width="1111" alt="Screen Shot 2022-08-24 at 4 01 22 PM" src="https://user-images.githubusercontent.com/26037657/186535249-b18d337b-ab1a-48c8-bbdb-1479cd2c305e.png">
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] commented on pull request #22616: BEAM-11587 Wrapper PTransform
Posted by GitBox <gi...@apache.org>.
codecov[bot] commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1207277186
# [Codecov](https://codecov.io/gh/apache/beam/pull/22616?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#22616](https://codecov.io/gh/apache/beam/pull/22616?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (fe07f19) into [master](https://codecov.io/gh/apache/beam/commit/720ee14e9c56e55750c1fed4b5c47d4486da5c80?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (720ee14) will **decrease** coverage by `0.01%`.
> The diff coverage is `72.72%`.
```diff
@@ Coverage Diff @@
## master #22616 +/- ##
==========================================
- Coverage 74.23% 74.22% -0.02%
==========================================
Files 707 707
Lines 93295 93322 +27
==========================================
+ Hits 69257 69267 +10
- Misses 22770 22787 +17
Partials 1268 1268
```
| Flag | Coverage Δ | |
|---|---|---|
| python | `83.60% <72.72%> (-0.02%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/22616?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [sdks/python/apache\_beam/dataframe/transforms.py](https://codecov.io/gh/apache/beam/pull/22616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vZGF0YWZyYW1lL3RyYW5zZm9ybXMucHk=) | `95.08% <72.72%> (-0.90%)` | :arrow_down: |
| [.../apache\_beam/runners/interactive/dataproc/types.py](https://codecov.io/gh/apache/beam/pull/22616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9pbnRlcmFjdGl2ZS9kYXRhcHJvYy90eXBlcy5weQ==) | `93.10% <0.00%> (-3.45%)` | :arrow_down: |
| [sdks/python/apache\_beam/utils/interactive\_utils.py](https://codecov.io/gh/apache/beam/pull/22616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdXRpbHMvaW50ZXJhY3RpdmVfdXRpbHMucHk=) | `95.12% <0.00%> (-2.44%)` | :arrow_down: |
| [...n/apache\_beam/ml/gcp/recommendations\_ai\_test\_it.py](https://codecov.io/gh/apache/beam/pull/22616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWwvZ2NwL3JlY29tbWVuZGF0aW9uc19haV90ZXN0X2l0LnB5) | `73.46% <0.00%> (-2.05%)` | :arrow_down: |
| [sdks/python/apache\_beam/internal/metrics/metric.py](https://codecov.io/gh/apache/beam/pull/22616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW50ZXJuYWwvbWV0cmljcy9tZXRyaWMucHk=) | `93.00% <0.00%> (-1.00%)` | :arrow_down: |
| [sdks/python/apache\_beam/runners/direct/executor.py](https://codecov.io/gh/apache/beam/pull/22616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9kaXJlY3QvZXhlY3V0b3IucHk=) | `96.46% <0.00%> (-0.55%)` | :arrow_down: |
| [...eam/runners/interactive/interactive\_environment.py](https://codecov.io/gh/apache/beam/pull/22616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9pbnRlcmFjdGl2ZS9pbnRlcmFjdGl2ZV9lbnZpcm9ubWVudC5weQ==) | `91.71% <0.00%> (-0.31%)` | :arrow_down: |
| [...thon/apache\_beam/ml/inference/sklearn\_inference.py](https://codecov.io/gh/apache/beam/pull/22616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWwvaW5mZXJlbmNlL3NrbGVhcm5faW5mZXJlbmNlLnB5) | `95.08% <0.00%> (-0.16%)` | :arrow_down: |
| [...hon/apache\_beam/runners/worker/bundle\_processor.py](https://codecov.io/gh/apache/beam/pull/22616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvYnVuZGxlX3Byb2Nlc3Nvci5weQ==) | `93.54% <0.00%> (-0.13%)` | :arrow_down: |
| [...thon/apache\_beam/ml/inference/pytorch\_inference.py](https://codecov.io/gh/apache/beam/pull/22616/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWwvaW5mZXJlbmNlL3B5dG9yY2hfaW5mZXJlbmNlLnB5) | `0.00% <0.00%> (ø)` | |
| ... and [5 more](https://codecov.io/gh/apache/beam/pull/22616/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
:mega: Codecov can now indicate which changes are the most critical in Pull Requests. [Learn more](https://about.codecov.io/product/feature/runtime-insights/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on a diff in pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on code in PR #22616:
URL: https://github.com/apache/beam/pull/22616#discussion_r942489348
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -58,6 +58,17 @@
_DEFAULT_BYTES_CHUNKSIZE = 1 << 20
+@frame_base.with_docs_from(pd)
+def read_gbq(table, *args, **kwargs):
+ """If you would like to use the 'DIRECT_READ' method ins ReadFromBigQuery,
+ please set use_bq_storage_api to True.
+ Otherwise, if you would like to use the 'EXPORT' method, please set
+ use_bq_storage_api to False, or leave it unspecified."""
Review Comment:
Hm, I disagree. I'm assuming that users that want to ReadFromBQ might already be use to the way the ReadFromBQ PTransform works, and thus it would be useful to let them know how to utilize 'DIRECT_READ' and 'EXPORT' via the ReadGbq PTransform.
What are the benefits of not mentioning 'DIRECT_READ' and 'EXPORT'?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1233453286
@TheNeuralBit ptal and lmk if the io_it_test looks viable
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1213455076
Thanks for the suggestions @TheNeuralBit , I think this is ready for another pass.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: BEAM-11587 Wrapper PTransform
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1207506329
@TheNeuralBit this is ready for an initial pass, PTAL.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1226370144
Run Python PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on a diff in pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on code in PR #22616:
URL: https://github.com/apache/beam/pull/22616#discussion_r942839270
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -58,6 +58,17 @@
_DEFAULT_BYTES_CHUNKSIZE = 1 << 20
+@frame_base.with_docs_from(pd)
+def read_gbq(table, *args, **kwargs):
+ """If you would like to use the 'DIRECT_READ' method ins ReadFromBigQuery,
+ please set use_bq_storage_api to True.
+ Otherwise, if you would like to use the 'EXPORT' method, please set
+ use_bq_storage_api to False, or leave it unspecified."""
Review Comment:
ah ok, thanks for sharing, I agree with your perspective on this and will use the pandas terminology.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a diff in pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on code in PR #22616:
URL: https://github.com/apache/beam/pull/22616#discussion_r944932683
##########
sdks/python/apache_beam/io/gcp/dataframe_io_it_test.py:
##########
@@ -0,0 +1,146 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Unit tests for Dataframe sources and sinks."""
Review Comment:
nit:
```suggestion
"""Integration tests for Dataframe sources and sinks."""
```
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -58,6 +58,28 @@
_DEFAULT_BYTES_CHUNKSIZE = 1 << 20
+def read_gbq(
+ table, dataset=None, project_id=None, use_bqstorage_api=False, **kwargs):
+ """This function reads data from a BigQuery source and outputs it into
+ a Beam deferred dataframe
+ (https://beam.apache.org/documentation/dsls/dataframes/overview/)
+ Please specify a table in the format 'PROJECT:dataset.table'
+ or use the table, dataset, and project_id args
+ to specify the table. If you would like to utilize the BigQuery
+ Storage API in ReadFromBigQuery,
+ please set use_bq_storage_api to True.
+ Otherwise, please set the flag to false or
+ leave it unspecified."""
+ if table is None:
+ raise ValueError("Please specify a BigQuery table to read from.")
+ elif len(kwargs) > 0:
+ raise ValueError(
+ "Unsupported parameter entered in read_gbq. Please enter only "
+ "supported parameters 'table', 'dataset', "
+ "'project_id', 'use_bqstorage_api'.")
Review Comment:
nit: maybe print the contents of kwargs.keys() instead, to identify the specific unsupported parameters?
##########
sdks/python/apache_beam/io/gcp/dataframe_io_it_test.py:
##########
@@ -0,0 +1,146 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Unit tests for Dataframe sources and sinks."""
+# pytype: skip-file
+
+import logging
+import unittest
+
+import pytest
+
+import apache_beam as beam
+import apache_beam.io.gcp.bigquery
+from apache_beam.io.gcp import bigquery_read_it_test
+from apache_beam.io.gcp import bigquery_schema_tools
+from apache_beam.io.gcp import bigquery_tools
+from apache_beam.testing.util import assert_that
+from apache_beam.testing.util import equal_to
+
+# Protect against environments where bigquery library is not available.
+# pylint: disable=wrong-import-order, wrong-import-position
+# pylint: enable=wrong-import-order, wrong-import-position
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ReadUsingReadGbqTests(bigquery_read_it_test.BigQueryReadIntegrationTests):
+ @pytest.mark.it_postcommit
+ def test_ReadGbq(self):
+ from apache_beam.dataframe import convert
+ the_table = bigquery_tools.BigQueryWrapper().get_table(
+ project_id="apache-beam-testing",
+ dataset_id="beam_bigquery_io_test",
+ table_id="dfsqltable_3c7d6fd5_16e0460dfd0")
+ table = the_table.schema
+ utype = bigquery_schema_tools. \
+ generate_user_type_from_bq_schema(table)
+ with beam.Pipeline(argv=self.args) as p:
+ actual_df = p | apache_beam.dataframe.io.read_gbq(
+ table="apache-beam-testing:beam_bigquery_io_test."
+ "dfsqltable_3c7d6fd5_16e0460dfd0",
+ use_bqstorage_api=False)
+ assert_that(
+ convert.to_pcollection(actual_df),
+ equal_to([
+ utype(id=3, name='customer1', type='test'),
+ utype(id=1, name='customer1', type='test'),
+ utype(id=2, name='customer2', type='test'),
+ utype(id=4, name='customer2', type='test')
+ ]))
Review Comment:
I think you should be able to just compare the result to tuple instances, then you wouldn't need to generate the utype above. Will this work?
```suggestion
equal_to([
(3, 'customer1', 'test'),
(1, 'customer1', 'test'),
(2, 'customer2', 'test'),
(4, 'customer2', 'test')
]))
```
##########
sdks/python/apache_beam/io/gcp/dataframe_io_it_test.py:
##########
@@ -0,0 +1,146 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Unit tests for Dataframe sources and sinks."""
+# pytype: skip-file
+
+import logging
+import unittest
+
+import pytest
+
+import apache_beam as beam
+import apache_beam.io.gcp.bigquery
+from apache_beam.io.gcp import bigquery_read_it_test
+from apache_beam.io.gcp import bigquery_schema_tools
+from apache_beam.io.gcp import bigquery_tools
+from apache_beam.testing.util import assert_that
+from apache_beam.testing.util import equal_to
+
+# Protect against environments where bigquery library is not available.
+# pylint: disable=wrong-import-order, wrong-import-position
+# pylint: enable=wrong-import-order, wrong-import-position
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ReadUsingReadGbqTests(bigquery_read_it_test.BigQueryReadIntegrationTests):
+ @pytest.mark.it_postcommit
+ def test_ReadGbq(self):
+ from apache_beam.dataframe import convert
+ the_table = bigquery_tools.BigQueryWrapper().get_table(
+ project_id="apache-beam-testing",
+ dataset_id="beam_bigquery_io_test",
+ table_id="dfsqltable_3c7d6fd5_16e0460dfd0")
+ table = the_table.schema
+ utype = bigquery_schema_tools. \
+ generate_user_type_from_bq_schema(table)
+ with beam.Pipeline(argv=self.args) as p:
+ actual_df = p | apache_beam.dataframe.io.read_gbq(
+ table="apache-beam-testing:beam_bigquery_io_test."
+ "dfsqltable_3c7d6fd5_16e0460dfd0",
+ use_bqstorage_api=False)
+ assert_that(
+ convert.to_pcollection(actual_df),
+ equal_to([
+ utype(id=3, name='customer1', type='test'),
+ utype(id=1, name='customer1', type='test'),
+ utype(id=2, name='customer2', type='test'),
+ utype(id=4, name='customer2', type='test')
+ ]))
+
+ def test_ReadGbq_export_with_project(self):
+ from apache_beam.dataframe import convert
+ the_table = bigquery_tools.BigQueryWrapper().get_table(
+ project_id="apache-beam-testing",
+ dataset_id="beam_bigquery_io_test",
+ table_id="dfsqltable_3c7d6fd5_16e0460dfd0")
+ table = the_table.schema
+ utype = bigquery_schema_tools. \
+ generate_user_type_from_bq_schema(table)
+ with beam.Pipeline(argv=self.args) as p:
+ actual_df = p | apache_beam.dataframe.io.read_gbq(
+ table="dfsqltable_3c7d6fd5_16e0460dfd0",
+ dataset="beam_bigquery_io_test",
+ project_id="apache-beam-testing",
+ use_bqstorage_api=False)
+ assert_that(
+ convert.to_pcollection(actual_df),
+ equal_to([
+ utype(id=3, name='customer1', type='test'),
+ utype(id=1, name='customer1', type='test'),
+ utype(id=2, name='customer2', type='test'),
+ utype(id=4, name='customer2', type='test')
+ ]))
+
+ @pytest.mark.it_postcommit
+ def test_ReadGbq_direct_read(self):
+ from apache_beam.dataframe import convert
+ the_table = bigquery_tools.BigQueryWrapper().get_table(
+ project_id="apache-beam-testing",
+ dataset_id="beam_bigquery_io_test",
+ table_id="dfsqltable_3c7d6fd5_16e0460dfd0")
+ table = the_table.schema
+ utype = bigquery_schema_tools. \
+ generate_user_type_from_bq_schema(table)
+ with beam.Pipeline(argv=self.args) as p:
+ actual_df = p | apache_beam.dataframe.io.\
+ read_gbq(
+ table=
+ "apache-beam-testing:beam_bigquery_io_test."
+ "dfsqltable_3c7d6fd5_16e0460dfd0",
+ use_bqstorage_api=True)
+ assert_that(
+ convert.to_pcollection(actual_df),
+ equal_to([
+ utype(id=3, name='customer1', type='test'),
+ utype(id=1, name='customer1', type='test'),
+ utype(id=2, name='customer2', type='test'),
+ utype(id=4, name='customer2', type='test')
+ ]))
+
+ @pytest.mark.it_postcommit
+ def test_ReadGbq_direct_read_with_project(self):
+ from apache_beam.dataframe import convert
+ the_table = bigquery_tools.BigQueryWrapper().get_table(
+ project_id="apache-beam-testing",
+ dataset_id="beam_bigquery_io_test",
+ table_id="dfsqltable_3c7d6fd5_16e0460dfd0")
+ table = the_table.schema
+ utype = bigquery_schema_tools. \
+ generate_user_type_from_bq_schema(table)
+ with beam.Pipeline(argv=self.args) as p:
+ actual_df = p | apache_beam.dataframe.io.read_gbq(
+ table="dfsqltable_3c7d6fd5_16e0460dfd0",
+ dataset="beam_bigquery_io_test",
+ project_id="apache-beam-testing",
+ use_bqstorage_api=True)
+ assert_that(
+ convert.to_pcollection(actual_df),
+ equal_to([
+ utype(id=3, name='customer1', type='test'),
+ utype(id=1, name='customer1', type='test'),
+ utype(id=2, name='customer2', type='test'),
+ utype(id=4, name='customer2', type='test')
+ ]))
Review Comment:
It could be good to add one more test that actually applies some computation on the dataframe, just to test it end-to-end e.g:
```
beam_df = read_gbq(...)
actual_df = beam_df.groupby('name').count()
```
(Note in that case the 'name' column will be in the index, so you'll have to use `to_pcollection(actual_df, include_indexes=True)`)
Alternatively we could build a nice example pipeline that uses read_gbq with a computation and put it in `apache_beam.examples.dataframe`.
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -58,6 +58,28 @@
_DEFAULT_BYTES_CHUNKSIZE = 1 << 20
+def read_gbq(
+ table, dataset=None, project_id=None, use_bqstorage_api=False, **kwargs):
+ """This function reads data from a BigQuery source and outputs it into
+ a Beam deferred dataframe
+ (https://beam.apache.org/documentation/dsls/dataframes/overview/)
+ Please specify a table in the format 'PROJECT:dataset.table'
+ or use the table, dataset, and project_id args
+ to specify the table. If you would like to utilize the BigQuery
+ Storage API in ReadFromBigQuery,
+ please set use_bq_storage_api to True.
+ Otherwise, please set the flag to false or
+ leave it unspecified."""
Review Comment:
nit: Could you format this with "Args:" instead?
```suggestion
Args:
table: ...
dataset: ...
project_id: ...
use_bqstorage_api: ...
```
##########
sdks/python/apache_beam/io/gcp/dataframe_io_it_test.py:
##########
@@ -0,0 +1,146 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Unit tests for Dataframe sources and sinks."""
+# pytype: skip-file
+
+import logging
+import unittest
+
+import pytest
+
+import apache_beam as beam
+import apache_beam.io.gcp.bigquery
+from apache_beam.io.gcp import bigquery_read_it_test
+from apache_beam.io.gcp import bigquery_schema_tools
+from apache_beam.io.gcp import bigquery_tools
+from apache_beam.testing.util import assert_that
+from apache_beam.testing.util import equal_to
+
+# Protect against environments where bigquery library is not available.
+# pylint: disable=wrong-import-order, wrong-import-position
+# pylint: enable=wrong-import-order, wrong-import-position
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ReadUsingReadGbqTests(bigquery_read_it_test.BigQueryReadIntegrationTests):
Review Comment:
Why is this inheriting from `BigQueryReadIntegrationTests`?
##########
sdks/python/apache_beam/io/gcp/dataframe_io_it_test.py:
##########
@@ -0,0 +1,146 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Unit tests for Dataframe sources and sinks."""
+# pytype: skip-file
+
+import logging
+import unittest
+
+import pytest
+
+import apache_beam as beam
+import apache_beam.io.gcp.bigquery
+from apache_beam.io.gcp import bigquery_read_it_test
+from apache_beam.io.gcp import bigquery_schema_tools
+from apache_beam.io.gcp import bigquery_tools
+from apache_beam.testing.util import assert_that
+from apache_beam.testing.util import equal_to
+
+# Protect against environments where bigquery library is not available.
+# pylint: disable=wrong-import-order, wrong-import-position
+# pylint: enable=wrong-import-order, wrong-import-position
Review Comment:
These comments look out of place
##########
sdks/python/apache_beam/dataframe/io_test.py:
##########
@@ -410,5 +419,50 @@ def test_double_write(self):
set(self.read_all_lines(output + 'out2.csv*')))
+@unittest.skipIf(HttpError is None, 'GCP dependencies are not installed')
Review Comment:
Could you do an experiment to see what the error looks like if you try to use `read_gbq` without gcp dependencies installed? It should be sufficient to just make a new virtualenv, and install beam with `python -m pip install -e '.[test,dataframe]'` (note there's no gcp extra).
I'd like to make sure that the user gets a helpful error directing them to install GCP deps.
Even better would be to add a test that confirms this, it could be skipped unless GCP deps are _not_ installed.
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -58,6 +58,28 @@
_DEFAULT_BYTES_CHUNKSIZE = 1 << 20
+def read_gbq(
+ table, dataset=None, project_id=None, use_bqstorage_api=False, **kwargs):
+ """This function reads data from a BigQuery source and outputs it into
+ a Beam deferred dataframe
+ (https://beam.apache.org/documentation/dsls/dataframes/overview/)
Review Comment:
```suggestion
"""This function reads data from a BigQuery table and produces a :class:`~apache_beam.dataframe.frames.DeferredDataFrame`.
```
(this will make a link in the generated documentation, like you see [here](https://beam.apache.org/releases/pydoc/current/apache_beam.dataframe.io.html#module-apache_beam.dataframe.io))
##########
sdks/python/apache_beam/io/gcp/dataframe_io_it_test.py:
##########
@@ -0,0 +1,146 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Unit tests for Dataframe sources and sinks."""
Review Comment:
Another nit: I'd prefer we put this in `apache_beam.dataframe.io_it_test`
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1233363557
Run Python PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1233458025
Run Python PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1233576000
Run Python PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1230600125
thanks! changes made
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1234329792
Run Python PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a diff in pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on code in PR #22616:
URL: https://github.com/apache/beam/pull/22616#discussion_r941878880
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -58,6 +58,17 @@
_DEFAULT_BYTES_CHUNKSIZE = 1 << 20
+@frame_base.with_docs_from(pd)
+def read_gbq(table, *args, **kwargs):
+ """If you would like to use the 'DIRECT_READ' method ins ReadFromBigQuery,
+ please set use_bq_storage_api to True.
+ Otherwise, if you would like to use the 'EXPORT' method, please set
+ use_bq_storage_api to False, or leave it unspecified."""
Review Comment:
I could be swayed, but I think it would be preferable not to document how this is working under the hood (e.g. that it relies on `ReadFromBigQuery`), and instead just present the user-facing API.
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -58,6 +58,17 @@
_DEFAULT_BYTES_CHUNKSIZE = 1 << 20
+@frame_base.with_docs_from(pd)
Review Comment:
I'm not sure it makes sense to pull the docs from pandas in this case. We're diverging from pandas pretty significantly here: there are a lot of arguments we don't need (e.g. auth related ones), and we prefer a `table` argument rather than `query`. I'd suggest writing a whole new docstring (possibly drawing heavily from the pandas one).
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -756,3 +767,81 @@ def expand(self, pcoll):
| beam.Map(lambda file_result: file_result.file_name).with_output_types(
str)
}
+
+
+class _ReadGbq(beam.PTransform):
+ """Read data from BigQuery with output type 'BEAM_ROW',
+ then convert it into a deferred dataframe.
+
+ This PTransform wraps the Python ReadFromBigQuery PTransform,
+ and sets the output_type as 'BEAM_ROW' to convert
+ into a Beam Schema. Once applied to a pipeline object,
+ it is passed into the to_dataframe() function to convert the
+ PCollection into a deferred dataframe.
+
+ This PTransform currently does not support queries.
+ Note that all Python ReadFromBigQuery args can be passed in
+ to this PTransform, in addition to the args listed below.
+
+ Args:
+ table (str): The ID of the table. The ID must contain only
+ letters ``a-z``, ``A-Z``,
+ numbers ``0-9``, underscores ``_`` or white spaces.
+ Note that the table argument must contain the entire table
+ reference specified as: ``'PROJECT:DATASET.TABLE'``.
+ use_bq_storage_api (bool): The method to use to read from BigQuery.
+ It may be 'EXPORT' or
+ 'DIRECT_READ'. EXPORT invokes a BigQuery export request
+ (https://cloud.google.com/bigquery/docs/exporting-data).
+ 'DIRECT_READ' reads
+ directly from BigQuery storage using the BigQuery Read API
+ (https://cloud.google.com/bigquery/docs/reference/storage). If
+ unspecified or set to false, the default is currently utilized (EXPORT).
+ If the flag is set to true,
+ 'DIRECT_READ' will be utilized."""
+ def __init__(
+ self,
+ table=None,
Review Comment:
I don't see pandas' `project_id` parameter here, I think that's one that it makes sense to add and plumb through to ReadFromBigQuery. We could also add a Beam-specific one `dataset`.
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -756,3 +767,81 @@ def expand(self, pcoll):
| beam.Map(lambda file_result: file_result.file_name).with_output_types(
str)
}
+
+
+class _ReadGbq(beam.PTransform):
+ """Read data from BigQuery with output type 'BEAM_ROW',
+ then convert it into a deferred dataframe.
+
+ This PTransform wraps the Python ReadFromBigQuery PTransform,
+ and sets the output_type as 'BEAM_ROW' to convert
+ into a Beam Schema. Once applied to a pipeline object,
+ it is passed into the to_dataframe() function to convert the
+ PCollection into a deferred dataframe.
+
+ This PTransform currently does not support queries.
+ Note that all Python ReadFromBigQuery args can be passed in
+ to this PTransform, in addition to the args listed below.
+
+ Args:
+ table (str): The ID of the table. The ID must contain only
+ letters ``a-z``, ``A-Z``,
+ numbers ``0-9``, underscores ``_`` or white spaces.
+ Note that the table argument must contain the entire table
+ reference specified as: ``'PROJECT:DATASET.TABLE'``.
+ use_bq_storage_api (bool): The method to use to read from BigQuery.
+ It may be 'EXPORT' or
+ 'DIRECT_READ'. EXPORT invokes a BigQuery export request
+ (https://cloud.google.com/bigquery/docs/exporting-data).
+ 'DIRECT_READ' reads
+ directly from BigQuery storage using the BigQuery Read API
+ (https://cloud.google.com/bigquery/docs/reference/storage). If
+ unspecified or set to false, the default is currently utilized (EXPORT).
+ If the flag is set to true,
+ 'DIRECT_READ' will be utilized."""
+ def __init__(
+ self,
+ table=None,
+ index_col=None,
+ col_order=None,
+ reauth=None,
+ auth_local_webserver=None,
+ dialect=None,
+ location=None,
+ configuration=None,
+ credentials=None,
+ use_bqstorage_api=False,
+ max_results=None,
+ progress_bar_type=None):
+
+ self.table = table
+ self.index_col = index_col
+ self.col_order = col_order
+ self.reauth = reauth
+ self.auth_local_webserver = auth_local_webserver
+ self.dialect = dialect
+ self.location = location
+ self.configuration = configuration
+ self.credentials = credentials
+ self.use_bqstorage_api = use_bqstorage_api
+ self.max_results = max_results
+ self.progress_bar_type = progress_bar_type
+
+ if (self.index_col is not None or self.col_order is not None or
+ self.reauth is not None or self.auth_local_webserver is not None or
+ self.dialect is not None or self.location is not None or
+ self.configuration is not None or self.credentials is not None or
+ self.max_results is not None or self.progress_bar_type) is not None:
+ raise ValueError(
+ "Unsupported parameter entered in ReadGbq. "
+ "Please enter only supported parameters.")
Review Comment:
Instead of checking each unsupported parameter explicitly, could we capture all of these in `**kwargs`, (so the argument list becomes `(table=None, use_bqstorage_api=False, **kwargs)`) and then raise this error if kwargs is non-empty? Then it could also include a listing of the unsupported arguments, like `auth_local_webserver, progress_bar_type, ...`
##########
sdks/python/apache_beam/dataframe/io_test.py:
##########
@@ -410,5 +419,50 @@ def test_double_write(self):
set(self.read_all_lines(output + 'out2.csv*')))
+@unittest.skipIf(HttpError is None, 'GCP dependencies are not installed')
+class ReadGbqTransformTests(unittest.TestCase):
+ @mock.patch.object(BigQueryWrapper, 'get_table')
+ def test_bad_schema_public_api_direct_read(self, get_table):
+ fields = [
+ bigquery.TableFieldSchema(name='stn', type='DOUBLE', mode="NULLABLE"),
+ bigquery.TableFieldSchema(name='temp', type='FLOAT64', mode="REPEATED"),
+ bigquery.TableFieldSchema(name='count', type='INTEGER', mode=None)
+ ]
+ schema = bigquery.TableSchema(fields=fields)
+ table = apache_beam.io.gcp.internal.clients.bigquery. \
+ bigquery_v2_messages.Table(
+ schema=schema)
+ get_table.return_value = table
+
+ with self.assertRaisesRegex(ValueError,
+ "Encountered an unsupported type: 'DOUBLE'"):
+ p = apache_beam.Pipeline()
+ pipeline = p | apache_beam.dataframe.io.read_gbq(
+ table="dataset.sample_table", use_bqstorage_api=True)
+ pipeline
Review Comment:
Note the output here should be a deferred dataframe object, it's confusing to name it `pipeline` (`p` is actually a Pipeline). Regardless, I'd suggest assigning this to _ so pylint doesn't complain about an unassigned output or an unused variable:
```suggestion
_ = p | apache_beam.dataframe.io.read_gbq(
table="dataset.sample_table", use_bqstorage_api=True)
```
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -756,3 +767,81 @@ def expand(self, pcoll):
| beam.Map(lambda file_result: file_result.file_name).with_output_types(
str)
}
+
+
+class _ReadGbq(beam.PTransform):
+ """Read data from BigQuery with output type 'BEAM_ROW',
+ then convert it into a deferred dataframe.
+
+ This PTransform wraps the Python ReadFromBigQuery PTransform,
+ and sets the output_type as 'BEAM_ROW' to convert
+ into a Beam Schema. Once applied to a pipeline object,
+ it is passed into the to_dataframe() function to convert the
+ PCollection into a deferred dataframe.
+
+ This PTransform currently does not support queries.
+ Note that all Python ReadFromBigQuery args can be passed in
+ to this PTransform, in addition to the args listed below.
+
+ Args:
+ table (str): The ID of the table. The ID must contain only
+ letters ``a-z``, ``A-Z``,
+ numbers ``0-9``, underscores ``_`` or white spaces.
+ Note that the table argument must contain the entire table
+ reference specified as: ``'PROJECT:DATASET.TABLE'``.
+ use_bq_storage_api (bool): The method to use to read from BigQuery.
+ It may be 'EXPORT' or
+ 'DIRECT_READ'. EXPORT invokes a BigQuery export request
+ (https://cloud.google.com/bigquery/docs/exporting-data).
+ 'DIRECT_READ' reads
+ directly from BigQuery storage using the BigQuery Read API
+ (https://cloud.google.com/bigquery/docs/reference/storage). If
+ unspecified or set to false, the default is currently utilized (EXPORT).
+ If the flag is set to true,
+ 'DIRECT_READ' will be utilized."""
+ def __init__(
+ self,
+ table=None,
+ index_col=None,
+ col_order=None,
+ reauth=None,
+ auth_local_webserver=None,
+ dialect=None,
+ location=None,
+ configuration=None,
+ credentials=None,
+ use_bqstorage_api=False,
+ max_results=None,
+ progress_bar_type=None):
+
+ self.table = table
+ self.index_col = index_col
+ self.col_order = col_order
+ self.reauth = reauth
+ self.auth_local_webserver = auth_local_webserver
+ self.dialect = dialect
+ self.location = location
+ self.configuration = configuration
+ self.credentials = credentials
+ self.use_bqstorage_api = use_bqstorage_api
+ self.max_results = max_results
+ self.progress_bar_type = progress_bar_type
+
+ if (self.index_col is not None or self.col_order is not None or
+ self.reauth is not None or self.auth_local_webserver is not None or
+ self.dialect is not None or self.location is not None or
+ self.configuration is not None or self.credentials is not None or
+ self.max_results is not None or self.progress_bar_type) is not None:
+ raise ValueError(
+ "Unsupported parameter entered in ReadGbq. "
+ "Please enter only supported parameters.")
Review Comment:
I'd also suggest moving this to read_gbq, and having `_ReadGbq` just accept supported params.
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -756,3 +767,81 @@ def expand(self, pcoll):
| beam.Map(lambda file_result: file_result.file_name).with_output_types(
str)
}
+
+
+class _ReadGbq(beam.PTransform):
+ """Read data from BigQuery with output type 'BEAM_ROW',
+ then convert it into a deferred dataframe.
+
+ This PTransform wraps the Python ReadFromBigQuery PTransform,
+ and sets the output_type as 'BEAM_ROW' to convert
+ into a Beam Schema. Once applied to a pipeline object,
+ it is passed into the to_dataframe() function to convert the
+ PCollection into a deferred dataframe.
+
+ This PTransform currently does not support queries.
+ Note that all Python ReadFromBigQuery args can be passed in
+ to this PTransform, in addition to the args listed below.
+
+ Args:
+ table (str): The ID of the table. The ID must contain only
+ letters ``a-z``, ``A-Z``,
+ numbers ``0-9``, underscores ``_`` or white spaces.
+ Note that the table argument must contain the entire table
+ reference specified as: ``'PROJECT:DATASET.TABLE'``.
+ use_bq_storage_api (bool): The method to use to read from BigQuery.
+ It may be 'EXPORT' or
+ 'DIRECT_READ'. EXPORT invokes a BigQuery export request
+ (https://cloud.google.com/bigquery/docs/exporting-data).
+ 'DIRECT_READ' reads
+ directly from BigQuery storage using the BigQuery Read API
+ (https://cloud.google.com/bigquery/docs/reference/storage). If
+ unspecified or set to false, the default is currently utilized (EXPORT).
+ If the flag is set to true,
+ 'DIRECT_READ' will be utilized."""
+ def __init__(
+ self,
+ table=None,
+ index_col=None,
+ col_order=None,
+ reauth=None,
+ auth_local_webserver=None,
+ dialect=None,
+ location=None,
+ configuration=None,
+ credentials=None,
+ use_bqstorage_api=False,
+ max_results=None,
+ progress_bar_type=None):
+
+ self.table = table
+ self.index_col = index_col
+ self.col_order = col_order
+ self.reauth = reauth
+ self.auth_local_webserver = auth_local_webserver
+ self.dialect = dialect
+ self.location = location
+ self.configuration = configuration
+ self.credentials = credentials
+ self.use_bqstorage_api = use_bqstorage_api
+ self.max_results = max_results
+ self.progress_bar_type = progress_bar_type
+
+ if (self.index_col is not None or self.col_order is not None or
+ self.reauth is not None or self.auth_local_webserver is not None or
+ self.dialect is not None or self.location is not None or
+ self.configuration is not None or self.credentials is not None or
+ self.max_results is not None or self.progress_bar_type) is not None:
+ raise ValueError(
+ "Unsupported parameter entered in ReadGbq. "
+ "Please enter only supported parameters.")
+
+ def expand(self, root):
+ from apache_beam.dataframe import convert # avoid circular import
+ if self.use_bqstorage_api:
+ return convert.to_dataframe(
+ root | beam.io.ReadFromBigQuery(
+ table=self.table, method='DIRECT_READ', output_type='BEAM_ROW'))
+ return convert.to_dataframe(
+ root
+ | beam.io.ReadFromBigQuery(table=self.table, output_type='BEAM_ROW'))
Review Comment:
nit: you could just select the method inside the branching logic, and re-use the rest.
```suggestion
if self.use_bqstorage_api:
method = 'DIRECT_READ'
else:
method = 'EXPORT'
return convert.to_dataframe(
root
| beam.io.ReadFromBigQuery(table=self.table, method=method, output_type='BEAM_ROW'))
```
##########
sdks/python/apache_beam/io/gcp/bigquery_read_it_test.py:
##########
@@ -259,6 +259,43 @@ def test_table_schema_retrieve_with_direct_read(self):
]))
+class ReadUsingReadGbqTests(BigQueryReadIntegrationTests):
Review Comment:
I'd suggest making a new `dataframe.io_it_test` for these, rather than adding them here. Will that work?
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -756,3 +767,81 @@ def expand(self, pcoll):
| beam.Map(lambda file_result: file_result.file_name).with_output_types(
str)
}
+
+
+class _ReadGbq(beam.PTransform):
+ """Read data from BigQuery with output type 'BEAM_ROW',
+ then convert it into a deferred dataframe.
+
+ This PTransform wraps the Python ReadFromBigQuery PTransform,
+ and sets the output_type as 'BEAM_ROW' to convert
+ into a Beam Schema. Once applied to a pipeline object,
+ it is passed into the to_dataframe() function to convert the
+ PCollection into a deferred dataframe.
+
+ This PTransform currently does not support queries.
+ Note that all Python ReadFromBigQuery args can be passed in
+ to this PTransform, in addition to the args listed below.
Review Comment:
I don't think this is right, I don't see any logic to pass the ReadFromBigQuery args through.
##########
sdks/python/apache_beam/io/gcp/bigquery_read_it_test.py:
##########
@@ -259,6 +259,43 @@ def test_table_schema_retrieve_with_direct_read(self):
]))
+class ReadUsingReadGbqTests(BigQueryReadIntegrationTests):
+ @pytest.mark.it_postcommit
+ def test_ReadGbq(self):
+ with beam.Pipeline(argv=self.args) as p:
+ expected_df = (
+ p | apache_beam.io.gcp.bigquery.ReadFromBigQuery(
+ method=beam.io.ReadFromBigQuery.Method.EXPORT,
+ table="apache-beam-testing:"
+ "beam_bigquery_io_test."
+ "dfsqltable_3c7d6fd5_16e0460dfd0",
+ output_type='BEAM_ROW'))
+ with beam.Pipeline(argv=self.args) as p:
+ actual_df = p | apache_beam.dataframe.io.read_gbq(
+ table="apache-beam-testing:"
+ "beam_bigquery_io_test."
+ "dfsqltable_3c7d6fd5_16e0460dfd0")
+ assert_that(expected_df, equal_to(actual_df))
Review Comment:
It's a bit odd to have two pipelines like this and verify them against each other. I also wouldn't expect this to work with a deferred dataframe instance. Is it actually working?
I think it would be preferable to just have an explicit expected value, as in the other ITs. Can we do that instead?
##########
sdks/python/apache_beam/dataframe/io_test.py:
##########
@@ -410,5 +419,50 @@ def test_double_write(self):
set(self.read_all_lines(output + 'out2.csv*')))
+@unittest.skipIf(HttpError is None, 'GCP dependencies are not installed')
Review Comment:
What happens with read_gbq if GCP dependencies are not installed? Maybe we should not define it in that case?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1233458275
> @TheNeuralBit ptal and lmk if the io_it_test looks viable
Looks great! I can merge when we get CI green
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1223123396
Run RAT PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on a diff in pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on code in PR #22616:
URL: https://github.com/apache/beam/pull/22616#discussion_r944847100
##########
sdks/python/apache_beam/dataframe/io_test.py:
##########
@@ -410,5 +419,50 @@ def test_double_write(self):
set(self.read_all_lines(output + 'out2.csv*')))
+@unittest.skipIf(HttpError is None, 'GCP dependencies are not installed')
Review Comment:
I tried without this, and the tests passed locally but failed on Jenkins and thus I wasn't able to easily reproduce the issue. Can discuss this offline if you think it's worth looking further into.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1213513405
Run Python_PVR_Flink PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a diff in pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on code in PR #22616:
URL: https://github.com/apache/beam/pull/22616#discussion_r941876254
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -756,3 +767,81 @@ def expand(self, pcoll):
| beam.Map(lambda file_result: file_result.file_name).with_output_types(
str)
}
+
+
+class _ReadGbq(beam.PTransform):
+ """Read data from BigQuery with output type 'BEAM_ROW',
+ then convert it into a deferred dataframe.
+
+ This PTransform wraps the Python ReadFromBigQuery PTransform,
+ and sets the output_type as 'BEAM_ROW' to convert
+ into a Beam Schema. Once applied to a pipeline object,
+ it is passed into the to_dataframe() function to convert the
+ PCollection into a deferred dataframe.
+
+ This PTransform currently does not support queries.
+ Note that all Python ReadFromBigQuery args can be passed in
+ to this PTransform, in addition to the args listed below.
+
+ Args:
+ table (str): The ID of the table. The ID must contain only
+ letters ``a-z``, ``A-Z``,
+ numbers ``0-9``, underscores ``_`` or white spaces.
+ Note that the table argument must contain the entire table
+ reference specified as: ``'PROJECT:DATASET.TABLE'``.
+ use_bq_storage_api (bool): The method to use to read from BigQuery.
+ It may be 'EXPORT' or
+ 'DIRECT_READ'. EXPORT invokes a BigQuery export request
+ (https://cloud.google.com/bigquery/docs/exporting-data).
+ 'DIRECT_READ' reads
+ directly from BigQuery storage using the BigQuery Read API
+ (https://cloud.google.com/bigquery/docs/reference/storage). If
+ unspecified or set to false, the default is currently utilized (EXPORT).
+ If the flag is set to true,
+ 'DIRECT_READ' will be utilized."""
+ def __init__(
+ self,
+ table=None,
+ index_col=None,
+ col_order=None,
+ reauth=None,
+ auth_local_webserver=None,
+ dialect=None,
+ location=None,
+ configuration=None,
+ credentials=None,
+ use_bqstorage_api=False,
+ max_results=None,
+ progress_bar_type=None):
+
+ self.table = table
+ self.index_col = index_col
+ self.col_order = col_order
+ self.reauth = reauth
+ self.auth_local_webserver = auth_local_webserver
+ self.dialect = dialect
+ self.location = location
+ self.configuration = configuration
+ self.credentials = credentials
+ self.use_bqstorage_api = use_bqstorage_api
+ self.max_results = max_results
+ self.progress_bar_type = progress_bar_type
+
+ if (self.index_col is not None or self.col_order is not None or
+ self.reauth is not None or self.auth_local_webserver is not None or
+ self.dialect is not None or self.location is not None or
+ self.configuration is not None or self.credentials is not None or
+ self.max_results is not None or self.progress_bar_type) is not None:
+ raise ValueError(
+ "Unsupported parameter entered in ReadGbq. "
+ "Please enter only supported parameters.")
Review Comment:
Instead of checking each unsupported parameter explicitly, could we capture all of these in `**kwargs`, (so the argument list becomes `(table=None, use_bqstorage_api=False, **kwargs)`) and then raise this error if kwargs is non-empty? Then it could also include a listing of the unsupported arguments, like `auth_local_webserver, progress_bar_type, ...`
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a diff in pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on code in PR #22616:
URL: https://github.com/apache/beam/pull/22616#discussion_r942830917
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -58,6 +58,17 @@
_DEFAULT_BYTES_CHUNKSIZE = 1 << 20
+@frame_base.with_docs_from(pd)
Review Comment:
Yeah, the new docstring here should be all of the user-facing documentation. _ReadGbq doesn't need to have a docstring since it's a private implementation detail (it could still have one to document its purpose, but this is for Beam devs, not for users).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a diff in pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on code in PR #22616:
URL: https://github.com/apache/beam/pull/22616#discussion_r956055366
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -756,3 +784,59 @@ def expand(self, pcoll):
| beam.Map(lambda file_result: file_result.file_name).with_output_types(
str)
}
+
+
+class _ReadGbq(beam.PTransform):
+ """Read data from BigQuery with output type 'BEAM_ROW',
+ then convert it into a deferred dataframe.
+
+ This PTransform wraps the Python ReadFromBigQuery PTransform,
+ and sets the output_type as 'BEAM_ROW' to convert
+ into a Beam Schema. Once applied to a pipeline object,
+ it is passed into the to_dataframe() function to convert the
+ PCollection into a deferred dataframe.
+
+ This PTransform currently does not support queries.
+
+ Args:
+ table (str): The ID of the table. The ID must contain only
+ letters ``a-z``, ``A-Z``,
+ numbers ``0-9``, underscores ``_`` or white spaces.
+ Note that the table argument must contain the entire table
+ reference specified as: ``'PROJECT:DATASET.TABLE'``.
+ use_bq_storage_api (bool): The method to use to read from BigQuery.
+ It may be 'EXPORT' or
+ 'DIRECT_READ'. EXPORT invokes a BigQuery export request
+ (https://cloud.google.com/bigquery/docs/exporting-data).
+ 'DIRECT_READ' reads
+ directly from BigQuery storage using the BigQuery Read API
+ (https://cloud.google.com/bigquery/docs/reference/storage). If
+ unspecified or set to false, the default is currently utilized (EXPORT).
+ If the flag is set to true,
+ 'DIRECT_READ' will be utilized."""
Review Comment:
nit: I think the second line should be indented here
```suggestion
table (str): The ID of the table. The ID must contain only
letters ``a-z``, ``A-Z``,
numbers ``0-9``, underscores ``_`` or white spaces.
Note that the table argument must contain the entire table
reference specified as: ``'PROJECT:DATASET.TABLE'``.
use_bq_storage_api (bool): The method to use to read from BigQuery.
It may be 'EXPORT' or
'DIRECT_READ'. EXPORT invokes a BigQuery export request
(https://cloud.google.com/bigquery/docs/exporting-data).
'DIRECT_READ' reads
directly from BigQuery storage using the BigQuery Read API
(https://cloud.google.com/bigquery/docs/reference/storage). If
unspecified or set to false, the default is currently utilized (EXPORT).
If the flag is set to true,
'DIRECT_READ' will be utilized."""
```
##########
sdks/python/apache_beam/dataframe/io_it_test.py:
##########
@@ -0,0 +1,123 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
Review Comment:
```suggestion
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
```
I think this will make the RAT check happy. To confirm you could run `./gradlew rat` locally and inspect the output (`build/reports/rat/index.html`).
##########
sdks/python/apache_beam/dataframe/io.py:
##########
@@ -58,6 +58,28 @@
_DEFAULT_BYTES_CHUNKSIZE = 1 << 20
+def read_gbq(
+ table, dataset=None, project_id=None, use_bqstorage_api=False, **kwargs):
+ """This function reads data from a BigQuery source and outputs it into
+ a Beam deferred dataframe
+ (https://beam.apache.org/documentation/dsls/dataframes/overview/)
+ Please specify a table in the format 'PROJECT:dataset.table'
+ or use the table, dataset, and project_id args
+ to specify the table. If you would like to utilize the BigQuery
+ Storage API in ReadFromBigQuery,
+ please set use_bq_storage_api to True.
+ Otherwise, please set the flag to false or
+ leave it unspecified."""
+ if table is None:
+ raise ValueError("Please specify a BigQuery table to read from.")
+ elif len(kwargs) > 0:
+ raise ValueError(
+ "Unsupported parameter entered in read_gbq. Please enter only "
+ "supported parameters 'table', 'dataset', "
+ "'project_id', 'use_bqstorage_api'.")
Review Comment:
What about using `kwargs.keys()` as suggested previously?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1232200011
Run Python 3.8 PostCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] svetakvsundhar commented on pull request #22616: Add initial read_gbq wrapper
Posted by GitBox <gi...@apache.org>.
svetakvsundhar commented on PR #22616:
URL: https://github.com/apache/beam/pull/22616#issuecomment-1226198695
Run Python 3.8 PostCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org