使用Python與BigQuery進行交互的代碼詳解

更新時間：2025年04月08日 08:51:02 作者：程序員小jobleap

在大數(shù)據(jù)分析的領域中,Google BigQuery 是一個被廣泛使用的云端數(shù)據(jù)倉庫解決方案,它由 Google Cloud 提供,并且專為處理大規(guī)模數(shù)據(jù)集、進行快速的數(shù)據(jù)分析和復雜的查詢而設計,本文給大家講解了如何使用Python與BigQuery進行交互,需要的朋友可以參考下

選擇合適的 Python 庫

在使用 BigQuery 時，您可以根據(jù)自己的需求選擇以下三個 Python 庫：

BigQuery DataFrame：通過服務器端處理，支持 Pandas 和 Scikit-learn API，適合數(shù)據(jù)處理和機器學習任務。
pandas-gbq：客戶端庫，用于在 Python 中讀寫 BigQuery 數(shù)據(jù)，適合簡單的數(shù)據(jù)處理和分析。
google-cloud-bigquery：Google 維護的庫，提供完整的 BigQuery API 功能，適合復雜的數(shù)據(jù)管理和分析。

安裝庫

要使用這些庫，您需要安裝以下包：

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

運行查詢

使用 GoogleSQL 語法

以下示例展示了如何使用 pandas-gbq 和 google-cloud-bigquery 運行 GoogleSQL 查詢：

pandas-gbq

import pandas

sql = """
    SELECT name
    FROM `bigquery-public-data.usa_names.usa_1910_current`
    WHERE state = 'TX'
    LIMIT 100
"""

# 使用標準 SQL 查詢
df = pandas.read_gbq(sql, dialect="standard")

# 指定項目 ID
project_id = "your-project-id"
df = pandas.read_gbq(sql, project_id=project_id, dialect="standard")

google-cloud-bigquery

from google.cloud import bigquery

client = bigquery.Client()
sql = """
    SELECT name
    FROM `bigquery-public-data.usa_names.usa_1910_current`
    WHERE state = 'TX'
    LIMIT 100
"""

# 使用標準 SQL 查詢
df = client.query(sql).to_dataframe()

# 指定項目 ID
project_id = "your-project-id"
df = client.query(sql, project=project_id).to_dataframe()

使用舊版 SQL 語法

如果需要使用舊版 SQL 語法，可以通過以下方式進行：

pandas-gbq

import pandas

sql = """
    SELECT name
    FROM [bigquery-public-data:usa_names.usa_1910_current]
    WHERE state = 'TX'
    LIMIT 100
"""

df = pandas.read_gbq(sql, dialect="legacy")

google-cloud-bigquery

from google.cloud import bigquery

client = bigquery.Client()
sql = """
    SELECT name
    FROM [bigquery-public-data:usa_names.usa_1910_current]
    WHERE state = 'TX'
    LIMIT 100
"""
query_config = bigquery.QueryJobConfig(use_legacy_sql=True)

df = client.query(sql, job_config=query_config).to_dataframe()

使用 BigQuery Storage API 加速數(shù)據(jù)下載

BigQuery Storage API 可以顯著提高大型結果的下載速度。以下示例展示了如何使用此 API：

pandas-gbq

import pandas

sql = "SELECT * FROM `bigquery-public-data.irs_990.irs_990_2012`"

# 使用 BigQuery Storage API 加速下載
df = pandas.read_gbq(sql, dialect="standard", use_bqstorage_api=True)

google-cloud-bigquery

from google.cloud import bigquery

client = bigquery.Client()
sql = "SELECT * FROM `bigquery-public-data.irs_990.irs_990_2012`"

# 如果 BigQuery Storage API 已啟用，則自動使用
df = client.query(sql).to_dataframe()

配置查詢

參數(shù)化查詢

以下示例展示了如何使用參數(shù)化查詢：

pandas-gbq

import pandas

sql = """
    SELECT name
    FROM `bigquery-public-data.usa_names.usa_1910_current`
    WHERE state = @state
    LIMIT @limit
"""
query_config = {
    "query": {
        "parameterMode": "NAMED",
        "queryParameters": [
            {
                "name": "state",
                "parameterType": {"type": "STRING"},
                "parameterValue": {"value": "TX"},
            },
            {
                "name": "limit",
                "parameterType": {"type": "INTEGER"},
                "parameterValue": {"value": 100},
            },
        ],
    }
}

df = pandas.read_gbq(sql, configuration=query_config)

google-cloud-bigquery

from google.cloud import bigquery

client = bigquery.Client()
sql = """
    SELECT name
    FROM `bigquery-public-data.usa_names.usa_1910_current`
    WHERE state = @state
    LIMIT @limit
"""
query_config = bigquery.QueryJobConfig(
    query_parameters=[
        bigquery.ScalarQueryParameter("state", "STRING", "TX"),
        bigquery.ScalarQueryParameter("limit", "INTEGER", 100),
    ]
)

df = client.query(sql, job_config=query_config).to_dataframe()

將 pandas DataFrame 加載到 BigQuery 表中

以下示例展示了如何將 pandas DataFrame 加載到 BigQuery 表中：

pandas-gbq

import pandas

df = pandas.DataFrame(
    {
        "my_string": ["a", "b", "c"],
        "my_int64": [1, 2, 3],
        "my_float64": [4.0, 5.0, 6.0],
        "my_timestamp": [
            pandas.Timestamp("1998-09-04T16:03:14"),
            pandas.Timestamp("2010-09-13T12:03:45"),
            pandas.Timestamp("2015-10-02T16:00:00"),
        ],
    }
)
table_id = "my_dataset.new_table"

df.to_gbq(table_id)

google-cloud-bigquery

from google.cloud import bigquery
import pandas

df = pandas.DataFrame(
    {
        "my_string": ["a", "b", "c"],
        "my_int64": [1, 2, 3],
        "my_float64": [4.0, 5.0, 6.0],
        "my_timestamp": [
            pandas.Timestamp("1998-09-04T16:03:14"),
            pandas.Timestamp("2010-09-13T12:03:45"),
            pandas.Timestamp("2015-10-02T16:00:00"),
        ],
    }
)
client = bigquery.Client()
table_id = "my_dataset.new_table"

# 確保正確的數(shù)據(jù)類型
job_config = bigquery.LoadJobConfig(
    schema=[
        bigquery.SchemaField("my_string", "STRING"),
    ]
)

job = client.load_table_from_dataframe(df, table_id, job_config=job_config)

# 等待加載完成
job.result()

pandas-gbq 的局限性

數(shù)據(jù)集管理：不支持創(chuàng)建、更新或刪除數(shù)據(jù)集。
數(shù)據(jù)格式支持：僅支持 CSV 格式，不支持嵌套值或數(shù)組值。
表管理：不支持列出表、復制表或刪除表。
數(shù)據(jù)導出：不支持直接導出數(shù)據(jù)到 Cloud Storage。

解決連接池錯誤

如果遇到連接池錯誤，可以通過以下方式增加連接池大小：

import requests

client = bigquery.Client()
adapter = requests.adapters.HTTPAdapter(pool_connections=128, pool_maxsize=128, max_retries=3)
client._http.mount("https://", adapter)
client._http._auth_request.session.mount("https://", adapter)

以上就是使用Python與BigQuery進行交互的代碼詳解的詳細內容，更多關于Python與BigQuery交互的資料請關注腳本之家其它相關文章！

您可能感興趣的文章: