Menyiapkan Spark dengan tabel Delta

Dokumen ini menjelaskan cara mengonfigurasi Apache Spark dengan katalog runtime Lakehouse untuk menggunakan tabel Delta Lake. Anda akan mempelajari cara membuat katalog Apache Hive, mengonfigurasi sesi Spark untuk terhubung ke metastore, dan menjalankan workload untuk membuat dan mendaftarkan tabel Delta untuk dikueri di BigQuery.

Catatan: Anda tidak dapat membuat tabel Delta baru secara langsung melalui endpoint Hive katalog runtime Lakehouse atau menggunakan DDL Hive. Sebagai gantinya, gunakan mesin yang kompatibel dengan Delta seperti Apache Spark untuk membuat file tabel yang mendasarinya di Cloud Storage dan mendaftarkan tabel di metastore.

Sebelum memulai

1. Baca Tentang endpoint katalog Hive.
2. Tinjau Format penyimpanan dan jenis data yang didukung.
3. Tinjau Batasan dan pertimbangan.
Login ke akun Google Cloud Anda. Jika Anda baru menggunakan Google Cloud, buat akun untuk mengevaluasi performa produk kami dalam skenario dunia nyata. Pelanggan baru juga mendapatkan kredit gratis senilai $300 untuk menjalankan, menguji, dan men-deploy workload.

Verify that billing is enabled for your Google Cloud project.

Enable the Lakehouse, Managed Service for Apache Spark APIs, if any are not already enabled.

Roles required to enable APIs

To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

Enable the APIs

Verify that billing is enabled for your Google Cloud project.

Enable the Lakehouse, Managed Service for Apache Spark APIs, if any are not already enabled.

Roles required to enable APIs

To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

Enable the APIs

Peran yang diperlukan

Untuk mendapatkan izin yang Anda perlukan guna menggunakan katalog runtime Lakehouse dengan tabel Delta, minta administrator untuk memberi Anda peran IAM berikut di project Anda:

• Buat cluster Managed Service untuk Apache Spark: Dataproc Editor (roles/dataproc.editor)
• Akun layanan cluster:
  • Dataproc Worker (roles/dataproc.worker)
  • Storage Object User (roles/storage.objectUser)
  • BigLake Editor (roles/biglake.editor)
  • Service Usage Consumer (roles/serviceusage.serviceUsageConsumer)
• Akses tulis ke semua resource katalog runtime Lakehouse: BigLake Editor (roles/biglake.editor)
• Akses hanya baca ke semua resource katalog runtime Lakehouse: BigLake Viewer (roles/biglake.viewer)

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.

Untuk mengetahui petunjuknya, lihat Memberikan satu peran.

Alur kerja umum

Untuk menggunakan katalog runtime Lakehouse dengan tabel Delta, ikuti alur kerja ini:

1. Buat katalog Hive Lakehouse.
2. Konfigurasikan sesi Spark Anda dengan klien metastore Lakehouse dan dependensi Delta Lake.
3. Membuat dan mengisi tabel Delta di Cloud Storage menggunakan Spark, yang mendaftarkannya ke metastore.
4. Buat kueri tabel Delta dari Apache Spark (baca dan tulis) atau langsung dari BigQuery (hanya baca).

Membuat katalog Hive Lakehouse

Untuk menggunakan katalog runtime Lakehouse dengan tabel Spark dan Delta, buat katalog Hive.

Katalog Hive Lakehouse adalah kumpulan database Hive. Sebelum menjalankan tugas Spark, buat katalog untuk mendaftarkannya ke Lakehouse Metastore. Katalog memerlukan nama dan lokasi Cloud Storage untuk data.

Konsol

1. Di konsol Google Cloud , buka halaman Lakehouse.

   Buka Lakehouse

2. Klik Buat katalog.

3. Pilih katalog runtime Lakehouse.

4. Untuk Catalog type, pilih Hive Metastore.

5. Di kolom Select a Cloud Storage bucket, masukkan nama bucket Cloud Storage untuk katalog Anda. Atau, klik Browse untuk memilih atau membuat bucket.

6. Untuk Catalog ID, masukkan nama untuk katalog Lakehouse Hive Anda.

7. Untuk Primary location, tentukan region yang sama dengan bucket Anda.

8. Klik Create.

gcloud

