Blog

Como Usar as APIs do BigQuery para Iniciantes

@apidog

@apidog

26 junho 2025

Como Usar as APIs do BigQuery para Iniciantes

O Google BigQuery revolucionou a forma como as organizações lidam com análises de dados em grande escala. Sua arquitetura sem servidor, escalabilidade e interface SQL familiar fazem dele uma ferramenta poderosa para derivar insights de conjuntos de dados massivos. Enquanto a interação com o BigQuery através do Google Cloud Console ou da ferramenta de linha de comando bq é comum, o verdadeiro poder da automação, integração e desenvolvimento de aplicações personalizadas é desbloqueado através de seu conjunto abrangente de Interfaces de Programação de Aplicações (APIs).

Este guia fornece uma exploração passo a passo de como usar as APIs do BigQuery, permitindo que você interaja programaticamente com seu data warehouse, construa pipelines de dados, integre o BigQuery em suas aplicações e gerencie recursos de forma eficaz. Abordaremos os diferentes tipos de APIs disponíveis, como configurar seu ambiente, exemplos práticos usando a biblioteca de cliente Python e apresentaremos APIs especializadas para casos de uso avançados.

💡
Você quer uma ótima ferramenta de teste de API que gera documentação de API bonita?

Quer uma plataforma integrada, All-in-One, para sua equipe de Desenvolvimento trabalhar em conjunto com máxima produtividade?

O Apidog atende todas as suas demandas e substitui o Postman a um preço muito mais acessível!
button

Entendendo as APIs do BigQuery

Antes de mergulhar no código, é crucial entender os conceitos fundamentais e as diferentes maneiras de interagir programaticamente com o BigQuery.

Conceitos Fundamentais do BigQuery:

Tipos de APIs do BigQuery:

O BigQuery oferece várias maneiras de interagir com seus serviços programaticamente:

API REST: Esta é a API fundamental construída sobre HTTP e JSON. Ela fornece acesso direto aos recursos e operações do BigQuery. Você pode interagir com ela usando requisições HTTP padrão (GET, POST, PUT, DELETE) direcionadas a endpoints específicos (por exemplo, https://bigquery.googleapis.com/bigquery/v2/projects/{projectId}/datasets). Embora seja poderosa e ofereça controle detalhado, usar a API REST diretamente requer lidar manualmente com autenticação, formatação de requisições, análise de respostas e tratamento de erros. A autenticação normalmente envolve tokens de acesso OAuth 2.0.

Bibliotecas de Cliente: O Google fornece bibliotecas de cliente de alto nível para várias linguagens de programação populares (incluindo Python, Java, Go, Node.js, C#, PHP, Ruby). Essas bibliotecas envolvem a API REST subjacente, oferecendo uma experiência mais idiomática e amigável ao desenvolvedor. Elas simplificam tarefas comuns, lidam com autenticação (geralmente automaticamente através de Credenciais Padrão de Aplicação), gerenciam tentativas e reduzem a quantidade de código padrão que você precisa escrever. Esta é a abordagem recomendada para a maioria do desenvolvimento de aplicações.

APIs Especializadas: Para tarefas específicas de alto desempenho ou especializadas, o BigQuery oferece APIs dedicadas:

Configurando Seu Ambiente

Antes de começar a fazer chamadas de API, você precisa configurar seu ambiente local ou do servidor.

Pré-requisitos:

  1. Conta do Google Cloud: Você precisa de uma conta ativa do Google Cloud.
  2. Projeto do Google Cloud: Crie um novo projeto ou selecione um existente no Google Cloud Console. Anote seu ID do Projeto.
  3. Ativar a API do BigQuery: Certifique-se de que a API do BigQuery está ativada para seu projeto. Você pode fazer isso via Cloud Console (APIs & Serviços > Biblioteca > Pesquisar por "API do BigQuery" > Ativar). Você também pode precisar ativar outras APIs, como a API de Leitura de Armazenamento do BigQuery ou a API de Conexão do BigQuery, dependendo de seu caso de uso.
  4. Faturamento: Certifique-se de que o faturamento está ativado para seu projeto. As operações do BigQuery geram custos com base no armazenamento de dados, análise processada e inserções em streaming.

Autenticação:

Sua aplicação precisa se autenticar no Google Cloud para provar sua identidade e autorização para acessar os recursos do BigQuery. O método recomendado para a maioria dos cenários são as Credenciais Padrão de Aplicação (ADC).

  1. Crie uma conta de serviço no Cloud Console (IAM & Admin > Contas de Serviço).
  2. Conceda os papéis necessários do BigQuery (por exemplo, BigQuery Data Editor, BigQuery Job User, BigQuery User) à conta de serviço.
  3. Baixe o arquivo de chave da conta de serviço (formato JSON).
  4. Defina a variável de ambiente GOOGLE_APPLICATION_CREDENTIALS para o caminho absoluto do arquivo de chave JSON baixado. As bibliotecas de cliente usarão automaticamente esse arquivo de chave para autenticação se a variável de ambiente estiver definida.

Instalando Bibliotecas de Cliente (Exemplo em Python):

Focaremos no Python para nossos exemplos. Você pode instalar as bibliotecas necessárias usando pip:

pip install google-cloud-bigquery
# Opcional: Instale a biblioteca da API de armazenamento para leituras mais rápidas
pip install google-cloud-bigquery-storage
# Opcional: Instale a integração do pandas e db-dtypes para melhor manipulação de tipos
pip install pandas db-dtypes pyarrow

Certifique-se de ter o Python instalado (versão 3.7+ recomendado para os recursos mais recentes da biblioteca).

Usando a Biblioteca do Cliente do BigQuery (Exemplos em Python)

Agora, vamos explorar operações comuns do BigQuery usando a biblioteca Python google-cloud-bigquery.

1. Importando e Inicializando o Cliente:

Primeiro, importe a biblioteca. Em seguida, crie uma instância do cliente. Se o ADC estiver configurado corretamente, o cliente será autenticado automaticamente.

from google.cloud import bigquery
import pandas as pd

# Construa um objeto cliente do BigQuery.
# Se GOOGLE_APPLICATION_CREDENTIALS estiver definido, usará a conta de serviço.
# Se gcloud auth application-default login foi executado, usará essas credenciais.
# Se estiver executando na infraestrutura do GCP, usará a conta de serviço da instância.
client = bigquery.Client()

# Você pode especificar explicitamente o ID do projeto, se necessário,
# caso contrário, muitas vezes ele infere do ambiente/credenciais do ADC.
# client = bigquery.Client(project='seu-id-do-projeto')

print("Cliente criado com sucesso.")

2. Executando Consultas:

A operação mais comum é executar consultas SQL.

# Defina sua consulta SQL
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
"""

# Faça uma solicitação à API e aguarde a conclusão do trabalho.
query_job = client.query(query)  # Solicitação à API
print(f"Trabalho iniciado: {query_job.job_id}")

# Aguarde a conclusão do trabalho e obtenha os resultados.
rows = query_job.result()  # Aguarda a conclusão do trabalho.

print("\nTop 10 nomes no TX (1910-2013):")
for row in rows:
    # Os valores das linhas podem ser acessados pelo nome do campo ou índice.
    print(f"Nome: {row.name}, Contagem: {row['total_people']}") # Acesso por atributo ou chave

# Converta os resultados em um DataFrame do Pandas
df = rows.to_dataframe()
print("\nResultados como DataFrame do Pandas:")
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 a sintaxe SQL padrão para consultas.
    use_legacy_sql=False
)

# Inicie a consulta, passando a configuração extra.
query_job = client.query(query, job_config=job_config) # Não aguarda
print(f"Trabalho assíncrono iniciado: {query_job.job_id}")

# --- Mais tarde em sua aplicação ---
# Verifique o status do trabalho (opcional)
# from google.cloud.exceptions import NotFound
# try:
#     job = client.get_job(query_job.job_id, location=query_job.location)
#     print(f"Status do trabalho {job.job_id}: {job.state}")
#     if job.state == "DONE":
#         if job.error_result:
#             print(f"Trabalho falhou: {job.errors}")
#         else:
#             results = job.result() # Obter resultados
#             print("Resultados obtidos.")
#             # Processar resultados...
# except NotFound:
#     print(f"Trabalho {query_job.job_id} não encontrado.")

# Ou simplesmente aguarde a conclusão quando necessário
results = query_job.result() # Isso bloqueará até que o trabalho seja concluído
print("Trabalho assíncrono concluído.")
for row in results:
    print(f"Corpus: {row.corpus}, Palavras Distintas: {row.distinct_words}")


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

# Exemplo: Encontrar nomes que começam com um prefixo específico em um dado estado
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%' para o operador LIKE
        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"Trabalho de consulta parameterizada iniciado: {query_job.job_id}")

rows = query_job.result()

print(f"\nNomes começando com '{prefix_param}' em {state_param} com >= {min_count_param} pessoas:")
for row in rows:
    print(f"Nome: {row.name}, Contagem: {row.total_people}")

3. Gerenciando Conjuntos de Dados:

Você pode criar, listar, obter detalhes e excluir conjuntos de dados.

# Defina o ID do conjunto de dados e a localização
project_id = client.project
dataset_id = f"{project_id}.meu_novo_conjunto_de_dados"
dataset_location = "US" # ex: "US", "EU", "asia-northeast1"

# Construa um objeto Dataset completo para enviar à API.
dataset = bigquery.Dataset(dataset_id)
dataset.location = dataset_location
dataset.description = "Conjunto de dados criado via biblioteca de cliente Python"

try:
    # Faça uma solicitação à API para criar o conjunto de dados.
    dataset = client.create_dataset(dataset, timeout=30)  # Faça uma solicitação à API.
    print(f"Conjunto de dados {client.project}.{dataset.dataset_id} criado.")

    # Liste os conjuntos de dados no projeto
    print("\nConjuntos de dados no projeto:")
    datasets = list(client.list_datasets()) # Solicitação à API
    if datasets:
        for ds in datasets:
            print(f"\t{ds.dataset_id}")
    else:
        print(f"\tO projeto {client.project} não contém nenhum conjunto de dados.")

    # Obtenha informações do conjunto de dados
    retrieved_dataset = client.get_dataset(dataset_id) # Solicitação à API
    print(f"\nInformações do conjunto de dados recuperadas para {dataset_id}:")
    print(f"\tDescrição: {retrieved_dataset.description}")
    print(f"\tLocalização: {retrieved_dataset.location}")

except Exception as e:
    print(f"Erro durante as operações no conjunto de dados: {e}")

finally:
    # Limpeza: Exclua o conjunto de dados
    try:
        client.delete_dataset(
            dataset_id, delete_contents=True, not_found_ok=True
        )  # Solicitação à API
        print(f"\nConjunto de dados '{dataset_id}' excluído com sucesso.")
    except Exception as e:
         print(f"Erro ao excluir o conjunto de dados {dataset_id}: {e}")

4. Gerenciando Tabelas:

Operações semelhantes existem para tabelas: criação de tabelas (definindo o esquema), carregamento de dados, obtendo metadados e excluindo tabelas.

# Usando o ID do conjunto de dados criado anteriormente (certifique-se de que existe ou remova a etapa de exclusão acima)
dataset_id_for_table = "meu_novo_conjunto_de_dados" # Use um ID de conjunto de dados válido
table_id = f"{client.project}.{dataset_id_for_table}.minha_nova_tabela"

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

# Crie a tabela
table = bigquery.Table(table_id, schema=schema)
try:
    # Certifique-se de que o conjunto de dados existe primeiro
    client.create_dataset(dataset_id_for_table, exists_ok=True)

    table = client.create_table(table)  # Solicitação à API
    print(
        f"Tabela {table.project}.{table.dataset_id}.{table.table_id} criada."
    )

    # --- Carregando Dados (Exemplo: de um DataFrame do Pandas) ---
    data = {'full_name': ['Alice Smith', 'Bob Johnson'],
            'age': [30, 45],
            'email': ['alice@example.com', None]}
    dataframe = pd.DataFrame(data)

    job_config = bigquery.LoadJobConfig(
        # Especificar o esquema é recomendado, garante tipos adequados
        schema=schema,
        # Opcional: sobrescrever os dados da tabela
        write_disposition="WRITE_TRUNCATE",
        # Ou anexar: write_disposition="WRITE_APPEND",
    )

    load_job = client.load_table_from_dataframe(
        dataframe, table_id, job_config=job_config
    )  # Solicitação à API
    print(f"Iniciando o trabalho {load_job.job_id} para carregar dados do DataFrame")

    load_job.result()  # Aguarda a conclusão do trabalho.
    print("Carga do DataFrame concluída.")

    destination_table = client.get_table(table_id) # Solicitação à API
    print(f"Carregados {destination_table.num_rows} linhas na tabela {table_id}")

    # --- Carregando Dados (Exemplo: de um URI do Google Cloud Storage) ---
    # Suponha que um arquivo CSV gs://your-bucket/data.csv exista com dados compatíveis
    # uri = "gs://your-bucket/data.csv"
    # job_config_gcs = bigquery.LoadJobConfig(
    #     schema=schema,
    #     skip_leading_rows=1, # Ignorar a linha de cabeçalho
    #     source_format=bigquery.SourceFormat.CSV,
    #     write_disposition="WRITE_APPEND", # Anexar aos dados existentes
    # )
    # load_job_gcs = client.load_table_from_uri(
    #     uri, table_id, job_config=job_config_gcs
    # )
    # print(f"Iniciando o trabalho {load_job_gcs.job_id} para carregar dados do GCS")
    # load_job_gcs.result()
    # print("Carga do GCS concluída.")
    # destination_table = client.get_table(table_id)
    # print(f"Total de linhas após carga do GCS: {destination_table.num_rows}")

except Exception as e:
    print(f"Erro durante as operações na tabela: {e}")

finally:
    # Limpeza: Excluir a tabela
    try:
        client.delete_table(table_id, not_found_ok=True)  # Solicitação à API
        print(f"Tabela '{table_id}' excluída com sucesso.")
        # Opcionalmente, exclua o conjunto de dados novamente se foi apenas para este exemplo
        # client.delete_dataset(dataset_id_for_table, delete_contents=True, not_found_ok=True)
        # print(f"Conjunto de dados '{dataset_id_for_table}' excluído com sucesso.")
    except Exception as e:
        print(f"Erro ao excluir a tabela {table_id}: {e}")

5. Trabalhando com Trabalhos:

Todas as operações assíncronas (consulta, carga, exportação, cópia) criam recursos de Trabalho. Você pode listar e gerenciar esses trabalhos.

# Liste os trabalhos recentes
print("\nTrabalhos recentes do BigQuery:")
for job in client.list_jobs(max_results=10): # Solicitação à API
    print(f"ID do Trabalho: {job.job_id}, Tipo: {job.job_type}, Estado: {job.state}, Criado: {job.created}")

# Obtenha um trabalho específico (substitua por um ID de trabalho válido de execuções anteriores)
# try:
#     job_id_to_get = "..." # Substitua por um ID de trabalho real
#     location = "US"      # Substitua pela localização do trabalho se não for padrão
#     retrieved_job = client.get_job(job_id_to_get, location=location) # Solicitação à API
#     print(f"\nDetalhes do trabalho {retrieved_job.job_id}:")
#     print(f"\tEstado: {retrieved_job.state}")
#     if retrieved_job.error_result:
#         print(f"\tErro: {retrieved_job.errors}")
# except NotFound:
#     print(f"Trabalho {job_id_to_get} não encontrado.")
# except Exception as e:
#     print(f"Erro ao recuperar o trabalho: {e}")

Aproveitando APIs Especializadas (Conceitos e Casos de Uso)

Enquanto a biblioteca cliente central cobre muitos casos de uso, APIs especializadas oferecem desempenho ou funcionalidade aprimorada para tarefas específicas.

1. API de Leitura de Armazenamento do BigQuery (Python):

# Necessário: 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

# --- Usando Pandas read_gbq (Integração mais simples) ---
# Isso utiliza automaticamente a API de Armazenamento se instalada e benéfica
# 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, # Solicitar explicitamente a API de Armazenamento
#          progress_bar_type='tqdm' # Barra de progresso opcional
#      )
#      print("\nLido dados usando a API de Armazenamento via pandas.read_gbq:")
#      print(df_storage.head())
#      print(f"Lidos {len(df_storage)} linhas.")
# except Exception as e:
#      print(f"Erro ao ler com a API de Armazenamento via read_gbq: {e}")


# --- Uso Manual da API de Armazenamento (Mais Controle) ---
# bqstorageclient = bigquery_storage_v1.BigQueryReadClient()
# table = f"projects/{project_id}/datasets/{dataset_id}/tables/{table_name}" # Substitua pelos detalhes de sua tabela

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

# read_session = bqstorageclient.create_read_session(
#     parent=parent,
#     read_session=requested_session,
#     max_stream_count=1, # Solicitar número de streams paralelos
# )

# 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("\nLido dados manualmente usando a API de Armazenamento:")
#     print(df_manual.head())
# else:
#     print("Nenhum dado lido usando a API de Armazenamento manual.")

2. API de Conexão do BigQuery:

  1. Use a API (ou o Console da Nuvem/bq ferramenta) para criar um recurso de Connection, especificando o tipo e detalhes da fonte externa.
  2. Conceda à conta de serviço da conexão as permissões apropriadas no recurso externo (por exemplo, papel de usuário do Cloud SQL).
  3. Use a função EXTERNAL_QUERY("connection_id", "external_sql_query") dentro de seu SQL do BigQuery.

3. API do Hub de Análise:

4. API do BigLake:

Usando a API REST Diretamente

Embora as bibliotecas de cliente sejam geralmente preferidas, você pode usar a API REST diretamente se:

Fazendo Requisições:

Você normalmente usará um cliente HTTP (como curl ou a biblioteca requests do Python). Você precisa:

  1. Obter um Token de Acesso OAuth 2.0 (por exemplo, usando gcloud auth print-access-token).
  2. Construir a URL do endpoint da API correta.
  3. Criar o corpo da requisição JSON de acordo com a especificação do método da API.
  4. Incluir o token de acesso no cabeçalho Authorization: Bearer <token>.
  5. Lidar com a resposta HTTP (códigos de status, análise JSON, mensagens de erro).

Exemplo: Executando uma Consulta via REST (jobs.query)

# 1. Obter Token de Acesso
TOKEN=$(gcloud auth print-access-token)

# 2. Definir ID do Projeto e Corpo da Requisição
PROJECT_ID="seu-id-do-projeto" # Substitua pelo seu ID do projeto
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. Fazer a Chamada à API usando 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}"

# A resposta conterá informações do trabalho, incluindo o ID do trabalho.
# Você precisará então fazer chamadas subsequentes para jobs.getQueryResults
# usando o ID do trabalho para recuperar as linhas de dados reais uma vez que o trabalho esteja concluído.

Este exemplo apenas inicia o trabalho de consulta. Recuperar resultados requer polling do status do trabalho e então chamar o endpoint jobs.getQueryResults. Isso destaca os passos extras envolvidos em comparação com as bibliotecas de cliente.

Melhores Práticas e Dicas

Conclusão

As APIs do BigQuery fornecem uma maneira poderosa e flexível de interagir programaticamente com seu data warehouse. Seja para automatizar o carregamento de dados, executar consultas analíticas complexas como parte de uma aplicação, integrar insights do BigQuery em painéis, ou gerenciar recursos dinamicamente, as APIs oferecem as ferramentas necessárias.

Ao entender os diferentes tipos de APIs, configurar corretamente seu ambiente e aproveitar a conveniência das bibliotecas de cliente (especialmente em Python), você pode desbloquear todo o potencial do BigQuery além do console interativo. Comece com a API central para tarefas comuns, explore APIs especializadas como a API de Leitura de Armazenamento para operações críticas de desempenho, e lembre-se de que a API REST está sempre disponível para controle absoluto. À medida que você constrói aplicações no BigQuery, essas APIs serão componentes essenciais da sua arquitetura de dados.

Pratique o design de API no Apidog

Descubra uma forma mais fácil de construir e usar APIs

Cadastre-se gratuitamenteBaixar