Blog

Cara Menggunakan BigQuery API untuk Pemula

Panduan ini langkah demi langkah cara menggunakan BigQuery API. Kelola sumber daya, bangun *pipeline* data, dan integrasikan ke aplikasi.

Ardianto Nugroho

Ardianto Nugroho

14 April 2025

Cara Menggunakan BigQuery API untuk Pemula

Google BigQuery telah merevolusi cara organisasi menangani analisis data skala besar. Arsitektur tanpa server, skalabilitas, dan antarmuka SQL yang familiar menjadikannya alat yang ampuh untuk memperoleh wawasan dari dataset masif. Meskipun berinteraksi dengan BigQuery melalui Google Cloud Console atau alat baris perintah bq adalah hal yang umum, kekuatan sebenarnya dari otomatisasi, integrasi, dan pengembangan aplikasi kustom dibuka melalui serangkaian Application Programming Interfaces (API) yang komprehensif.

Panduan ini memberikan eksplorasi langkah demi langkah tentang cara menggunakan BigQuery API, memungkinkan Anda untuk berinteraksi secara terprogram dengan gudang data Anda, membangun alur data, mengintegrasikan BigQuery ke dalam aplikasi Anda, dan mengelola sumber daya secara efektif. Kami akan membahas berbagai jenis API yang tersedia, cara menyiapkan lingkungan Anda, contoh praktis menggunakan pustaka klien Python, dan memperkenalkan API khusus untuk kasus penggunaan tingkat lanjut.

💡
Ingin alat Pengujian API hebat yang menghasilkan Dokumentasi API yang indah?

Ingin platform terintegrasi, All-in-One untuk Tim Pengembang Anda untuk bekerja bersama dengan produktivitas maksimum?

Apidog memberikan semua demans Anda, dan menggantikan Postman dengan harga yang jauh lebih terjangkau!
button

Memahami BigQuery API

Sebelum masuk ke kode, penting untuk memahami konsep inti dan berbagai cara Anda dapat berinteraksi dengan BigQuery secara terprogram.

Konsep Inti BigQuery:

Jenis BigQuery API:

BigQuery menawarkan beberapa cara untuk berinteraksi dengan layanannya secara terprogram:

REST API: Ini adalah API dasar yang dibangun di atas HTTP dan JSON. Ini menyediakan akses langsung ke sumber daya dan operasi BigQuery. Anda dapat berinteraksi dengannya menggunakan permintaan HTTP standar (GET, POST, PUT, DELETE) yang menargetkan endpoint tertentu (misalnya, https://bigquery.googleapis.com/bigquery/v2/projects/{projectId}/datasets). Meskipun kuat dan menawarkan kontrol yang terperinci, menggunakan REST API secara langsung mengharuskan penanganan autentikasi, pemformatan permintaan, penguraian respons, dan penanganan kesalahan secara manual. Autentikasi biasanya melibatkan token akses OAuth 2.0.

Pustaka Klien: Google menyediakan pustaka klien tingkat tinggi untuk berbagai bahasa pemrograman populer (termasuk Python, Java, Go, Node.js, C#, PHP, Ruby). Pustaka ini membungkus REST API yang mendasarinya, menawarkan pengalaman yang lebih idiomatis dan ramah pengembang. Mereka menyederhanakan tugas-tugas umum, menangani autentikasi (sering kali secara otomatis melalui Application Default Credentials), mengelola percobaan ulang, dan mengurangi jumlah kode boilerplate yang perlu Anda tulis. Ini adalah pendekatan yang direkomendasikan untuk sebagian besar pengembangan aplikasi.

API Khusus: Untuk tugas-tugas khusus atau berkinerja tinggi tertentu, BigQuery menawarkan API khusus:

Menyiapkan Lingkungan Anda

Sebelum Anda dapat mulai membuat panggilan API, Anda perlu mengonfigurasi lingkungan lokal atau server Anda.

Prasyarat:

  1. Akun Google Cloud: Anda memerlukan akun Google Cloud yang aktif.
  2. Project Google Cloud: Buat project baru atau pilih yang sudah ada di Google Cloud Console. Catat ID Project Anda.
  3. Aktifkan BigQuery API: Pastikan BigQuery API diaktifkan untuk project Anda. Anda dapat melakukan ini melalui Cloud Console (API & Layanan > Pustaka > Cari "BigQuery API" > Aktifkan). Anda mungkin juga perlu mengaktifkan API lain seperti BigQuery Storage Read API atau BigQuery Connection API tergantung pada kasus penggunaan Anda.
  4. Penagihan: Pastikan penagihan diaktifkan untuk project Anda. Operasi BigQuery menimbulkan biaya berdasarkan penyimpanan data, analisis yang diproses, dan penyisipan streaming.

Autentikasi:

Aplikasi Anda perlu melakukan autentikasi ke Google Cloud untuk membuktikan identitas dan otorisasi untuk mengakses sumber daya BigQuery. Metode yang direkomendasikan untuk sebagian besar skenario adalah Application Default Credentials (ADC).

  1. Buat akun layanan di Cloud Console (IAM & Admin > Akun Layanan).
  2. Berikan peran BigQuery yang diperlukan (misalnya, BigQuery Data Editor, BigQuery Job User, BigQuery User) ke akun layanan.
  3. Unduh file kunci akun layanan (format JSON).
  4. Atur variabel lingkungan GOOGLE_APPLICATION_CREDENTIALS ke jalur absolut file kunci JSON yang diunduh. Pustaka klien akan secara otomatis menggunakan file kunci ini untuk autentikasi jika variabel lingkungan diatur.

Menginstal Pustaka Klien (Contoh Python):

Kami akan fokus pada Python untuk contoh kami. Anda dapat menginstal pustaka yang diperlukan menggunakan pip:

pip install google-cloud-bigquery
# Opsional: Instal pustaka Storage API untuk pembacaan yang lebih cepat
pip install google-cloud-bigquery-storage
# Opsional: Instal integrasi pandas dan db-dtypes untuk penanganan tipe yang lebih baik
pip install pandas db-dtypes pyarrow

Pastikan Anda telah menginstal Python (versi 3.7+ direkomendasikan untuk fitur pustaka terbaru).

Menggunakan Pustaka Klien BigQuery (Contoh Python)

Sekarang, mari kita jelajahi operasi BigQuery umum menggunakan pustaka Python google-cloud-bigquery.

1. Mengimpor dan Menginisialisasi Klien:

Pertama, impor pustaka. Kemudian, buat instance klien. Jika ADC dikonfigurasi dengan benar, klien akan melakukan autentikasi secara otomatis.

from google.cloud import bigquery
import pandas as pd

# Buat objek klien BigQuery.
# Jika GOOGLE_APPLICATION_CREDENTIALS diatur, ia menggunakan akun layanan.
# Jika gcloud auth application-default login dijalankan, ia menggunakan kredensial tersebut.
# Jika berjalan di infrastruktur GCP, ia menggunakan akun layanan instance.
client = bigquery.Client()

# Anda dapat secara eksplisit menentukan ID project jika diperlukan,
# jika tidak, sering kali menyimpulkan dari lingkungan/kredensial ADC.
# client = bigquery.Client(project='your-project-id')

print("Klien berhasil dibuat.")

2. Menjalankan Kueri:

Operasi yang paling umum adalah menjalankan kueri SQL.

# Tentukan kueri SQL Anda
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
"""

# Buat permintaan API dan tunggu hingga pekerjaan selesai.
query_job = client.query(query)  # Permintaan API
print(f"Pekerjaan dimulai: {query_job.job_id}")

# Tunggu hingga pekerjaan selesai dan dapatkan hasilnya.
rows = query_job.result()  # Menunggu pekerjaan selesai.

print("\n10 nama teratas di TX (1910-2013):")
for row in rows:
    # Nilai baris dapat diakses berdasarkan nama bidang atau indeks.
    print(f"Nama: {row.name}, Jumlah: {row['total_people']}") # Akses berdasarkan atribut atau kunci

# Konversi hasil ke Pandas DataFrame
df = rows.to_dataframe()
print("\nHasil sebagai 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(
    # Gunakan sintaks SQL standar untuk kueri.
    use_legacy_sql=False
)

# Mulai kueri, dengan meneruskan konfigurasi tambahan.
query_job = client.query(query, job_config=job_config) # Tidak menunggu
print(f"Pekerjaan asinkron dimulai: {query_job.job_id}")

# --- Nanti dalam aplikasi Anda ---
# Periksa status pekerjaan (opsional)
# from google.cloud.exceptions import NotFound
# try:
#     job = client.get_job(query_job.job_id, location=query_job.location)
#     print(f"Status pekerjaan {job.job_id}: {job.state}")
#     if job.state == "DONE":
#         if job.error_result:
#             print(f"Pekerjaan gagal: {job.errors}")
#         else:
#             results = job.result() # Dapatkan hasil
#             print("Hasil diambil.")
#             # Proses hasil...
# except NotFound:
#     print(f"Pekerjaan {query_job.job_id} tidak ditemukan.")

# Atau cukup tunggu hingga selesai saat dibutuhkan
results = query_job.result() # Ini akan memblokir hingga pekerjaan selesai
print("Pekerjaan asinkron selesai.")
for row in results:
    print(f"Corpus: {row.corpus}, Kata yang Berbeda: {row.distinct_words}")


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

# Contoh: Temukan nama yang dimulai dengan awalan tertentu di negara bagian tertentu
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),
        # Gunakan 'val%' untuk operator 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"Pekerjaan kueri parameter dimulai: {query_job.job_id}")

rows = query_job.result()

print(f"\nNama yang dimulai dengan '{prefix_param}' di {state_param} dengan >= {min_count_param} orang:")
for row in rows:
    print(f"Nama: {row.name}, Jumlah: {row.total_people}")

3. Mengelola Dataset:

Anda dapat membuat, membuat daftar, mendapatkan detail, dan menghapus dataset.

# Tentukan ID dataset dan lokasi
project_id = client.project
dataset_id = f"{project_id}.my_new_dataset"
dataset_location = "US" # misalnya, "US", "EU", "asia-northeast1"

# Buat objek Dataset lengkap untuk dikirim ke API.
dataset = bigquery.Dataset(dataset_id)
dataset.location = dataset_location
dataset.description = "Dataset dibuat melalui pustaka klien Python"

try:
    # Buat permintaan API untuk membuat dataset.
    dataset = client.create_dataset(dataset, timeout=30)  # Buat permintaan API.
    print(f"Dataset {client.project}.{dataset.dataset_id} dibuat")

    # Daftar dataset dalam project
    print("\nDataset dalam project:")
    datasets = list(client.list_datasets()) # Permintaan API
    if datasets:
        for ds in datasets:
            print(f"\t{ds.dataset_id}")
    else:
        print(f"\tProject {client.project} tidak berisi dataset apa pun.")

    # Dapatkan info dataset
    retrieved_dataset = client.get_dataset(dataset_id) # Permintaan API
    print(f"\nInfo dataset yang diambil untuk {dataset_id}:")
    print(f"\tDeskripsi: {retrieved_dataset.description}")
    print(f"\tLokasi: {retrieved_dataset.location}")

except Exception as e:
    print(f"Kesalahan selama operasi dataset: {e}")

finally:
    # Bersihkan: Hapus dataset
    try:
        client.delete_dataset(
            dataset_id, delete_contents=True, not_found_ok=True
        )  # Permintaan API
        print(f"\nBerhasil menghapus dataset '{dataset_id}'.")
    except Exception as e:
         print(f"Kesalahan menghapus dataset {dataset_id}: {e}")

4. Mengelola Tabel:

Operasi serupa ada untuk tabel: membuat tabel (mendefinisikan skema), memuat data, mendapatkan metadata, dan menghapus tabel.

# Menggunakan ID dataset yang dibuat sebelumnya (pastikan itu ada atau hapus langkah penghapusan di atas)
dataset_id_for_table = "my_new_dataset" # Gunakan ID dataset yang valid
table_id = f"{client.project}.{dataset_id_for_table}.my_new_table"

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

# Buat tabel
table = bigquery.Table(table_id, schema=schema)
try:
    # Pastikan dataset ada terlebih dahulu
    client.create_dataset(dataset_id_for_table, exists_ok=True)

    table = client.create_table(table)  # Permintaan API
    print(
        f"Tabel {table.project}.{table.dataset_id}.{table.table_id} dibuat"
    )

    # --- Memuat Data (Contoh: dari 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(
        # Menentukan skema direkomendasikan, memastikan tipe yang tepat
        schema=schema,
        # Opsional: timpa data tabel
        write_disposition="WRITE_TRUNCATE",
        # Atau tambahkan: write_disposition="WRITE_APPEND",
    )

    load_job = client.load_table_from_dataframe(
        dataframe, table_id, job_config=job_config
    )  # Permintaan API
    print(f"Memulai pekerjaan {load_job.job_id} untuk memuat data dari DataFrame")

    load_job.result()  # Menunggu pekerjaan selesai.
    print("Pekerjaan pemuatan DataFrame selesai.")

    destination_table = client.get_table(table_id) # Permintaan API
    print(f"{destination_table.num_rows} baris dimuat ke dalam tabel {table_id}")

    # --- Memuat Data (Contoh: dari Google Cloud Storage URI) ---
    # Asumsikan file CSV gs://your-bucket/data.csv ada dengan data yang kompatibel
    # uri = "gs://your-bucket/data.csv"
    # job_config_gcs = bigquery.LoadJobConfig(
    #     schema=schema,
    #     skip_leading_rows=1, # Lewati baris header
    #     source_format=bigquery.SourceFormat.CSV,
    #     write_disposition="WRITE_APPEND", # Tambahkan ke data yang ada
    # )
    # load_job_gcs = client.load_table_from_uri(
    #     uri, table_id, job_config=job_config_gcs
    # )
    # print(f"Memulai pekerjaan {load_job_gcs.job_id} untuk memuat data dari GCS")
    # load_job_gcs.result()
    # print("Pekerjaan pemuatan GCS selesai.")
    # destination_table = client.get_table(table_id)
    # print(f"Total baris setelah pemuatan GCS: {destination_table.num_rows}")

except Exception as e:
    print(f"Kesalahan selama operasi tabel: {e}")

finally:
    # Bersihkan: Hapus tabel
    try:
        client.delete_table(table_id, not_found_ok=True)  # Permintaan API
        print(f"Berhasil menghapus tabel '{table_id}'.")
        # Opsional, hapus dataset lagi jika hanya untuk contoh ini
        # client.delete_dataset(dataset_id_for_table, delete_contents=True, not_found_ok=True)
        # print(f"Berhasil menghapus dataset '{dataset_id_for_table}'.")
    except Exception as e:
        print(f"Kesalahan menghapus tabel {table_id}: {e}")

5. Bekerja dengan Pekerjaan:

Semua operasi asinkron (kueri, muat, ekspor, salin) membuat sumber daya Pekerjaan. Anda dapat membuat daftar dan mengelola pekerjaan ini.

# Daftar pekerjaan terbaru
print("\nPekerjaan BigQuery Terbaru:")
for job in client.list_jobs(max_results=10): # Permintaan API
    print(f"ID Pekerjaan: {job.job_id}, Tipe: {job.job_type}, Status: {job.state}, Dibuat: {job.created}")

# Dapatkan pekerjaan tertentu (ganti dengan ID pekerjaan yang valid dari eksekusi sebelumnya)
# try:
#     job_id_to_get = "..." # Ganti dengan ID pekerjaan yang sebenarnya
#     location = "US"      # Ganti dengan lokasi pekerjaan jika bukan default
#     retrieved_job = client.get_job(job_id_to_get, location=location) # Permintaan API
#     print(f"\nDetail untuk pekerjaan {retrieved_job.job_id}:")
#     print(f"\tStatus: {retrieved_job.state}")
#     if retrieved_job.error_result:
#         print(f"\tKesalahan: {retrieved_job.errors}")
# except NotFound:
#     print(f"Pekerjaan {job_id_to_get} tidak ditemukan.")
# except Exception as e:
#     print(f"Kesalahan mengambil pekerjaan: {e}")

Memanfaatkan API Khusus (Konsep dan Kasus Penggunaan)

Meskipun pustaka klien inti mencakup banyak kasus penggunaan, API khusus menawarkan peningkatan kinerja atau fungsionalitas untuk tugas-tugas tertentu.

1. BigQuery Storage Read API (Python):

# Membutuhkan: 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

# --- Menggunakan Pandas read_gbq (Integrasi paling sederhana) ---
# Ini secara otomatis menggunakan Storage API jika diinstal dan bermanfaat
# 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, # Secara eksplisit meminta Storage API
#          progress_bar_type='tqdm' # Bilah kemajuan opsional
#      )
#      print("\nBaca data menggunakan Storage API melalui pandas.read_gbq:")
#      print(df_storage.head())
#      print(f"Baca {len(df_storage)} baris.")
# except Exception as e:
#      print(f"Kesalahan membaca dengan Storage API melalui read_gbq: {e}")


# --- Penggunaan Storage API Manual (Kontrol Lebih Banyak) ---
# bqstorageclient = bigquery_storage_v1.BigQueryReadClient()
# table = f"projects/{project_id}/datasets/{dataset_id}/tables/{table_name}" # Ganti dengan detail tabel Anda

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

# read_session = bqstorageclient.create_read_session(
#     parent=parent,
#     read_session=requested_session,
#     max_stream_count=1, # Minta jumlah aliran paralel
# )

# 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("\nBaca data secara manual menggunakan Storage API:")
#     print(df_manual.head())
# else:
#     print("Tidak ada data yang dibaca menggunakan Storage API manual.")

2. BigQuery Connection API:

  1. Gunakan API (atau alat Cloud Console/bq) untuk membuat sumber daya Connection, yang menentukan tipe dan detail sumber eksternal.
  2. Berikan izin yang sesuai pada sumber daya eksternal ke akun layanan koneksi (misalnya, peran Pengguna Cloud SQL).
  3. Gunakan fungsi EXTERNAL_QUERY("connection_id", "external_sql_query") dalam SQL BigQuery Anda.

3. Analytics Hub API:

4. BigLake API:

Menggunakan REST API Secara Langsung

Meskipun pustaka klien umumnya lebih disukai, Anda mungkin menggunakan REST API secara langsung jika:

Membuat Permintaan:

Anda biasanya akan menggunakan klien HTTP (seperti curl atau pustaka requests Python). Anda perlu:

  1. Dapatkan Token Akses OAuth 2.0 (misalnya, menggunakan gcloud auth print-access-token).
  2. Buat URL endpoint API yang benar.
  3. Buat badan permintaan JSON sesuai dengan spesifikasi metode API.
  4. Sertakan token akses di header Authorization: Bearer <token>.
  5. Tangani respons HTTP (kode status, penguraian JSON, pesan kesalahan).

Contoh: Menjalankan Kueri melalui REST (jobs.query)

# 1. Dapatkan Token Akses
TOKEN=$(gcloud auth print-access-token)

# 2. Tentukan ID Project dan Badan Permintaan
PROJECT_ID="your-project-id" # Ganti dengan ID project Anda
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. Buat Panggilan API menggunakan 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}"

# Respons akan berisi informasi pekerjaan, termasuk ID pekerjaan.
# Anda kemudian perlu membuat panggilan berikutnya ke jobs.getQueryResults
# menggunakan ID pekerjaan untuk mengambil baris data aktual setelah pekerjaan selesai.

Contoh ini hanya memulai pekerjaan kueri. Mengambil hasil memerlukan polling status pekerjaan dan kemudian memanggil endpoint jobs.getQueryResults. Ini menyoroti langkah-langkah tambahan yang terlibat dibandingkan dengan pustaka klien.

Praktik Terbaik dan Tips

Kesimpulan

BigQuery API menyediakan cara yang ampuh dan fleksibel untuk berinteraksi dengan gudang data Anda secara terprogram. Apakah Anda perlu mengotomatiskan pemuatan data, menjalankan kueri analitis kompleks sebagai bagian dari aplikasi, mengintegrasikan wawasan BigQuery ke dalam dasbor, atau mengelola sumber daya secara dinamis, API menawarkan alat yang diperlukan.

Dengan memahami berbagai jenis API, menyiapkan lingkungan Anda dengan benar, dan memanfaatkan kenyamanan pustaka klien (terutama di Python), Anda dapat membuka potensi penuh BigQuery di luar konsol interaktif. Mulailah dengan API inti untuk tugas-tugas umum, jelajahi API khusus seperti Storage Read API untuk operasi yang penting bagi kinerja, dan ingat bahwa REST API selalu tersedia untuk kontrol tertinggi. Saat Anda membangun aplikasi di BigQuery, API ini akan menjadi komponen penting dari arsitektur data Anda.

Mengembangkan API dengan Apidog

Apidog adalah alat pengembangan API yang membantu Anda mengembangkan API dengan lebih mudah dan efisien.

Daftar gratisUnduh