Halaman ini diterjemahkan oleh Cloud Translation API.

Menggunakan metastore BigQuery dengan Dataproc

Dokumen ini menjelaskan cara menggunakan metastore BigQuery dengan Dataproc di Compute Engine. Koneksi ini memberi Anda satu metastore bersama yang berfungsi di seluruh mesin software open source, seperti Apache Spark atau Apache Flink.

Sebelum memulai

Aktifkan penagihan untuk project Google Cloud Anda. Pelajari cara memeriksa apakah penagihan telah diaktifkan pada suatu project.
Aktifkan BigQuery dan Dataproc API.

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

Peran yang diperlukan

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

Buat cluster Dataproc: Dataproc Worker (roles/dataproc.worker) di akun layanan default Compute Engine dalam project
Buat tabel metastore BigQuery di Spark atau Flink:
- Dataproc Worker (roles/dataproc.worker) di akun layanan VM Dataproc dalam project
- BigQuery Data Editor (roles/bigquery.dataEditor) di akun layanan VM Dataproc dalam project
- Storage Object Admin (roles/storage.objectAdmin) di akun layanan VM Dataproc dalam project
Buat kueri tabel metastore BigQuery di BigQuery:
- BigQuery Data Viewer (roles/bigquery.dataViewer) di project
- Pengguna BigQuery (roles/bigquery.user) di project
- Storage Object Viewer (roles/storage.objectViewer) di project

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.

Alur kerja umum

Untuk menggunakan Dataproc di Compute Engine dengan metastore BigQuery, ikuti langkah-langkah umum berikut:

Buat cluster Dataproc atau konfigurasikan cluster yang ada.
Hubungkan ke mesin software open source pilihan Anda, seperti Spark atau Flink.
Gunakan file JAR untuk menginstal plugin katalog Apache Iceberg di cluster.
Buat dan kelola resource metastore BigQuery sesuai kebutuhan, bergantung pada mesin software open source yang Anda gunakan.
Di BigQuery, akses dan gunakan resource metastore BigQuery Anda.

Menghubungkan metastore BigQuery ke Spark

Petunjuk berikut menunjukkan cara menghubungkan Dataproc ke metastore BigQuery menggunakan Spark SQL interaktif.

Mendownload plugin katalog Iceberg

Untuk menghubungkan metastore BigQuery dengan Dataproc dan Spark, Anda harus menggunakan file jar plugin katalog Iceberg metastore BigQuery.

File ini disertakan secara default dalam image Dataproc versi 2.2. Jika cluster Dataproc Anda tidak memiliki akses langsung ke internet, Anda harus mendownload plugin dan menguploadnya ke bucket Cloud Storage yang dapat diakses oleh cluster Dataproc.

Mendownload plugin katalog Iceberg metastore BigQuery.

Mengonfigurasi cluster Dataproc

Sebelum terhubung ke metastore BigQuery, Anda harus menyiapkan cluster Dataproc.

Untuk melakukannya, Anda dapat membuat cluster baru atau menggunakan cluster yang ada. Setelah itu, Anda akan menggunakan cluster ini untuk menjalankan Spark SQL interaktif dan mengelola resource metastore BigQuery.

Subnet di region tempat cluster dibuat harus mengaktifkan Akses Google Pribadi (PGA). Secara default, VM cluster Dataproc, yang dibuat dengan versi image 2.2 (default) atau yang lebih baru, memiliki alamat IP internal saja. Agar VM cluster dapat berkomunikasi dengan Google API, aktifkan Akses Google Pribadi di subnet jaringan default (atau nama jaringan yang ditentukan pengguna, jika berlaku) di region tempat cluster dibuat.
Jika ingin menjalankan contoh antarmuka web Zeppelin dalam panduan ini, Anda harus menggunakan atau membuat cluster Dataproc dengan komponen opsional Zeppelin yang diaktifkan.

Cluster baru

Untuk membuat cluster Dataproc baru, jalankan perintah gcloud dataproc clusters create berikut. Konfigurasi ini berisi setelan yang Anda perlukan untuk menggunakan metastore BigQuery.

gcloud dataproc clusters create CLUSTER_NAME \
  --project=PROJECT_ID \
  --region=LOCATION \
  --optional-components=ZEPPELIN \
  --enable-component-gateway \
  --single-node

Ganti kode berikut:

CLUSTER_NAME: nama untuk cluster Dataproc Anda.
PROJECT_ID: ID Google Cloud project tempat Anda membuat cluster.
LOCATION: region Google Cloud tempat Anda membuat cluster.

Cluster yang ada

Untuk mengonfigurasi cluster yang ada, tambahkan runtime Iceberg Spark berikut ke cluster Anda.

org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.6.1

Anda dapat menambahkan runtime menggunakan salah satu opsi berikut:

Skrip Inisialisasi. Tambahkan dependensi runtime ke skrip inisialisasi kustom yang berjalan saat dibuat.

Setelah menambahkan dependensi runtime ke skrip, ikuti petunjuk untuk membuat ulang dan mengupdate cluster.
Penginstalan Manual. Tambahkan file JAR plugin katalog Iceberg secara manual dan konfigurasikan properti Spark untuk menyertakan runtime di cluster Anda.

Mengirim tugas Spark

Untuk mengirimkan tugas Spark, gunakan salah satu metode berikut:

gcloud CLI

gcloud dataproc jobs submit spark-sql \
--project=PROJECT_ID \
--cluster=CLUSTER_NAME \
--region==REGION \
--jars=https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.6.1/iceberg-spark-runtime-3.5_2.12-1.6.1.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.6.1-1.0.1-beta.jar \
--properties=spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog, \
spark.sql.catalog.CATALOG_NAME.catalog-impl=org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog, \
spark.sql.catalog.CATALOG_NAME.gcp_project=PROJECT_ID, \
spark.sql.catalog.CATALOG_NAME.gcp_location=LOCATION, \
spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY \
--execute="SPARK_SQL_COMMAND"

Ganti kode berikut:

PROJECT_ID: ID Google Cloud project yang berisi cluster Dataproc.
CLUSTER_NAME: nama cluster Dataproc yang Anda gunakan untuk menjalankan tugas Spark SQL.
REGION: region Compute Engine tempat cluster Anda berada.
LOCATION: lokasi resource BigQuery.
CATALOG_NAME: nama katalog Spark yang Anda gunakan dengan tugas SQL.
WAREHOUSE_DIRECTORY: folder Cloud Storage yang berisi data warehouse Anda. Nilai ini diawali dengan gs://.
SPARK_SQL_COMMAND: kueri Spark SQL yang ingin Anda jalankan. Kueri ini menyertakan perintah untuk membuat resource Anda. Misalnya, untuk membuat namespace dan tabel.

Spark Interaktif

Menghubungkan ke Spark dan menginstal plugin katalog

Untuk menginstal plugin katalog untuk metastore BigQuery, hubungkan ke cluster Dataproc menggunakan SSH.

Di konsol Google Cloud , buka halaman VM Instances.
Untuk terhubung ke instance VM Dataproc, klik SSH di daftar instance virtual machine. Outputnya mirip dengan hal berikut ini:
```
Connected, host fingerprint: ssh-rsa ...
Linux cluster-1-m 3.16.0-0.bpo.4-amd64 ...
...
example-cluster@cluster-1-m:~$
```

Di terminal, jalankan perintah inisialisasi metastore BigQuery berikut:

spark-sql \
--jars https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.6.1/iceberg-spark-runtime-3.5_2.12-1.6.1.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.6.1-1.0.1-beta.jar \
--conf spark.sql.catalog.CATALOG_NAME=org.apache.iceberg.spark.SparkCatalog \
--conf spark.sql.catalog.CATALOG_NAME.catalog-impl=org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog \
--conf spark.sql.catalog.CATALOG_NAME.gcp_project=PROJECT_ID \
--conf spark.sql.catalog.CATALOG_NAME.gcp_location=LOCATION \
--conf spark.sql.catalog.CATALOG_NAME.warehouse=WAREHOUSE_DIRECTORY

