BigQuery-Daten mithilfe der BigQuery Storage API in pandas herunterladen

Die BigQuery Storage API ermöglicht den schnellen Zugriff auf Daten, die in BigQuery gespeichert sind. Mit der BigQuery Storage API können Sie in BigQuery gespeicherte Daten herunterladen, um sie in Analysetools wie der pandas-Bibliothek für Python zu verwenden.

Ziele

In dieser Anleitung können Sie:

  • Abfrageergebnisse mithilfe der BigQuery Storage API aus den magischen IPython-Befehlen für BigQuery in einem Jupyter-Notebook in einen pandas-DataFrame herunterladen.
  • Abfrageergebnisse mithilfe der BigQuery-Clientbibliothek für Python in einen pandas-DataFrame herunterladen.
  • Tabellendaten aus BigQuery mithilfe der BigQuery-Clientbibliothek für Python in einen pandas-DataFrame herunterladen.
  • Tabellendaten aus BigQuery mithilfe der BigQuery Storage API-Clientbibliothek für Python in einen pandas-DataFrame herunterladen.

Kosten

BigQuery ist ein kostenpflichtiges Produkt. Für die von Ihnen ausgeführten Abfragen fallen BigQuery-Nutzungskosten an. Das erste Terabyte (1 TB) an verarbeiteten Abfragedaten pro Monat ist kostenlos. Weitere Informationen finden Sie auf der Seite "Preise" für BigQuery.

Die BigQuery Storage API ist ein kostenpflichtiges Produkt. Für die von Ihnen beim Herunterladen eines DataFrames gescannten Tabellendaten fallen Nutzungskosten an. Weitere Informationen finden Sie auf der Seite "Preise" für BigQuery.

Hinweis

Bevor Sie mit dieser Anleitung beginnen, rufen Sie die Google Cloud Console auf, erstellen Sie dort ein Projekt oder wählen es aus und aktivieren Sie die Abrechnung.

  1. Melden Sie sich bei Ihrem Google-Konto an.

    Wenn Sie noch kein Konto haben, registrieren Sie sich hier für ein neues Konto.

  2. Wählen Sie in der Cloud Console auf der Projektauswahlseite ein Cloud-Projekt aus oder erstellen Sie eines.

    Zur Projektauswahl

  3. Die Abrechnung für das Google Cloud-Projekt muss aktiviert sein. So prüfen Sie, ob die Abrechnung für Ihr Projekt aktiviert ist.

  4. BigQuery ist in neuen Projekten automatisch aktiviert. So aktivieren Sie BigQuery in einem vorhandenen Projekt: Aktivieren Sie die BigQuery, BigQuery Storage API APIs.

    Aktivieren Sie die APIs

  5. Richten Sie eine Python-Entwicklungsumgebung ein.
    Python einrichten
  6. Richten Sie die Authentifizierung für Ihre Entwicklungsumgebung ein.
    Authentifizierung einrichten

Sie sollten außerdem mit den magischen IPython-Befehlen für BigQuery, der BigQuery-Clientbibliothek und der Verwendung der Clientbibliothek mit pandas vertraut sein, bevor Sie diese Anleitung durcharbeiten.

Clientbibliotheken installieren

Installieren Sie mindestens Version 1.9.0 der BigQuery-Python-Clientbibliothek und die Python-Clientbibliothek der BigQuery Storage API.

PIP

Installieren Sie die Pakete google-cloud-bigquery und google-cloud-bigquery-storage.

pip install --upgrade google-cloud-bigquery[bqstorage,pandas]

Conda

Installieren Sie die Conda-Pakete für BigQuery und die BigQuery Storage API aus dem Community-Kanal conda-forge.

conda install -c conda-forge google-cloud-bigquery \
  google-cloud-bigquery-storage \
  pandas \
  pyarrow

Abfrageergebnisse mit den magischen IPython-Befehlen für BigQuery herunterladen

Starten Sie den Jupyter-Notebook-Server und erstellen Sie ein neues Jupyter-Notebook. Laden Sie die magischen IPython-Befehle für BigQuery mithilfe des magischen Befehls %load_ext.

%load_ext google.cloud.bigquery

Laden Sie große Abfrageergebnisse mit der BigQuery Storage API herunter, indem Sie dem magischen Befehl %%bigquery das Argument --use_bq_storage_api hinzufügen.

%%bigquery tax_forms --use_bqstorage_api
SELECT * FROM `bigquery-public-data.irs_990.irs_990_2012`

Wenn dieses Argument für kleine Abfrageergebnisse verwendet wird, nutzen die magischen Befehle die BigQuery API zum Herunterladen der Ergebnisse.

%%bigquery stackoverflow --use_bqstorage_api
SELECT
  CONCAT(
    'https://stackoverflow.com/questions/',
    CAST(id as STRING)) as url,
  view_count
FROM `bigquery-public-data.stackoverflow.posts_questions`
WHERE tags like '%google-bigquery%'
ORDER BY view_count DESC
LIMIT 10

