Menggunakan metastore BigLake dengan tabel di BigQuery

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

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

Sebelum memulai

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

  2. Aktifkan BigQuery dan Dataproc API.

    Aktifkan API

Peran yang diperlukan

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

Untuk mengetahui informasi selengkapnya tentang pemberian 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 BigLake untuk Apache Iceberg di BigQuery.

    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 BigLake untuk Apache Iceberg di BigQuery. Contoh, iceberg_managed_table.
    • CONNECTION_NAME: nama koneksi Anda. Anda membuatnya pada langkah sebelumnya. Contoh, myproject.us.myconnection.
    • STORAGE_URI: URI Cloud Storage yang sepenuhnya memenuhi syarat. Contoh, gs://mybucket/table.

  6. Menyisipkan data ke dalam tabel BigLake untuk Apache Iceberg di BigQuery.

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

  7. Buat tabel eksternal Apache Iceberg.

    Misalnya, untuk membuat tabel eksternal Iceberg, 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 Anda.
    • 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, kueri tabel standar, tabel BigLake untuk Apache Iceberg di BigQuery, dan tabel eksternal Apache Iceberg.

    from pyspark.sql import SparkSession

# Create a spark session
spark = SparkSession.builder \
.appName("BigLake 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 blms_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 the tables
sql = """SELECT * FROM DATASET_NAME.TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

sql = """SELECT * FROM DATASET_NAME.ICEBERG_TABLE_NAME"""
df = spark.read.format("bigquery").load(sql)
df.show()

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 terhubung ke tabel BigLake Iceberg di BigQuery dan tabel eksternal Iceberg.
    • 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 \

    Ganti kode berikut:

    • SCRIPT_PATH: jalur ke skrip yang digunakan tugas batch.
    • PROJECT_ID: ID Google Cloud project untuk 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