Untuk membuat katalog Hive, jalankan perintah berikut:

gcloud biglake hive catalogs create LAKEHOUSE_CATALOG_ID \
    --project=PROJECT_ID \
    --location-uri="gs://GCS_WAREHOUSE_PATH" \
    --primary-location=REGION \
    --description="DESCRIPTION"

Ganti kode berikut:

• LAKEHOUSE_CATALOG_ID: nama katalog Hive.

• GCS_WAREHOUSE_PATH: jalur Cloud Storage untuk data warehouse Anda.

• PROJECT_ID: Google Cloud Project ID Anda.

• REGION: region utama metastore. Untuk bucket satu region, ini cocok dengan region bucket. Untuk bucket dual-region atau multi-region, tentukan region konstituen untuk replika utama. Region lainnya menjadi replika sekunder.

• DESCRIPTION: deskripsi katalog.

Mengonfigurasi Spark untuk menggunakan tabel Delta

Untuk menggunakan katalog runtime Lakehouse dengan tabel Delta, konfigurasikan sesi Spark Anda dengan factory klien metastore Lakehouse dan ekstensi Delta Lake.

spark-sql

1. Gunakan SSH untuk terhubung ke node utama cluster Managed Service untuk Apache Spark.

2. Jalankan spark-sql di command line dengan properti berikut untuk memulai sesi Spark SQL interaktif:

   spark-sql \
       --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension \
       --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog \
       --conf spark.hive.metastore.client.factory.class=com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory \
       --conf spark.hive.metastore.blms.project.id=PROJECT_ID \
       --conf spark.hive.metastore.blms.catalog.default=LAKEHOUSE_CATALOG_ID \
       --conf spark.hive.metastore.warehouse.dir=gs://GCS_WAREHOUSE_PATH
   

3. Setelah sesi dimulai, Spark akan terhubung ke katalog runtime Lakehouse.

   Jalankan perintah berikut untuk membuat dan mengkueri resource Delta:

   -- Show all the databases in the current project.
   SHOW DATABASES;
   
   -- Create a database.
   CREATE DATABASE spark_delta_blms_database;
   
   -- Create a Delta data source table.
   CREATE TABLE spark_delta_blms_database.delta_quick_start (id INT, name STRING) USING DELTA;
   
   -- Insert data into the table.
   INSERT INTO TABLE spark_delta_blms_database.delta_quick_start VALUES (1, 'my-first-user');
   
   -- Query the Delta table.
   SELECT * FROM spark_delta_blms_database.delta_quick_start;
   

   Ganti kode berikut:

   • PROJECT_ID: project ID Google Cloud Anda.

   • LAKEHOUSE_CATALOG_ID: nama katalog Hive.

   • GCS_WAREHOUSE_PATH: jalur Cloud Storage yang menyimpan data gudang Anda.

Notebook Jupyter

1. Menjalankan notebook Jupyter di cluster Managed Service untuk Apache Spark.
2. Akses antarmuka web Jupyter dari tab Web interfaces di halaman detail cluster di konsol Google Cloud .

3. Di notebook baru, buat sesi Spark yang dikonfigurasi untuk Delta Lake dan jalankan kueri:

   from pyspark.sql import SparkSession
   
   # If a Spark session exists, stop it first by running spark.stop()
   spark = SparkSession.builder\
       .master("local")\
       .appName("Lakehouse runtime catalog Delta tutorial")\
       .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")\
       .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")\
       .config("spark.hive.metastore.client.factory.class", "com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory")\
       .config("spark.hive.metastore.blms.project.id", "PROJECT_ID")\
       .config("spark.hive.metastore.warehouse.dir", "gs://GCS_WAREHOUSE_PATH")\
       .config("spark.hive.metastore.blms.catalog.default", "LAKEHOUSE_CATALOG_ID")\
       .getOrCreate()
   
   # Show all the databases.
   df = spark.sql("SHOW DATABASES;")
   df.show()
   
   # Create a database.
   spark.sql("CREATE DATABASE jupyter_delta_blms_db")
   
   # Create a Delta data source table.
   spark.sql("CREATE TABLE jupyter_delta_blms_db.delta_table(id INT, name STRING) USING DELTA")
   
   # Insert data into the table.
   spark.sql("INSERT INTO TABLE jupyter_delta_blms_db.delta_table VALUES (1, 'my-first-user');")
   
   # Query from table.
   spark.sql("SELECT * FROM jupyter_delta_blms_db.delta_table;").show()
   

   Ganti kode berikut:

   • PROJECT_ID: Project ID Google Cloud Anda.

   • GCS_WAREHOUSE_PATH: jalur Cloud Storage yang menyimpan data gudang Anda.

   • LAKEHOUSE_CATALOG_ID: nama katalog Hive.