Legen Sie für das Attribut context.use_bqstorage_api die Option True fest, um standardmäßig die BigQuery Storage API zu verwenden.

import google.cloud.bigquery.magics

google.cloud.bigquery.magics.context.use_bqstorage_api = True

Führen Sie nach dem Festlegen des Attributs context.use_bqstorage_api den magischen Befehl %%bigquery ohne weitere Argumente aus, um große Ergebnisse mit der BigQuery Storage API herunterzuladen.

%%bigquery tax_forms
SELECT * FROM `bigquery-public-data.irs_990.irs_990_2012`

Python-Clientbibliotheken verwenden

Python-Clients erstellen

Erstellen Sie ein BigQuery-Client-Objekt und einen BigQueryStorageClient mit dem folgenden Code.

import google.auth
from google.cloud import bigquery
from google.cloud import bigquery_storage_v1beta1

# Explicitly create a credentials object. This allows you to use the same
# credentials for both the BigQuery and BigQuery Storage clients, avoiding
# unnecessary API calls to fetch duplicate authentication tokens.
credentials, your_project_id = google.auth.default(
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)

# Make clients.
bqclient = bigquery.Client(
    credentials=credentials,
    project=your_project_id,
)
bqstorageclient = bigquery_storage_v1beta1.BigQueryStorageClient(
    credentials=credentials
)

Verwenden Sie die Python-Bibliothek "google-auth", um Anmeldedaten mit ausreichender Autorisierung für beide APIs zu erstellen. Zur Vermeidung einer zweimaligen Authentifizierung übergeben Sie an jeden Konstruktor ein Anmeldedatenobjekt.

Abfrageergebnisse mithilfe der BigQuery-Clientbibliothek herunterladen

Führen Sie mit der Methode query eine Abfrage aus. Rufen Sie die Methode to_dataframe auf. Damit werden die Ergebnisse mithilfe der BigQuery Storage API vollständig heruntergeladen.

# Download query results.
query_string = """
SELECT
CONCAT(
    'https://stackoverflow.com/questions/',
    CAST(id as STRING)) as url,
view_count
FROM `bigquery-public-data.stackoverflow.posts_questions`
WHERE tags like '%google-bigquery%'
ORDER BY view_count DESC
"""

dataframe = (
    bqclient.query(query_string)
    .result()
    .to_dataframe(bqstorage_client=bqstorageclient)
)
print(dataframe.head())

Tabellendaten mithilfe der BigQuery-Clientbibliothek herunterladen

Mit der Methode list_rows können Sie alle Zeilen in einer Tabelle herunterladen. Dabei wird ein RowIterator-Objekt zurückgegeben. Laden Sie Zeilen mit der BigQuery Storage API herunter. Dazu rufen Sie die Methode to_dataframe mit dem Argument bqstorage_client auf.

# Download a table.
table = bigquery.TableReference.from_string(
    "bigquery-public-data.utility_us.country_code_iso"
)
rows = bqclient.list_rows(
    table,
    selected_fields=[
        bigquery.SchemaField("country_name", "STRING"),
        bigquery.SchemaField("fips_code", "STRING"),
    ],
)
dataframe = rows.to_dataframe(bqstorage_client=bqstorageclient)
print(dataframe.head())

Tabellendaten mithilfe der BigQuery Storage API-Clientbibliothek herunterladen

Für eine präzise Steuerung der Filter und des Parallelismus verwenden Sie direkt die BigQuery Storage API-Clientbibliothek. Wenn nur einfache Zeilenfilter erforderlich sind, kann anstelle einer Abfrage eine BigQuery Storage API-Lesesitzung verwendet werden.

Erstellen Sie ein TableReference-Objekt mit der gewünschten zu lesenden Tabelle. Erstellen Sie ein TableReadOptions-Objekt, um Spalten oder Filterzeilen auszuwählen. Erstellen Sie mit der Methode create_read_session eine Lesesitzung.

Wenn sich in der Sitzung Streams befinden, lesen Sie die Zeilen mit der Methode read_rows. Rufen Sie die Methode to_dataframe mit dem Reader auf, um den gesamten Stream in einen Pandas DataFrame zu schreiben. Lesen Sie für eine bessere Leistung parallel aus mehreren Streams. In diesem Codebeispiel wird jedoch der Einfachheit halber nur aus einem einzigen Stream gelesen.

table = bigquery_storage_v1beta1.types.TableReference()
table.project_id = "bigquery-public-data"
table.dataset_id = "new_york_trees"
table.table_id = "tree_species"

# Select columns to read with read options. If no read options are
# specified, the whole table is read.
read_options = bigquery_storage_v1beta1.types.TableReadOptions()
read_options.selected_fields.append("species_common_name")
read_options.selected_fields.append("fall_color")

