Use BigQuery metastore with Dataproc

This document explains how to use BigQuery metastore with Dataproc on Compute Engine. This connection provides you with a single, shared metastore that works across open source software engines, such as Apache Spark or Apache Flink.

Before you begin

1. Enable billing for your Google Cloud project. Learn how to check if billing is enabled on a project.

2. Enable the BigQuery, and Dataproc APIs.

   Enable the APIs

3. Optional: Understand how BigQuery metastore works and why you should use it.

Required roles

To get the permissions that you need to use Spark or Flink and Dataproc with BigQuery metastore as a metadata store, ask your administrator to grant you the following IAM roles:

• Create a Dataproc cluster: Dataproc Worker (roles/dataproc.worker) on the Compute Engine default service account in the project
• Create BigQuery metastore tables in Spark or Flink:
  • Dataproc Worker (roles/dataproc.worker) on the Dataproc VM service account in the project
  • BigQuery Data Editor (roles/bigquery.dataEditor) on the Dataproc VM service account in the project
  • Storage Object Admin (roles/storage.objectAdmin) on the Dataproc VM service account in the project
• Query BigQuery metastore tables in BigQuery:
  • BigQuery Data Viewer (roles/bigquery.dataViewer) on the project
  • BigQuery User (roles/bigquery.user) on the project
  • Storage Object Viewer (roles/storage.objectViewer) on the project

For more information about granting roles, see Manage access to projects, folders, and organizations.

You might also be able to get the required permissions through custom roles or other predefined roles.

General workflow

To use Dataproc on Compute Engine with BigQuery metastore, you follow these general steps:

1. Create a Dataproc cluster or configure an existing cluster.
2. Connect to your preferred open source software engine, such as Spark or Flink.
3. Use a JAR file to install the Apache Iceberg catalog plugin on the cluster.
4. Create and manage your BigQuery metastore resources as needed, depending on the open source software engine that you're using.
5. In BigQuery, access and use your BigQuery metastore resources.

Connect BigQuery metastore to Spark

The following instructions show you how to connect Dataproc to BigQuery metastore using interactive Spark SQL.

Download the Iceberg catalog plugin

To connect BigQuery metastore with Dataproc and Spark, you must use the BigQuery metastore Iceberg catalog plugin jar file.

This file is included by default in Dataproc image version 2.2. If your Dataproc clusters don't have direct access to the internet, you must download the plugin and upload it to a Cloud Storage bucket that your Dataproc cluster can access.

Download the BigQuery metastore Apache Iceberg catalog plugin.

Configure a Dataproc cluster

Before you connect to BigQuery metastore, you must set up a Dataproc cluster.

To do this, you can create a new cluster or use an existing cluster. After, you use this cluster to run interactive Spark SQL and manage your BigQuery metastore resources.

• The subnet in the region where the cluster is created must have Private Google Access (PGA) enabled. By default, Dataproc cluster VMs, created with a 2.2 (default) or later image version, have internal IP addresses only. To allow cluster VMs to communicate with Google APIs, enable Private Google Access on the default (or user-specified network name, if applicable) network subnet in the region where the cluster is created.

• If you want to run the Zeppelin web interface example in this guide, you must use or create a Dataproc cluster with the Zeppelin optional component enabled.

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

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

Submit a Spark job

To submit a Spark job, use one of the following methods:

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"

Connect BigQuery metastore to Flink

The following instructions show you how to connect Dataproc to BigQuery metastore using the Flink SQL client.

Install the catalog plugin and connect to a Flink session

To connect BigQuery metastore to Flink, do the following:

1. Create a Dataproc cluster with the optional Flink component enabled, and ensure that you're using Dataproc 2.2 or later.

2. In the Google Cloud console, go to the VM instances page.

   Go to VM instances

3. In the list of virtual machine instances, click SSH to connect to a Dataproc VM instance.

4. Configure the Iceberg custom catalog plugin for BigQuery metastore:

   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/

5. Start the Flink session on YARN:

   HADOOP_CLASSPATH=`hadoop classpath`
   
   sudo bin/yarn-session.sh -nm flink-dataproc -d
   
   sudo bin/sql-client.sh embedded \
     -s yarn-session

6. Create a catalog in 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'
   );

   Replace the following:

   • CATALOG_NAME: the Flink catalog identifier, which is linked to a BigQuery metastore catalog.
   • WAREHOUSE_DIRECTORY: the base path for the warehouse directory (the Cloud Storage folder where Flink creates files). This value starts with gs://.
   • PROJECT_ID: the project ID of the BigQuery metastore catalog that the Flink catalog links with.
   • LOCATION: the location of the BigQuery resources.

Your Flink session is now connected to BigQuery metastore, and you can run Flink SQL commands.

Manage BigQuery metastore resources

Now that you're connected to BigQuery metastore, you can create and view resources based on the metadata stored in BigQuery metastore.

For example, try running the following commands in your interactive Flink SQL session to create an Iceberg database and table.

1. Use the custom Iceberg catalog:

   USE CATALOG CATALOG_NAME;

   Replace CATALOG_NAME with your Flink catalog identifier.

2. Create a database, which creates a dataset in BigQuery:

   CREATE DATABASE IF NOT EXISTS DATABASE_NAME;

   Replace DATABASE_NAME with the name of your new database.

3. Use the database that you created:

   USE DATABASE_NAME;

4. Create an Iceberg table. The following creates an example sales table:

   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)
   );

   Replace ICEBERG_TABLE_NAME with a name for your new table.

5. View table metadata:

   DESCRIBE EXTENDED ICEBERG_TABLE_NAME;

6. List tables in the database:

   SHOW TABLES;

Ingest data into your table

After you create an Iceberg table in the previous section, you can use Flink DataGen as a data source to ingest real-time data into your table. The following steps are an example of this workflow:

1. Create a temporary table using 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);

   Replace the following:

   • DATABASE_NAME: the name of the database to store your temporary table.
   • TEMP_TABLE_NAME: a name for your temporary table.
   • ICEBERG_TABLE_NAME: the name of the Iceberg table that you created in the previous section.

2. Set the parallelism to 1:

   SET 'parallelism.default' = '1';

3. Set the checkpoint interval:

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

4. Set the checkpoint:

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

5. Start the real-time streaming job:

   INSERT INTO ICEBERG_TABLE_NAME SELECT * FROM TEMP_TABLE_NAME;

   The output is similar to the following:

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

6. To check the status of the streaming job, do the following:

   1. In the Google Cloud console, go to the Clusters page.

      Go to Clusters

   2. Select your cluster.

   3. Click the Web interfaces tab.

   4. Click the YARN ResourceManager link.

   5. In the YARN ResourceManager interface, find your Flink session, and click the ApplicationMaster link under Tracking UI.

   6. In the Status column, confirm that your job status is Running.

7. Query streaming data in the Flink SQL client:

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

8. Query streaming data in BigQuery:

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

9. Terminate the streaming job in the Flink SQL client:

   STOP JOB 'JOB_ID';

   Replace JOB_ID with the job ID that was displayed in the output when you created the streaming job.

What's next

• Set up optional BigQuery metastore features.
• View and query tables from Spark in BigQuery.