Ganti kode berikut:

CATALOG_NAME: nama katalog Spark yang Anda gunakan dengan tugas SQL.
PROJECT_ID: Google Cloud project ID katalog metastore BigQuery yang ditautkan dengan katalog Spark Anda.
LOCATION: Google Cloud lokasi metastore BigQuery.
WAREHOUSE_DIRECTORY: folder Cloud Storage yang berisi data warehouse Anda. Nilai ini diawali dengan gs://.

Setelah berhasil terhubung ke cluster, terminal Spark akan menampilkan perintah spark-sql.

spark-sql (default)>

Mengelola resource metastore BigQuery

Anda kini terhubung ke metastore BigQuery. Anda dapat melihat resource yang ada atau membuat resource baru berdasarkan metadata yang disimpan di metastore BigQuery.

Misalnya, coba jalankan perintah berikut dalam sesi Spark SQL interaktif untuk membuat namespace dan tabel Iceberg.

Gunakan katalog Iceberg kustom:
```
USE `CATALOG_NAME`;
```

Buat namespace

CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME;

Gunakan namespace yang dibuat:
```
USE NAMESPACE_NAME;
```

Buat tabel Iceberg:

CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG;

Menyisipkan baris tabel:

INSERT INTO TABLE_NAME VALUES (1, "first row");

Tambahkan kolom tabel:

ALTER TABLE TABLE_NAME ADD COLUMNS (newDoubleCol double);

Melihat metadata tabel:
```
DESCRIBE EXTENDED TABLE_NAME;
```
Mencantumkan tabel dalam namespace:
```
SHOW TABLES;
```

Notebook Zeppelin

Di Google Cloud console, buka halaman Dataproc Clusters.

Buka cluster Dataproc
Klik nama cluster yang ingin Anda gunakan.

Halaman Cluster Details akan terbuka.
Di menu navigasi, klik Antarmuka web.
Di bagian Component gateway, klik Zeppelin. Halaman notebook Zeppelin akan terbuka.
Di menu navigasi, klik Notebook, lalu klik +Create new note.
Dalam dialog, masukkan nama notebook. Biarkan Spark dipilih sebagai interpretor default.
Klik Buat. Notebook baru akan dibuat.
Di notebook, klik menu setelan, lalu klik Interpreter.
Di kolom Telusuri penafsir, telusuri Spark.
Klik Edit.

Di kolom Spark.jars, masukkan URI jar Spark.

https://storage-download.googleapis.com/maven-central/maven2/org/apache/iceberg/iceberg-spark-runtime-3.5_2.12/1.6.1/iceberg-spark-runtime-3.5_2.12-1.6.1.jar,gs://spark-lib/bigquery/iceberg-bigquery-catalog-1.6.1-1.0.1-beta.jar

Klik Simpan.
Klik Oke.

Salin kode PySpark berikut ke notebook Zeppelin Anda.