parent = "projects/{}".format(your_project_id)
session = bqstorageclient.create_read_session(
    table,
    parent,
    read_options=read_options,
    # This API can also deliver data serialized in Apache Avro format.
    # This example leverages Apache Arrow.
    format_=bigquery_storage_v1beta1.enums.DataFormat.ARROW,
    # We use a LIQUID strategy in this example because we only read from a
    # single stream. Consider BALANCED if you're consuming multiple streams
    # concurrently and want more consistent stream sizes.
    sharding_strategy=(
        bigquery_storage_v1beta1.enums.ShardingStrategy.LIQUID
    ),
)

# This example reads from only a single stream. Read from multiple streams
# to fetch data faster. Note that the session may not contain any streams
# if there are no rows to read.
stream = session.streams[0]
position = bigquery_storage_v1beta1.types.StreamPosition(stream=stream)
reader = bqstorageclient.read_rows(position)

# Parse all Avro blocks and create a dataframe. This call requires a
# session, because the session contains the schema for the row blocks.
dataframe = reader.to_dataframe(session)
print(dataframe.head())

Quellcode für alle Beispiele

Klicken Sie hier, um sich den vollständigen Quellcode für alle Beispiele anzusehen.

import google.auth
from google.cloud import bigquery
from google.cloud import bigquery_storage_v1beta1

# Explicitly create a credentials object. This allows you to use the same
# credentials for both the BigQuery and BigQuery Storage clients, avoiding
# unnecessary API calls to fetch duplicate authentication tokens.
credentials, your_project_id = google.auth.default(
    scopes=["https://www.googleapis.com/auth/cloud-platform"]
)

# Make clients.
bqclient = bigquery.Client(
    credentials=credentials,
    project=your_project_id,
)
bqstorageclient = bigquery_storage_v1beta1.BigQueryStorageClient(
    credentials=credentials
)
# Download a table.
table = bigquery.TableReference.from_string(
    "bigquery-public-data.utility_us.country_code_iso"
)
rows = bqclient.list_rows(
    table,
    selected_fields=[
        bigquery.SchemaField("country_name", "STRING"),
        bigquery.SchemaField("fips_code", "STRING"),
    ],
)
dataframe = rows.to_dataframe(bqstorage_client=bqstorageclient)
print(dataframe.head())
# Download query results.
query_string = """
SELECT
CONCAT(
    'https://stackoverflow.com/questions/',
    CAST(id as STRING)) as url,
view_count
FROM `bigquery-public-data.stackoverflow.posts_questions`
WHERE tags like '%google-bigquery%'
ORDER BY view_count DESC
"""

dataframe = (
    bqclient.query(query_string)
    .result()
    .to_dataframe(bqstorage_client=bqstorageclient)
)
print(dataframe.head())
table = bigquery_storage_v1beta1.types.TableReference()
table.project_id = "bigquery-public-data"
table.dataset_id = "new_york_trees"
table.table_id = "tree_species"

# Select columns to read with read options. If no read options are
# specified, the whole table is read.
read_options = bigquery_storage_v1beta1.types.TableReadOptions()
read_options.selected_fields.append("species_common_name")
read_options.selected_fields.append("fall_color")

parent = "projects/{}".format(your_project_id)
session = bqstorageclient.create_read_session(
    table,
    parent,
    read_options=read_options,
    # This API can also deliver data serialized in Apache Avro format.
    # This example leverages Apache Arrow.
    format_=bigquery_storage_v1beta1.enums.DataFormat.ARROW,
    # We use a LIQUID strategy in this example because we only read from a
    # single stream. Consider BALANCED if you're consuming multiple streams
    # concurrently and want more consistent stream sizes.
    sharding_strategy=(
        bigquery_storage_v1beta1.enums.ShardingStrategy.LIQUID
    ),
)

# This example reads from only a single stream. Read from multiple streams
# to fetch data faster. Note that the session may not contain any streams
# if there are no rows to read.
stream = session.streams[0]
position = bigquery_storage_v1beta1.types.StreamPosition(stream=stream)
reader = bqstorageclient.read_rows(position)

# Parse all Avro blocks and create a dataframe. This call requires a
# session, because the session contains the schema for the row blocks.
dataframe = reader.to_dataframe(session)
print(dataframe.head())

Bereinigen

So vermeiden Sie, dass Ihrem Google Cloud Platform-Konto die in dieser Anleitung verwendeten Ressourcen in Rechnung gestellt werden:

Löschen Sie das Projekt. Sie haben in dieser Anleitung keine BigQuery-Ressourcen erstellt. Durch das Löschen Ihres Projekts werden jedoch alle anderen von Ihnen erstellten Ressourcen entfernt.

  1. Wechseln Sie in der Cloud Console zur Seite Ressourcen verwalten.

    Zur Seite "Ressourcen verwalten"

  2. Wählen Sie in der Projektliste das Projekt aus, das Sie löschen möchten, und klicken Sie dann auf Löschen .
  3. Geben Sie im Dialogfeld die Projekt-ID ein und klicken Sie auf Beenden, um das Projekt zu löschen.

Weitere Informationen