Menggunakan metastore BigQuery dengan tabel di BigQuery

Dokumen ini menjelaskan cara menggunakan metastore BigQuery dengan tabel BigQuery dan Spark.

Dengan metastore BigQuery, Anda dapat membuat dan menggunakan tabel standar (bawaan), tabel BigQuery untuk Apache Iceberg, dan tabel eksternal BigLake dari BigQuery.

Sebelum memulai

  1. Mengaktifkan penagihan untuk project Google Cloud Anda. Pelajari cara memeriksa apakah penagihan telah diaktifkan pada suatu project.

  2. Aktifkan BigQuery dan Dataproc API.

    Aktifkan API

  3. Opsional: Pahami cara kerja metastore BigQuery dan alasan Anda harus menggunakannya.

  4. Opsional: Jika Anda menggunakan Iceberg, gunakan kode berikut untuk membandingkan tabel Iceberg metastore BigQuery dengan variasi tabel serupa lainnya di BigQuery:

Tabel BigQuery standar Tabel eksternal BigLake Tabel eksternal BigLake untuk Apache Iceberg (alias tabel BigLake Iceberg) Tabel Iceberg metastore BigQuery (pratinjau) Tabel BigQuery untuk Apache Iceberg (alias tabel yang dikelola Iceberg/ tabel BigQuery Iceberg) (pratinjau)
Fitur utama Pengalaman terkelola sepenuhnya Dikelola (kontrol akses yang mendetail) dan berfungsi di seluruh mesin open source dan BigQuery Fitur tabel eksternal BigLake + konsistensi data, update skema. Tidak dapat dibuat dari Spark atau mesin terbuka lainnya. Fitur tabel BigLake Iceberg + dapat diubah dari mesin eksternal. Tidak dapat dibuat dengan DDL atau alat command line bq. Fitur tabel BigLake Iceberg + overhead pengelolaan yang rendah dengan data dan metadata terbuka
Penyimpanan data Penyimpanan terkelola BigQuery Membuka data format yang dihosting di bucket yang dikelola pengguna
Model terbuka BigQuery Storage Read API (melalui konektor) Format file terbuka (Parquet) Library terbuka (Iceberg) Kompatibel dengan open source (snapshot metadata Iceberg)
Pemerintahan Tata kelola BigQuery terpadu
Penulisan (DML dan Streaming) Melalui konektor BigQuery, API, DML dengan throughput tinggi, CDC Hanya menulis melalui mesin eksternal Melalui konektor BigQuery, API, DML dengan throughput tinggi, CDC

Peran yang diperlukan

Untuk mendapatkan izin yang diperlukan guna menggunakan Spark dan Dataproc dengan metastore BigQuery sebagai penyimpanan metadata, minta administrator untuk memberi Anda peran IAM berikut:

Untuk mengetahui informasi selengkapnya tentang cara memberikan peran, lihat Mengelola akses ke project, folder, dan organisasi.

Anda mungkin juga bisa mendapatkan izin yang diperlukan melalui peran khusus atau peran bawaan lainnya.

Menghubungkan ke tabel

  1. Buat set data di konsol Google Cloud.

    CREATE SCHEMA `PROJECT_ID`.DATASET_NAME;

    Ganti kode berikut:

    • PROJECT_ID: ID Google Cloud project untuk membuat set data.
    • DATASET_NAME: nama untuk set data Anda.

  2. Buat Koneksi Resource Cloud.

  3. Buat tabel BigQuery standar.

    CREATE TABLE `PROJECT_ID`.DATASET_NAME.TABLE_NAME (name STRING,id INT64);

    Ganti kode berikut:

    • TABLE_NAME: nama untuk tabel Anda.

  4. Menyisipkan data ke dalam tabel BigQuery standar.

    INSERT INTO `PROJECT_ID`.DATASET_NAME.TABLE_NAME VALUES ('test_name1', 123),('test_name2', 456),('test_name3', 789);

  5. Buat tabel BigQuery untuk Apache Iceberg.

    Misalnya, untuk membuat tabel, jalankan pernyataan CREATE berikut.

    CREATE TABLE `PROJECT_ID`.DATASET_NAME.ICEBERG_TABLE_NAME(
name STRING,id INT64
)
WITH CONNECTION `CONNECTION_NAME`
OPTIONS (
file_format = 'PARQUET',
table_format = 'ICEBERG',
storage_uri = 'STORAGE_URI');

    Ganti kode berikut:

    • ICEBERG_TABLE_NAME: nama untuk tabel Iceberg Anda. Contoh, iceberg_managed_table.
    • CONNECTION_NAME: nama koneksi Anda. Anda membuatnya di langkah sebelumnya. Contoh, myproject.us.myconnection.
    • STORAGE_URI: Cloud Storage URI yang sepenuhnya memenuhi syarat. Contoh, gs://mybucket/table.

  6. Menyisipkan data ke tabel BigQuery untuk Apache Iceberg.

    INSERT INTO `PROJECT_ID`.DATASET_NAME.ICEBERG_TABLE_NAME VALUES ('test_name1', 123),('test_name2', 456),('test_name3', 789);

  7. Buat tabel Iceberg hanya baca.

    Misalnya, untuk membuat tabel Iceberg hanya baca, jalankan pernyataan CREATE berikut.

    CREATE OR REPLACE EXTERNAL TABLE  `PROJECT_ID`.DATASET_NAME.READONLY_ICEBERG_TABLE_NAME