Notebook BigQuery

1. Di konsol Google Cloud , buka BigQuery.

   Buka BigQuery

2. Di panel Explorer, klik + ADD, lalu klik Python notebook.

3. Di sel kode, konfigurasikan sesi Managed Service untuk Apache Spark dengan ekstensi Delta dan properti katalog runtime Lakehouse:

   from google.cloud.dataproc_spark_connect import DataprocSparkSession
   from google.cloud.dataproc_v1 import Session
   import os
   
   os.environ['DATAPROC_SPARK_CONNECT_DEFAULT_DATASOURCE'] = ""
   
   session = Session()
   session.environment_config.execution_config.ttl = {"seconds": 864000}
   session.runtime_config.version = "2.3"
   session.runtime_config.properties = {
    "spark.sql.extensions": "io.delta.sql.DeltaSparkSessionExtension",
    "spark.sql.catalog.spark_catalog": "org.apache.spark.sql.delta.catalog.DeltaCatalog",
    "spark.hive.metastore.blms.project.id": "PROJECT_ID",
    "spark.hive.metastore.blms.catalog.default": "LAKEHOUSE_CATALOG_ID",
    "spark.hive.metastore.warehouse.dir": "gs://GCS_WAREHOUSE_PATH",
    "spark.hive.metastore.client.factory.class": "com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory",
    "spark.sql.catalogImplementation": "hive"
   }
   
   spark = DataprocSparkSession.builder.dataprocSessionConfig(session).getOrCreate()
   print("Spark session created successfully")
   

4. Di sel kode lain, buat database dan tabel Delta:

   # Create a database
   spark.sql("CREATE DATABASE bq_spark_delta_database;")
   
   # Create a Delta data source table
   spark.sql("CREATE TABLE bq_spark_delta_database.delta_quick_start (id INT, name STRING) USING DELTA;")
   
   # Insert data into the table
   spark.sql("INSERT INTO TABLE bq_spark_delta_database.delta_quick_start VALUES (1, 'my-first-user');")
   
   # Query from table
   spark.sql("SELECT * FROM bq_spark_delta_database.delta_quick_start;").show()
   

   Ganti kode berikut:

   • PROJECT_ID: Google Cloud Project ID Anda.

   • GCS_WAREHOUSE_PATH: jalur Cloud Storage yang menyimpan data warehouse Anda.

   • LAKEHOUSE_CATALOG_ID: nama katalog Hive.

Mengirimkan tugas batch Managed Service untuk Apache Spark

Anda dapat mengirimkan workload batch PySpark ke Managed Service untuk Apache Spark untuk membuat dan mengkueri tabel Delta dengan katalog runtime Lakehouse.

1. Buat file Python dengan tugas PySpark:

   from pyspark.sql import SparkSession
   
   spark = (
       SparkSession.builder.appName("Lakehouse runtime catalog Delta tutorial")
       .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
       .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
       .config(
           "spark.hive.metastore.client.factory.class",
           "com.google.cloud.bigquery.metastore.client.BigLakeMetastoreClientFactory",
       )
       .config("spark.hive.metastore.blms.project.id", "PROJECT_ID")
       .config("spark.hive.metastore.blms.catalog.default", "LAKEHOUSE_CATALOG_ID")
       .config(
           "spark.hive.metastore.warehouse.dir",
           "gs://GCS_WAREHOUSE_PATH",
       )
       .enableHiveSupport()
       .getOrCreate()
   )
   
   # Show all the databases.
   spark.sql("SHOW DATABASES;").show()
   
   # Create a database.
   spark.sql("CREATE DATABASE dp_serverless_delta_db")
   
   # Create a Delta data source table.
   spark.sql(
       "CREATE TABLE dp_serverless_delta_db.delta_table(id INT, name STRING) USING DELTA"
   )
   
   # Insert data into the table.
   spark.sql("INSERT INTO TABLE dp_serverless_delta_db.delta_table VALUES (1, 'my-first-user');")
   
   # Query from table.
   spark.sql("SELECT * FROM dp_serverless_delta_db.delta_table;").show()
   

   Ganti kode berikut:

   • PROJECT_ID: Project ID Google Cloud Anda.

   • LAKEHOUSE_CATALOG_ID: nama katalog Hive.

   • GCS_WAREHOUSE_PATH: jalur Cloud Storage yang menyimpan data warehouse Anda.

