diff --git a/sdks/python/apache_beam/io/gcp/bigquery_read_internal.py b/sdks/python/apache_beam/io/gcp/bigquery_read_internal.py index 0ca5c2e69a05..6841de1e26fb 100644 --- a/sdks/python/apache_beam/io/gcp/bigquery_read_internal.py +++ b/sdks/python/apache_beam/io/gcp/bigquery_read_internal.py @@ -150,7 +150,8 @@ def process(self, element): class CleanUpProjects(beam.DoFn): def process(self, unused_element, unused_signal, pipeline_details): - bq = bigquery_tools.BigQueryWrapper() + bq = bigquery_tools.BigQueryWrapper.from_pipeline_options( + input.pipeline.options) pipeline_details = pipeline_details[0] if 'temp_table_ref' in pipeline_details.keys(): temp_table_ref = pipeline_details['temp_table_ref'] @@ -230,7 +231,8 @@ def _get_temp_dataset(self): def process(self, element: 'ReadFromBigQueryRequest') -> Iterable[BoundedSource]: bq = bigquery_tools.BigQueryWrapper( - temp_dataset_id=self._get_temp_dataset().datasetId) + temp_dataset_id=self._get_temp_dataset().datasetId, + client=bigquery_tools.BigQueryWrapper._bigquery_client(self.options)) if element.query is not None: self._setup_temporary_dataset(bq, element)