WITH CONNECTION `CONNECTION_NAME`
OPTIONS (
  format = 'ICEBERG',
  uris =
    ['BUCKET_PATH'],
  require_partition_filter = FALSE);

    Ganti kode berikut:

    • READONLY_ICEBERG_TABLE_NAME: nama untuk tabel hanya baca.
    • BUCKET_PATH: jalur ke bucket Cloud Storage yang berisi data untuk tabel eksternal, dalam format ['gs://bucket_name/[folder_name/]file_name'].

  8. Dari PySpark, buat kueri tabel standar, tabel Iceberg yang dikelola, dan tabel Iceberg hanya baca.

    from pyspark.sql import SparkSession

# Create a spark session
spark = SparkSession.builder \
.appName("BigQuery Metastore Iceberg") \
.config("spark.sql.catalog.CATALOG_NAME", "org.apache.iceberg.spark.SparkCatalog") \
.config("spark.sql.catalog.CATALOG_NAME.catalog-impl", "org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog") \
.config("spark.sql.catalog.CATALOG_NAME.gcp_project", "PROJECT_ID") \
.config("spark.sql.catalog.CATALOG_NAME.gcp_location", "LOCATION") \
.config("spark.sql.catalog.CATALOG_NAME.warehouse", "WAREHOUSE_DIRECTORY") \
.getOrCreate()
spark.conf.set("viewsEnabled","true")

# Use the bqms_catalog
spark.sql("USE `CATALOG_NAME`;")
spark.sql("USE NAMESPACE DATASET_NAME;")

# Configure spark for temp results
spark.sql("CREATE namespace if not exists MATERIALIZATION_NAMESPACE");
spark.conf.set("materializationDataset","MATERIALIZATION_NAMESPACE")

# List the tables in the dataset
df = spark.sql("SHOW TABLES;")
df.show();

# Query a standard BigQuery table
sql = """SELECT * FROM DATASET_NAME.TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

# Query a BigQuery Managed Apache Iceberg table
sql = """SELECT * FROM DATASET_NAME.ICEBERG_TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

# Query a BigQuery Readonly Apache Iceberg table
sql = """SELECT * FROM DATASET_NAME.READONLY_ICEBERG_TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

    Ganti kode berikut:

    • WAREHOUSE_DIRECTORY: URI folder Cloud Storage yang berisi data warehouse Anda.
    • CATALOG_NAME: nama katalog yang Anda gunakan.
    • MATERIALIZATION_NAMESPACE: namespace untuk menyimpan hasil sementara.

  9. Jalankan skrip PySpark menggunakan Serverless Spark.

    gcloud dataproc batches submit pyspark SCRIPT_PATH \
  --version=2.2 \
  --project=PROJECT_ID \
  --region=REGION \
  --deps-bucket=YOUR_BUCKET \
  --jars=https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.5.2/iceberg-spark-runtime-3.5_2.12-1.5.2.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.5.2-1.0.1-beta.jar

    Ganti kode berikut:

    • SCRIPT_PATH: jalur ke skrip yang digunakan tugas batch.
    • PROJECT_ID: ID project Google Cloud tempat menjalankan tugas batch.
    • REGION: region tempat workload Anda berjalan.
    • YOUR_BUCKET: lokasi bucket Cloud Storage untuk mengupload dependensi beban kerja. Awalan URI gs:// bucket tidak diperlukan. Anda dapat menentukan jalur bucket atau nama bucket, misalnya, mybucketname1.

Langkah berikutnya