2. Kirim tugas batch:

   gcloud dataproc batches submit pyspark PYTHON_SCRIPT_FILE \
       --version=2.2 \
       --project=PROJECT_ID \
       --region=REGION \
       --deps-bucket=gs://CLOUD_STORAGE_BUCKET
   

   Ganti kode berikut:

   • PYTHON_SCRIPT_FILE: jalur lokal atau Cloud Storage ke file aplikasi PySpark.
   • PROJECT_ID: project ID Anda.
   • REGION: region tempat menjalankan tugas batch.
   • CLOUD_STORAGE_BUCKET: bucket Cloud Storage untuk melakukan staging dependensi workload.

Membuat kueri tabel Delta dari BigQuery

Setelah membuat tabel Delta dari Spark di katalog runtime Lakehouse, Anda dapat mengkuerinya dari BigQuery Studio.

Kueri yang dijalankan dari BigQuery pada tabel Delta di endpoint katalog Hive bersifat hanya baca.

1. Di konsol Google Cloud , buka BigQuery.

   Buka BigQuery

2. Di editor kueri, masukkan pernyataan berikut:

   SELECT * FROM `PROJECT_ID.LAKEHOUSE_CATALOG_ID.DATABASE_NAME.TABLE_NAME`;
   

   Ganti kode berikut:

   • PROJECT_ID: project ID Anda.
   • LAKEHOUSE_CATALOG_ID: nama katalog Hive Anda.
   • DATABASE_NAME: nama database Anda.
   • TABLE_NAME: nama tabel Delta Anda.

Batasan

Saat menggunakan tabel Delta dengan endpoint katalog Hive katalog runtime Lakehouse, batasan berikut berlaku:

• Kueri hanya baca di BigQuery: Kueri pada tabel Delta dari BigQuery bersifat hanya baca. Lakukan modifikasi data, operasi DML, dan perubahan skema menggunakan Apache Spark.
• Skema placeholder di GetHiveTable dan pemeriksaan skema yang sebenarnya: Panggilan GetHiveTable untuk tabel Delta menampilkan skema placeholder, bukan skema tabel yang sebenarnya. Untuk memeriksa skema sebenarnya dari tabel Delta, gunakan Apache Spark. BigQuery menghitung skema yang dimodifikasi dan disesuaikan untuk eksekusi kueri Dremel, yang tidak cocok 1:1 dengan skema tabel Delta yang sebenarnya.
• Ketidakcocokan klien Hive: Klien Hive (seperti Hive CLI) tidak mendukung tabel Delta dengan endpoint Hive katalog runtime Lakehouse. Gunakan Apache Spark untuk mengelola tabel Delta, dan Spark atau BigQuery untuk membuat kueri tabel tersebut.
• Operasi yang dikelola katalog tidak didukung: Mode yang dikelola katalog tidak didukung untuk tabel Delta. Anda tidak dapat membuat tabel Delta menggunakan DDL metastore mandiri. Sebagai gantinya, buat tabel Delta dengan lokasi penyimpanan eksternal di Cloud Storage (misalnya, menggunakan Apache Spark) dan daftarkan di katalog.
• Batasan tabel Delta eksternal BigQuery: Endpoint Hive katalog runtime Lakehouse menggunakan implementasi mesin kueri dasar yang sama dengan tabel Delta Lake eksternal BigQuery. Semua batasan tabel Delta Lake eksternal dan batasan tabel BigLake berlaku. Misalnya, operasi Change Data Capture (CDC) dan BigQuery Storage Read API tidak didukung, dan timestamp_ntz tidak didukung.

Langkah berikutnya

• Tentang endpoint katalog Hive.
• Format penyimpanan dan jenis data yang didukung.
• Batasan dan pertimbangan untuk katalog Hive.