Blog
Standpunkt

Wie man BigQuery APIs für Anfänger verwendet

Dieser Leitfaden zeigt Schritt für Schritt, wie man BigQuery APIs nutzt: Daten verarbeiten, Pipelines bauen, Anwendungen integrieren & Ressourcen verwalten.

Leo Schulz

Leo Schulz

10 September 2025

Wie man BigQuery APIs für Anfänger verwendet

Apidog für Unternehmen

On-Premises Bereitstellung

SSO & RBAC

SOC 2 konform

Apidog Enterprise entdecken

Google BigQuery hat die Art und Weise, wie Unternehmen Datenanalysen im großen Maßstab durchführen, revolutioniert. Seine serverlose Architektur, Skalierbarkeit und die vertraute SQL-Oberfläche machen es zu einem leistungsstarken Werkzeug, um Erkenntnisse aus riesigen Datensätzen zu gewinnen. Während die Interaktion mit BigQuery über die Google Cloud Console oder das Befehlszeilen-Tool bq üblich ist, wird die wahre Leistungsfähigkeit der Automatisierung, Integration und benutzerdefinierten Anwendungsentwicklung durch seinen umfassenden Satz von Application Programming Interfaces (APIs) freigesetzt.

Dieser Leitfaden bietet eine schrittweise Erkundung, wie man BigQuery-APIs verwendet, um programmatisch mit Ihrem Data Warehouse zu interagieren, Datenpipelines zu erstellen, BigQuery in Ihre Anwendungen zu integrieren und Ressourcen effektiv zu verwalten. Wir werden die verschiedenen Arten von verfügbaren APIs behandeln, wie Sie Ihre Umgebung einrichten, praktische Beispiele mit der Python-Client-Bibliothek und spezialisierte APIs für erweiterte Anwendungsfälle vorstellen.

💡
Want a great API Testing tool that generates beautiful API Documentation?

Want an integrated, All-in-One platform for your Developer Team to work together with maximum productivity?

Apidog delivers all your demands, and replaces Postman at a much more affordable price!
button

BigQuery-APIs verstehen

Bevor Sie in den Code eintauchen, ist es entscheidend, die Kernkonzepte und die verschiedenen Möglichkeiten, wie Sie programmatisch mit BigQuery interagieren können, zu verstehen.

Kernkonzepte von BigQuery:

Arten von BigQuery-APIs:

BigQuery bietet verschiedene Möglichkeiten, um programmatisch mit seinen Diensten zu interagieren:

REST API: Dies ist die grundlegende API, die auf HTTP und JSON basiert. Sie bietet direkten Zugriff auf BigQuery-Ressourcen und -Operationen. Sie können mit ihr über Standard-HTTP-Anforderungen (GET, POST, PUT, DELETE) interagieren, die auf bestimmte Endpunkte abzielen (z. B. https://bigquery.googleapis.com/bigquery/v2/projects/{projectId}/datasets). Obwohl sie leistungsstark ist und eine detaillierte Kontrolle bietet, erfordert die direkte Verwendung der REST API die manuelle Handhabung von Authentifizierung, Anforderungsformatierung, Antwortparsing und Fehlerbehandlung. Die Authentifizierung beinhaltet typischerweise OAuth 2.0-Zugriffstoken.

Client Libraries: Google stellt High-Level-Client-Bibliotheken für verschiedene gängige Programmiersprachen bereit (einschließlich Python, Java, Go, Node.js, C#, PHP, Ruby). Diese Bibliotheken umschließen die zugrunde liegende REST API und bieten eine idiomatischere, entwicklerfreundlichere Erfahrung. Sie vereinfachen gängige Aufgaben, handhaben die Authentifizierung (oft automatisch über Application Default Credentials), verwalten Wiederholungen und reduzieren die Menge an Boilerplate-Code, den Sie schreiben müssen. Dies ist der empfohlene Ansatz für die meisten Anwendungsentwicklungen.

Specialized APIs: Für bestimmte Hochleistungs- oder Spezialaufgaben bietet BigQuery dedizierte APIs:

Einrichten Ihrer Umgebung

Bevor Sie mit API-Aufrufen beginnen können, müssen Sie Ihre lokale oder Serverumgebung konfigurieren.

Voraussetzungen:

  1. Google Cloud Account: Sie benötigen ein aktives Google Cloud-Konto.
  2. Google Cloud Project: Erstellen Sie ein neues Projekt oder wählen Sie ein bestehendes in der Google Cloud Console aus. Notieren Sie sich Ihre Projekt-ID.
  3. BigQuery API aktivieren: Stellen Sie sicher, dass die BigQuery API für Ihr Projekt aktiviert ist. Sie können dies über die Cloud Console tun (APIs & Services > Library > Suchen Sie nach "BigQuery API" > Aktivieren). Möglicherweise müssen Sie auch andere APIs wie die BigQuery Storage Read API oder BigQuery Connection API aktivieren, je nach Anwendungsfall.
  4. Abrechnung: Stellen Sie sicher, dass die Abrechnung für Ihr Projekt aktiviert ist. BigQuery-Operationen verursachen Kosten, die auf Datenspeicherung, verarbeiteter Analyse und Streaming-Inserts basieren.

Authentifizierung:

Ihre Anwendung muss sich bei Google Cloud authentifizieren, um ihre Identität und Autorisierung für den Zugriff auf BigQuery-Ressourcen nachzuweisen. Die empfohlene Methode für die meisten Szenarien sind Application Default Credentials (ADC).

  1. Erstellen Sie ein Service Account in der Cloud Console (IAM & Admin > Service Accounts).
  2. Gewähren Sie dem Service Account die erforderlichen BigQuery-Rollen (z. B. BigQuery Data Editor, BigQuery Job User, BigQuery User).
  3. Laden Sie die Schlüsseldatei des Service Accounts herunter (JSON-Format).
  4. Legen Sie die Umgebungsvariable GOOGLE_APPLICATION_CREDENTIALS auf den absoluten Pfad der heruntergeladenen JSON-Schlüsseldatei fest. Client-Bibliotheken verwenden diese Schlüsseldatei automatisch für die Authentifizierung, wenn die Umgebungsvariable festgelegt ist.

Installieren von Client-Bibliotheken (Python-Beispiel):

Wir konzentrieren uns auf Python für unsere Beispiele. Sie können die erforderlichen Bibliotheken mit pip installieren:

pip install google-cloud-bigquery
# Optional: Install storage API library for faster reads
pip install google-cloud-bigquery-storage
# Optional: Install pandas integration and db-dtypes for better type handling
pip install pandas db-dtypes pyarrow

Stellen Sie sicher, dass Sie Python installiert haben (Version 3.7+ empfohlen für die neuesten Bibliotheksfunktionen).

Verwenden der BigQuery-Client-Bibliothek (Python-Beispiele)

Lassen Sie uns nun gängige BigQuery-Operationen mit der Python-Bibliothek google-cloud-bigquery untersuchen.

1. Importieren und Initialisieren des Clients:

Importieren Sie zuerst die Bibliothek. Erstellen Sie dann eine Client-Instanz. Wenn ADC korrekt konfiguriert ist, authentifiziert sich der Client automatisch.

from google.cloud import bigquery
import pandas as pd

# Construct a BigQuery client object.
# If GOOGLE_APPLICATION_CREDENTIALS is set, it uses the service account.
# If gcloud auth application-default login was run, it uses those credentials.
# If running on GCP infra, it uses the instance's service account.
client = bigquery.Client()

# You can explicitly specify the project ID if needed,
# otherwise it often infers from the environment/ADC credentials.
# client = bigquery.Client(project='your-project-id')

print("Client created successfully.")

2. Ausführen von Abfragen:

Die häufigste Operation ist das Ausführen von SQL-Abfragen.

# Define your SQL query
query = """
    SELECT name, SUM(number) as total_people
    FROM `bigquery-public-data.usa_names.usa_1910_2013`
    WHERE state = 'TX'
    GROUP BY name, state
    ORDER BY total_people DESC
    LIMIT 10
"""

# Make an API request and wait for the job to complete.
query_job = client.query(query)  # API request
print(f"Started job: {query_job.job_id}")

# Wait for the job to complete and get results.
rows = query_job.result()  # Waits for job to complete.

print("\nTop 10 names in TX (1910-2013):")
for row in rows:
    # Row values can be accessed by field name or index.
    print(f"Name: {row.name}, Count: {row['total_people']}") # Access by attribute or key

# Convert results to a Pandas DataFrame
df = rows.to_dataframe()
print("\nResults as Pandas DataFrame:")
print(df.head())

query = """
    SELECT corpus, COUNT(word) as distinct_words
    FROM `bigquery-public-data.samples.shakespeare`
    GROUP BY corpus
    ORDER BY distinct_words DESC;
"""
job_config = bigquery.QueryJobConfig(
    # Use standard SQL syntax for queries.
    use_legacy_sql=False
)

# Start the query, passing in the extra configuration.
query_job = client.query(query, job_config=job_config) # Does not wait
print(f"Started asynchronous job: {query_job.job_id}")

# --- Later in your application ---
# Check job status (optional)
# from google.cloud.exceptions import NotFound
# try:
#     job = client.get_job(query_job.job_id, location=query_job.location)
#     print(f"Job {job.job_id} status: {job.state}")
#     if job.state == "DONE":
#         if job.error_result:
#             print(f"Job failed: {job.errors}")
#         else:
#             results = job.result() # Get results
#             print("Results fetched.")
#             # Process results...
# except NotFound:
#     print(f"Job {query_job.job_id} not found.")

# Or simply wait for completion when needed
results = query_job.result() # This will block until the job is done
print("Asynchronous job completed.")
for row in results:
    print(f"Corpus: {row.corpus}, Distinct Words: {row.distinct_words}")


from google.cloud.bigquery import ScalarQueryParameter, ArrayQueryParameter, StructQueryParameter, QueryJobConfig

# Example: Find names starting with a specific prefix in a given state
state_param = "NY"
prefix_param = "Ma"
min_count_param = 1000

query = """
    SELECT name, SUM(number) as total_people
    FROM `bigquery-public-data.usa_names.usa_1910_2013`
    WHERE state = @state_abbr AND name LIKE @name_prefix
    GROUP BY name
    HAVING total_people >= @min_count
    ORDER BY total_people DESC;
"""

job_config = QueryJobConfig(
    query_parameters=[
        ScalarQueryParameter("state_abbr", "STRING", state_param),
        # Use 'val%' for LIKE operator
        ScalarQueryParameter("name_prefix", "STRING", f"{prefix_param}%"),
        ScalarQueryParameter("min_count", "INT64", min_count_param),
    ]
)

query_job = client.query(query, job_config=job_config)
print(f"Started parameterized query job: {query_job.job_id}")

rows = query_job.result()

print(f"\nNames starting with '{prefix_param}' in {state_param} with >= {min_count_param} people:")
for row in rows:
    print(f"Name: {row.name}, Count: {row.total_people}")

3. Verwalten von Datensätzen:

Sie können Datensätze erstellen, auflisten, Details abrufen und löschen.

# Define dataset ID and location
project_id = client.project
dataset_id = f"{project_id}.my_new_dataset"
dataset_location = "US" # e.g., "US", "EU", "asia-northeast1"

# Construct a full Dataset object to send to the API.
dataset = bigquery.Dataset(dataset_id)
dataset.location = dataset_location
dataset.description = "Dataset created via Python client library"

try:
    # Make an API request to create the dataset.
    dataset = client.create_dataset(dataset, timeout=30)  # Make an API request.
    print(f"Created dataset {client.project}.{dataset.dataset_id}")

    # List datasets in the project
    print("\nDatasets in project:")
    datasets = list(client.list_datasets()) # API request
    if datasets:
        for ds in datasets:
            print(f"\t{ds.dataset_id}")
    else:
        print(f"\t{client.project} project does not contain any datasets.")

    # Get dataset info
    retrieved_dataset = client.get_dataset(dataset_id) # API request
    print(f"\nRetrieved dataset info for {dataset_id}:")
    print(f"\tDescription: {retrieved_dataset.description}")
    print(f"\tLocation: {retrieved_dataset.location}")

except Exception as e:
    print(f"Error during dataset operations: {e}")

finally:
    # Clean up: Delete the dataset
    try:
        client.delete_dataset(
            dataset_id, delete_contents=True, not_found_ok=True
        )  # API request
        print(f"\nSuccessfully deleted dataset '{dataset_id}'.")
    except Exception as e:
         print(f"Error deleting dataset {dataset_id}: {e}")

4. Verwalten von Tabellen:

Für Tabellen gibt es ähnliche Operationen: Erstellen von Tabellen (Definieren des Schemas), Laden von Daten, Abrufen von Metadaten und Löschen von Tabellen.

# Using the previously created dataset ID (ensure it exists or remove deletion step above)
dataset_id_for_table = "my_new_dataset" # Use a valid dataset ID
table_id = f"{client.project}.{dataset_id_for_table}.my_new_table"

# Define the schema
schema = [
    bigquery.SchemaField("full_name", "STRING", mode="REQUIRED"),
    bigquery.SchemaField("age", "INTEGER", mode="REQUIRED"),
    bigquery.SchemaField("email", "STRING", mode="NULLABLE"),
]

# Create the table
table = bigquery.Table(table_id, schema=schema)
try:
    # Ensure dataset exists first
    client.create_dataset(dataset_id_for_table, exists_ok=True)

    table = client.create_table(table)  # API request
    print(
        f"Created table {table.project}.{table.dataset_id}.{table.table_id}"
    )

    # --- Loading Data (Example: from Pandas DataFrame) ---
    data = {'full_name': ['Alice Smith', 'Bob Johnson'],
            'age': [30, 45],
            'email': ['alice@example.com', None]}
    dataframe = pd.DataFrame(data)

    job_config = bigquery.LoadJobConfig(
        # Specify schema is recommended, ensures proper types
        schema=schema,
        # Optional: overwrite table data
        write_disposition="WRITE_TRUNCATE",
        # Or append: write_disposition="WRITE_APPEND",
    )

    load_job = client.load_table_from_dataframe(
        dataframe, table_id, job_config=job_config
    )  # API request
    print(f"Starting job {load_job.job_id} to load data from DataFrame")

    load_job.result()  # Waits for the job to complete.
    print("DataFrame load job finished.")

    destination_table = client.get_table(table_id) # API request
    print(f"Loaded {destination_table.num_rows} rows into table {table_id}")

    # --- Loading Data (Example: from Google Cloud Storage URI) ---
    # Assume a CSV file gs://your-bucket/data.csv exists with compatible data
    # uri = "gs://your-bucket/data.csv"
    # job_config_gcs = bigquery.LoadJobConfig(
    #     schema=schema,
    #     skip_leading_rows=1, # Skip header row
    #     source_format=bigquery.SourceFormat.CSV,
    #     write_disposition="WRITE_APPEND", # Append to existing data
    # )
    # load_job_gcs = client.load_table_from_uri(
    #     uri, table_id, job_config=job_config_gcs
    # )
    # print(f"Starting job {load_job_gcs.job_id} to load data from GCS")
    # load_job_gcs.result()
    # print("GCS load job finished.")
    # destination_table = client.get_table(table_id)
    # print(f"Total rows after GCS load: {destination_table.num_rows}")

except Exception as e:
    print(f"Error during table operations: {e}")

finally:
    # Clean up: Delete the table
    try:
        client.delete_table(table_id, not_found_ok=True)  # API request
        print(f"Successfully deleted table '{table_id}'.")
        # Optionally delete the dataset again if it was just for this example
        # client.delete_dataset(dataset_id_for_table, delete_contents=True, not_found_ok=True)
        # print(f"Successfully deleted dataset '{dataset_id_for_table}'.")
    except Exception as e:
        print(f"Error deleting table {table_id}: {e}")

5. Arbeiten mit Jobs:

Alle asynchronen Operationen (Abfrage, Laden, Exportieren, Kopieren) erstellen Job-Ressourcen. Sie können diese Jobs auflisten und verwalten.

# List recent jobs
print("\nRecent BigQuery Jobs:")
for job in client.list_jobs(max_results=10): # API request
    print(f"Job ID: {job.job_id}, Type: {job.job_type}, State: {job.state}, Created: {job.created}")

# Get a specific job (replace with a valid job ID from previous runs)
# try:
#     job_id_to_get = "..." # Replace with a real job ID
#     location = "US"      # Replace with the job's location if not default
#     retrieved_job = client.get_job(job_id_to_get, location=location) # API request
#     print(f"\nDetails for job {retrieved_job.job_id}:")
#     print(f"\tState: {retrieved_job.state}")
#     if retrieved_job.error_result:
#         print(f"\tError: {retrieved_job.errors}")
# except NotFound:
#     print(f"Job {job_id_to_get} not found.")
# except Exception as e:
#     print(f"Error retrieving job: {e}")

Nutzung spezialisierter APIs (Konzepte und Anwendungsfälle)

Während die Kern-Client-Bibliothek viele Anwendungsfälle abdeckt, bieten spezialisierte APIs eine verbesserte Leistung oder Funktionalität für bestimmte Aufgaben.

1. BigQuery Storage Read API (Python):

# Requires: pip install google-cloud-bigquery-storage pyarrow pandas db-dtypes

from google.cloud import bigquery_storage_v1
from google.cloud.bigquery_storage_v1 import types, GetDataStreamRequest

# --- Using Pandas read_gbq (Simplest integration) ---
# This automatically uses the Storage API if installed and beneficial
# table_id_read = "bigquery-public-data.usa_names.usa_1910_2013"
# cols_to_read = ["name", "number", "state"]
# row_filter = "state = 'CA' AND number > 5000"
#
# try:
#      df_storage = pd.read_gbq(
#          table_id_read,
#          project_id=client.project,
#          columns=cols_to_read,
#          row_filter=row_filter,
#          use_bqstorage_api=True, # Explicitly request Storage API
#          progress_bar_type='tqdm' # Optional progress bar
#      )
#      print("\nRead data using Storage API via pandas.read_gbq:")
#      print(df_storage.head())
#      print(f"Read {len(df_storage)} rows.")
# except Exception as e:
#      print(f"Error reading with Storage API via read_gbq: {e}")


# --- Manual Storage API Usage (More Control) ---
# bqstorageclient = bigquery_storage_v1.BigQueryReadClient()
# table = f"projects/{project_id}/datasets/{dataset_id}/tables/{table_name}" # Replace with your table details

# requested_session = types.ReadSession(
#     table=table,
#     data_format=types.DataFormat.ARROW,
#     read_options=types.ReadSession.TableReadOptions(
#         selected_fields=["col1", "col2"], # Specify columns
#         row_restriction="col1 > 100"     # Specify filter
#     ),
# )
# parent = f"projects/{project_id}"

# read_session = bqstorageclient.create_read_session(
#     parent=parent,
#     read_session=requested_session,
#     max_stream_count=1, # Request number of parallel streams
# )

# stream = read_session.streams[0]
# reader = bqstorageclient.read_rows(stream.name)
# frames = [message.arrow_record_batch for message in reader.messages()]
# if frames:
#     arrow_table = pa.Table.from_batches(frames)
#     df_manual = arrow_table.to_pandas()
#     print("\nRead data manually using Storage API:")
#     print(df_manual.head())
# else:
#     print("No data read using manual Storage API.")

2. BigQuery Connection API:

  1. Verwenden Sie die API (oder die Cloud Console/das bq-Tool), um eine Connection-Ressource zu erstellen und den externen Quelltyp und die Details anzugeben.
  2. Gewähren Sie dem Dienstkonto der Verbindung die entsprechenden Berechtigungen für die externe Ressource (z. B. die Rolle "Cloud SQL User").
  3. Verwenden Sie die Funktion EXTERNAL_QUERY("connection_id", "external_sql_query") innerhalb Ihres BigQuery SQL.

3. Analytics Hub API:

4. BigLake API:

Direkte Verwendung der REST API

Während Client-Bibliotheken im Allgemeinen bevorzugt werden, können Sie die REST API direkt verwenden, wenn:

Anforderungen stellen:

Sie verwenden typischerweise einen HTTP-Client (wie die curl- oder die requests-Bibliothek von Python). Sie müssen:

  1. Ein OAuth 2.0-Zugriffstoken abrufen (z. B. mit gcloud auth print-access-token).
  2. Die richtige API-Endpunkt-URL erstellen.
  3. Den JSON-Anforderungstext gemäß der Spezifikation der API-Methode erstellen.
  4. Das Zugriffstoken in den Header Authorization: Bearer <token> einfügen.
  5. Die HTTP-Antwort verarbeiten (Statuscodes, JSON-Parsing, Fehlermeldungen).

Beispiel: Ausführen einer Abfrage über REST (jobs.query)

# 1. Get Access Token
TOKEN=$(gcloud auth print-access-token)

# 2. Define Project ID and Request Body
PROJECT_ID="your-project-id" # Replace with your project ID
REQUEST_BODY=$(cat <<EOF
{
  "query": "SELECT name, SUM(number) as total_people FROM \`bigquery-public-data.usa_names.usa_1910_2013\` WHERE state = 'CA' GROUP BY name ORDER BY total_people DESC LIMIT 5;",
  "useLegacySql": false
}
EOF
)

# 3. Make the API Call using curl
curl -X POST \
  "https://bigquery.googleapis.com/bigquery/v2/projects/${PROJECT_ID}/jobs" \
  -H "Authorization: Bearer ${TOKEN}" \
  -H "Content-Type: application/json; charset=utf-8" \
  -d "${REQUEST_BODY}"

# The response will contain job information, including the job ID.
# You would then need to make subsequent calls to jobs.getQueryResults
# using the job ID to retrieve the actual data rows once the job completes.

Dieses Beispiel initiiert nur den Abfragejob. Das Abrufen von Ergebnissen erfordert das Abfragen des Jobstatus und dann das Aufrufen des Endpunkts jobs.getQueryResults. Dies unterstreicht die zusätzlichen Schritte im Vergleich zu Client-Bibliotheken.

Best Practices und Tipps

Explore more

Fathom-R1-14B: Fortschrittliches KI-Argumentationsmodell aus Indien

Fathom-R1-14B: Fortschrittliches KI-Argumentationsmodell aus Indien

Künstliche Intelligenz wächst rasant. FractalAIResearch/Fathom-R1-14B (14,8 Mrd. Parameter) glänzt in Mathe & Logik.

5 June 2025

Cursor 1.0 mit BugBot: KI-gestütztes Automatisierungstest-Tool ist da:

Cursor 1.0 mit BugBot: KI-gestütztes Automatisierungstest-Tool ist da:

Die Softwareentwicklung erlebt Innovationen durch KI. Cursor, ein KI-Editor, erreicht mit Version 1.0 einen Meilenstein.

5 June 2025

30+ öffentliche Web 3.0 APIs, die Sie jetzt nutzen können

30+ öffentliche Web 3.0 APIs, die Sie jetzt nutzen können

Der Aufstieg von Web 3.0: Dezentral, nutzerorientiert, transparent. APIs ermöglichen innovative dApps und Blockchain-Integration.

4 June 2025

Praktizieren Sie API Design-First in Apidog

Entdecken Sie eine einfachere Möglichkeit, APIs zu erstellen und zu nutzen

Kostenlos registrierenHerunterladen