%pyspark
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName("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.sql("select version()").show()
spark.sql("USE `CATALOG_NAME`;")
spark.sql("CREATE NAMESPACE IF NOT EXISTS NAMESPACE_NAME;")
spark.sql("USE NAMESPACE_NAME;")
spark.sql("CREATE TABLE TABLE_NAME (id int, data string) USING ICEBERG;")
spark.sql("DESCRIBE TABLE_NAME;").show()

Ganti kode berikut:

CATALOG_NAME: nama katalog Spark yang akan digunakan untuk tugas SQL.
PROJECT_ID: ID Google Cloud project yang berisi cluster Dataproc.
WAREHOUSE_DIRECTORY: folder Cloud Storage yang berisi data warehouse Anda. Nilai ini diawali dengan gs://.
NAMESPACE_NAME: nama namespace yang mereferensikan tabel Spark Anda.
WAREHOUSE_DIRECTORY: URI folder Cloud Storage tempat data warehouse Anda disimpan.
TABLE_NAME: nama tabel untuk tabel Spark Anda.

Klik ikon jalankan atau tekan Shift-Enter untuk menjalankan kode. Setelah tugas selesai, pesan status akan menampilkan "Spark Job Finished", dan output akan menampilkan konten tabel:

Menghubungkan metastore BigQuery ke Flink

Petunjuk berikut menunjukkan cara menghubungkan Dataproc ke metastore BigQuery menggunakan klien Flink SQL.

Menginstal plugin katalog dan terhubung ke sesi Flink

Untuk menghubungkan metastore BigQuery ke Flink, lakukan hal berikut:

Buat cluster Dataproc dengan komponen Flink opsional yang diaktifkan, dan pastikan Anda menggunakan Dataproc 2.2 atau yang lebih baru.
Di konsol Google Cloud , buka halaman Instance VM.

Buka instance VM
Dalam daftar instance virtual machine, klik SSH untuk terhubung ke instance VM Dataproc.

Konfigurasikan plugin katalog kustom Iceberg untuk metastore BigQuery:

FLINK_VERSION=1.17
ICEBERG_VERSION=1.5.2

cd /usr/lib/flink

sudo wget -c https://repo.maven.apache.org/maven2/org/apache/iceberg/iceberg-flink-runtime-${FLINK_VERSION}/${ICEBERG_VERSION}/iceberg-flink-runtime-${FLINK_VERSION}-${ICEBERG_VERSION}.jar -P lib

sudo gcloud storage cp gs://spark-lib/bigquery/iceberg-bigquery-catalog-${ICEBERG_VERSION}-1.0.1-beta.jar lib/

Mulai sesi Flink di YARN:

HADOOP_CLASSPATH=`hadoop classpath`

sudo bin/yarn-session.sh -nm flink-dataproc -d

sudo bin/sql-client.sh embedded \
  -s yarn-session

Buat katalog di Flink:
```
CREATE CATALOG CATALOG_NAME WITH (
  'type'='iceberg',
  'warehouse'='WAREHOUSE_DIRECTORY',
  'catalog-impl'='org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog',
  'gcp_project'='PROJECT_ID',
  'gcp_location'='LOCATION'
);
```
Ganti kode berikut:
- CATALOG_NAME: ID katalog Flink, yang ditautkan ke katalog metastore BigQuery.
- WAREHOUSE_DIRECTORY: jalur dasar untuk direktori warehouse (folder Cloud Storage tempat Flink membuat file). Nilai ini dimulai dengan gs://.
- PROJECT_ID: project ID katalog metastore BigQuery yang ditautkan oleh katalog Flink.
- LOCATION: Lokasi resource BigQuery.

Sesi Flink Anda kini terhubung ke metastore BigQuery, dan Anda dapat menjalankan perintah Flink SQL.

Mengelola resource metastore BigQuery

Setelah terhubung ke metastore BigQuery, Anda dapat membuat dan melihat resource berdasarkan metadata yang disimpan di metastore BigQuery.

Misalnya, coba jalankan perintah berikut di sesi SQL Flink interaktif untuk membuat database dan tabel Iceberg.

Gunakan katalog Iceberg kustom:
```
USE CATALOG CATALOG_NAME;
```
Ganti CATALOG_NAME dengan ID katalog Flink Anda.
Buat database, yang akan membuat set data di BigQuery:
```
CREATE DATABASE IF NOT EXISTS DATABASE_NAME;
```
Ganti DATABASE_NAME dengan nama database baru Anda.
Gunakan database yang Anda buat:
```
USE DATABASE_NAME;
```

Buat tabel Iceberg. Berikut ini contoh pembuatan tabel penjualan:

CREATE TABLE IF NOT EXISTS ICEBERG_TABLE_NAME (
    order_number BIGINT,
    price        DECIMAL(32,2),
    buyer        ROW<first_name STRING, last_name STRING>,
    order_time   TIMESTAMP(3)
);

Ganti ICEBERG_TABLE_NAME dengan nama untuk tabel baru Anda.

Melihat metadata tabel:
```
DESCRIBE EXTENDED ICEBERG_TABLE_NAME;
```
Menampilkan daftar tabel dalam database:
```
SHOW TABLES;
```

Menyerap data ke dalam tabel

Setelah membuat tabel Iceberg di bagian sebelumnya, Anda dapat menggunakan Flink DataGen sebagai sumber data untuk menyerap data real-time ke dalam tabel. Langkah-langkah berikut adalah contoh alur kerja ini:

Buat tabel sementara menggunakan DataGen:

CREATE TEMPORARY TABLE DATABASE_NAME.TEMP_TABLE_NAME
WITH (
    'connector' = 'datagen',
    'rows-per-second' = '10',
    'fields.order_number.kind' = 'sequence',
    'fields.order_number.start' = '1',
    'fields.order_number.end' = '1000000',
    'fields.price.min' = '0',
    'fields.price.max' = '10000',
    'fields.buyer.first_name.length' = '10',
    'fields.buyer.last_name.length' = '10'
)
LIKE DATABASE_NAME.ICEBERG_TABLE_NAME (EXCLUDING ALL);

Ganti kode berikut:

DATABASE_NAME: nama database untuk menyimpan tabel sementara Anda.
TEMP_TABLE_NAME: nama untuk tabel sementara Anda.
ICEBERG_TABLE_NAME: nama tabel Iceberg yang Anda buat di bagian sebelumnya.

Tetapkan paralelisme ke 1:
```
SET 'parallelism.default' = '1';
```

Tetapkan interval checkpoint:

SET 'execution.checkpointing.interval' = '10second';

Tetapkan checkpoint:

SET 'state.checkpoints.dir' = 'hdfs:///flink/checkpoints';

Mulai tugas streaming real-time:

INSERT INTO ICEBERG_TABLE_NAME SELECT * FROM TEMP_TABLE_NAME;

Outputnya mirip dengan hal berikut ini:

[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 0de23327237ad8a811d37748acd9c10b

Untuk memeriksa status tugas streaming, lakukan hal berikut:
1. Di Google Cloud console, buka halaman Clusters.
  
  Buka Cluster
2. Pilih cluster Anda.
3. Klik tab Antarmuka web.
4. Klik link YARN ResourceManager.
5. Di antarmuka YARN ResourceManager, temukan sesi Flink Anda, lalu klik link ApplicationMaster di bagian Tracking UI.
6. Di kolom Status, pastikan status tugas Anda adalah Running.

Buat kueri data streaming di klien Flink SQL:

SELECT * FROM ICEBERG_TABLE_NAME
/*+ OPTIONS('streaming'='true', 'monitor-interval'='3s')*/
ORDER BY order_time desc
LIMIT 20;

Buat kueri data streaming di BigQuery:

SELECT * FROM `DATABASE_NAME.ICEBERG_TABLE_NAME`
ORDER BY order_time desc
LIMIT 20;

Hentikan tugas streaming di klien Flink SQL:
```
STOP JOB 'JOB_ID';
```
Ganti JOB_ID dengan ID tugas yang ditampilkan dalam output saat Anda membuat tugas streaming.

Menggunakan metastore BigQuery dengan Dataproc

Sebelum memulai

Peran yang diperlukan

Alur kerja umum

Menghubungkan metastore BigQuery ke Spark

Mendownload plugin katalog Iceberg

Mengonfigurasi cluster Dataproc

Cluster baru

Cluster yang ada

Mengirim tugas Spark

gcloud CLI

Spark Interaktif

Menghubungkan ke Spark dan menginstal plugin katalog

Mengelola resource metastore BigQuery

Notebook Zeppelin

Menghubungkan metastore BigQuery ke Flink

Menginstal plugin katalog dan terhubung ke sesi Flink

Mengelola resource metastore BigQuery

Menyerap data ke dalam tabel

Langkah berikutnya