def _to_dataframe_bqstorage(self, bqstorage_client, dtypes, progress_bar=None):
        """Use (faster, but billable) BQ Storage API to construct DataFrame."""
        if bigquery_storage_v1beta1 is None:
            raise ValueError(_NO_BQSTORAGE_ERROR)

        if "$" in self._table.table_id:
            raise ValueError(
                "Reading from a specific partition is not currently supported."
            )
        if "@" in self._table.table_id:
            raise ValueError(
                "Reading from a specific snapshot is not currently supported."
            )

        read_options = bigquery_storage_v1beta1.types.TableReadOptions()
        if self._selected_fields is not None:
            for field in self._selected_fields:
                read_options.selected_fields.append(field.name)

        session = bqstorage_client.create_read_session(
            self._table.to_bqstorage(),
            "projects/{}".format(self._project),
            read_options=read_options,
        )

        # We need to parse the schema manually so that we can rearrange the
        # columns.
        schema = json.loads(session.avro_schema.schema)
        columns = [field["name"] for field in schema["fields"]]

        # Avoid reading rows from an empty table. pandas.concat will fail on an
        # empty list.
        if not session.streams:
            return pandas.DataFrame(columns=columns)

        # Use _to_dataframe_finished to notify worker threads when to quit.
        # See: https://stackoverflow.com/a/29237343/101923
        self._to_dataframe_finished = False

        # Create a queue to track progress updates across threads.
        worker_queue = _NoopProgressBarQueue()
        progress_queue = None
        progress_thread = None
        if progress_bar is not None:
            worker_queue = queue.Queue()
            progress_queue = queue.Queue()
            progress_thread = threading.Thread(
                target=self._process_worker_updates, args=(worker_queue, progress_queue)
            )
            progress_thread.start()

        def get_frames(pool):
            frames = []

            # Manually submit jobs and wait for download to complete rather
            # than using pool.map because pool.map continues running in the
            # background even if there is an exception on the main thread.
            # See: https://github.com/googleapis/google-cloud-python/pull/7698
            not_done = [
                pool.submit(
                    self._to_dataframe_bqstorage_stream,
                    bqstorage_client,
                    dtypes,
                    columns,
                    session,
                    stream,
                    worker_queue,
                )
                for stream in session.streams
            ]

            while not_done:
                done, not_done = concurrent.futures.wait(
                    not_done, timeout=_PROGRESS_INTERVAL
                )
                frames.extend([future.result() for future in done])

                # The progress bar needs to update on the main thread to avoid
                # contention over stdout / stderr.
                self._process_progress_updates(progress_queue, progress_bar)

            return frames

        with concurrent.futures.ThreadPoolExecutor() as pool:
            try:
                frames = get_frames(pool)
            finally:
                # No need for a lock because reading/replacing a variable is
                # defined to be an atomic operation in the Python language
                # definition (enforced by the global interpreter lock).
                self._to_dataframe_finished = True

                # Shutdown all background threads, now that they should know to
                # exit early.
                pool.shutdown(wait=True)
                if progress_thread is not None:
                    progress_thread.join()

        # Update the progress bar one last time to close it.
        self._process_progress_updates(progress_queue, progress_bar)
        return pandas.concat(frames)
def _to_dataframe_bqstorage ( self , bqstorage_client , dtypes , progress_bar = None ) : if bigquery_storage_v1beta1 is None : raise ValueError ( _NO_BQSTORAGE_ERROR ) if "$" in self . _table . table_id : raise ValueError ( "Reading from a specific partition is not currently supported." ) if "@" in self . _table . table_id : raise ValueError ( "Reading from a specific snapshot is not currently supported." ) read_options = bigquery_storage_v1beta1 . types . TableReadOptions ( ) if self . _selected_fields is not None : for field in self . _selected_fields : read_options . selected_fields . append ( field . name ) session = bqstorage_client . create_read_session ( self . _table . to_bqstorage ( ) , "projects/{}" . format ( self . _project ) , read_options = read_options , ) schema = json . loads ( session . avro_schema . schema ) columns = [ field [ "name" ] for field in schema [ "fields" ] ] if not session . streams : return pandas . DataFrame ( columns = columns ) self . _to_dataframe_finished = False worker_queue = _NoopProgressBarQueue ( ) progress_queue = None progress_thread = None if progress_bar is not None : worker_queue = queue . Queue ( ) progress_queue = queue . Queue ( ) progress_thread = threading . Thread ( target = self . _process_worker_updates , args = ( worker_queue , progress_queue ) ) progress_thread . start ( ) def get_frames ( pool ) : frames = [ ] not_done = [ pool . submit ( self . _to_dataframe_bqstorage_stream , bqstorage_client , dtypes , columns , session , stream , worker_queue , ) for stream in session . streams ] while not_done : done , not_done = concurrent . futures . wait ( not_done , timeout = _PROGRESS_INTERVAL ) frames . extend ( [ future . result ( ) for future in done ] ) self . _process_progress_updates ( progress_queue , progress_bar ) return frames with concurrent . futures . ThreadPoolExecutor ( ) as pool : try : frames = get_frames ( pool ) finally : self . _to_dataframe_finished = True pool . shutdown ( wait = True ) if progress_thread is not None : progress_thread . join ( ) self . _process_progress_updates ( progress_queue , progress_bar ) return pandas . concat ( frames )
Use (faster, but billable) BQ Storage API to construct DataFrame.