Create an Apache Iceberg table with the Lakehouse REST Catalog

Managed Service for Apache Spark job

You can submit a job to the Managed Service for Apache Spark service by submitting the job to a Managed Service for Apache Spark cluster using the Google Cloud console or the Google Cloud CLI , or via a HTTP REST request or programmatic gRPC Managed Service for Apache Spark Cloud Client Libraries call to the Managed Service for Apache Spark Jobs API.

The examples in this section show you how to submit a Managed Service for Apache Spark Spark SQL job to the Managed Service for Apache Spark service to create an Iceberg table with metadata in the Lakehouse REST Catalog using the gcloud CLI, Google Cloud console, or Managed Service for Apache Spark REST API.

Prepare job files

Perform the following steps to create a Spark SQL job file. The file contains Spark SQL commands to create and update an Iceberg table.

1. In a local terminal window or in Cloud Shell , use a text editor, such as vi or nano , to copy the following commands into an iceberg-table.sql file, then save the file in the current directory.

   USE CATALOG_NAME 
   ;
   CREATE NAMESPACE IF NOT EXISTS example_namespace;
   USE example_namespace;
   DROP TABLE IF EXISTS example_table;
   CREATE TABLE example_table (id int, data string) USING ICEBERG LOCATION 'gs:// BUCKET 
   / WAREHOUSE_FOLDER 
   ';
   INSERT INTO example_table VALUES (1, 'first row');
   ALTER TABLE example_table ADD COLUMNS (newDoubleCol double);
   DESCRIBE TABLE example_table;

   Replace the following:

   • CATALOG_NAME : Iceberg catalog name.
   • BUCKET and WAREHOUSE_FOLDER : Cloud Storage bucket and folder used for the Iceberg warehouse.

2. Use gcloud CLI to copy the local iceberg-table.sql to your bucket in Cloud Storage.

   gcloud storage cp iceberg-table.sql gs:// BUCKET 
   /

Submit the Spark SQL job

Select a tab to follow the instructions to submit the Spark SQL job to the Managed Service for Apache Spark service using the gcloud CLI, Google Cloud console, or Managed Service for Apache Spark REST API.

POST https://dataproc.googleapis.com/v1/projects/ PROJECT_ID 
/regions/ REGION 
/jobs:submit

{
  "projectId": " PROJECT_ID 
",
  "job": {
    "placement": {
      "clusterName": " CLUSTER_NAME 
"
    },
    "statusHistory": [],
    "reference": {
      "jobId": "",
      "projectId": " PROJECT_ID 
"
    },
    "sparkSqlJob": {
      "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": "gs:// BUCKET 
/ WAREHOUSE_FOLDER 
"
      },
      "jarFileUris": [
        "gs:// BUCKET 
/iceberg-spark-runtime-3.5_2.12-1.6.1.jar, BIGLAKE_ICEBERG_CATALOG_JAR 
"
      ],
      "scriptVariables": {},
      "queryFileUri": "gs:// BUCKET 
/iceberg-table.sql"
    }
  }
}

curl -X POST \
-H "Authorization: Bearer $(gcloud auth print-access-token)" \
-H "Content-Type: application/json; charset=utf-8" \
-d @request.json \
"https://dataproc.googleapis.com/v1/projects/ PROJECT_ID 
/regions/ REGION 
/jobs:submit"

$cred = gcloud auth print-access-token
$headers = @{ "Authorization" = "Bearer $cred" }

Invoke-WebRequest `
-Method POST `
-Headers $headers `
-ContentType: "application/json; charset=utf-8" `
-InFile request.json `
-Uri "https://dataproc.googleapis.com/v1/projects/ PROJECT_ID 
/regions/ REGION 
/jobs:submit" | Select-Object -Expand Content

{
  "reference": {
    "projectId": " PROJECT_ID 
",
    "jobId": "..."
  },
  "placement": {
    "clusterName": " CLUSTER_NAME 
",
    "clusterUuid": "..."
  },
  "status": {
    "state": "PENDING",
    "stateStartTime": "..."
  },
  "submittedBy": " USER 
",
  "sparkSqlJob": {
    "queryFileUri": "gs:// BUCKET 
/iceberg-table.sql",
    "properties": {
      "spark.sql.catalog. USER 
_catalog": "org.apache.iceberg.spark.SparkCatalog",
      "spark.sql.catalog. USER 
_catalog.catalog-impl": "org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog",
      "spark.sql.catalog. USER 
_catalog.gcp_project": " PROJECT_ID 
",
      "spark.sql.catalog. USER 
_catalog.gcp_location": " LOCATION 
",
      "spark.sql.catalog. USER 
_catalog.warehouse": "gs:// BUCKET 
/ WAREHOUSE_FOLDER 
"
    },
    "jarFileUris": [
      "gs:// BUCKET 
/iceberg-spark-runtime-3.5_2.12-1.6.1.jar",
      " BIGLAKE_ICEBERG_CATALOG_JAR 
"
    ]
  },
  "driverControlFilesUri": "gs://dataproc-...",
  "driverOutputResourceUri": "gs://dataproc-.../driveroutput",
  "jobUuid": "...",
  "region": " REGION 
"
}

Spark SQL CLI

The following steps show you how to create an Iceberg table with table metadata stored in the Lakehouse REST Catalog using the Spark SQL CLI running on the master node of a Managed Service for Apache Spark cluster.

1. Use SSH to connect to the master node of your Managed Service for Apache Spark cluster.

2. In the SSH session terminal, use the vi or nano text editor to copy the following commands into an iceberg-table.sql file.

   SET CATALOG_NAME = ` CATALOG_NAME 
   `;
   SET BUCKET = ` BUCKET 
   `;
   SET WAREHOUSE_FOLDER = ` WAREHOUSE_FOLDER 
   `;
   USE `${CATALOG_NAME}`;
   CREATE NAMESPACE IF NOT EXISTS `${CATALOG_NAME}`.example_namespace;
   DROP TABLE IF EXISTS `${CATALOG_NAME}`.example_namespace.example_table;
   CREATE TABLE `${CATALOG_NAME}`.example_namespace.example_table (id int, data string) USING ICEBERG LOCATION 'gs://${BUCKET}/${WAREHOUSE_FOLDER}';
   INSERT INTO `${CATALOG_NAME}`.example_namespace.example_table VALUES (1, 'first row');
   ALTER TABLE `${CATALOG_NAME}`.example_namespace.example_table ADD COLUMNS (newDoubleCol double);
   DESCRIBE TABLE `${CATALOG_NAME}`.example_namespace.example_table;

   Replace the following:

   • CATALOG_NAME : Iceberg catalog name.
   • BUCKET and WAREHOUSE_FOLDER : Cloud Storage bucket and folder used for the Iceberg warehouse.

3. In the SSH session terminal, run the following spark-sql command to create the iceberg table.

   spark-sql \
   --conf spark.sql.catalog. CATALOG_NAME 
   =org.apache.iceberg.spark.SparkCatalog \
   --conf spark.sql.catalog. CATALOG_NAME 
   .type=rest \
   --conf spark.sql.catalog. CATALOG_NAME 
   .uri=https://biglake.googleapis.com/iceberg/v1/restcatalog \
   --conf spark.sql.catalog. CATALOG_NAME 
   .warehouse=gs:// BUCKET 
   / WAREHOUSE_FOLDER 
   \
   --conf spark.sql.catalog. CATALOG_NAME 
   .io-impl=org.apache.iceberg.gcp.gcs.GCSFileIO \
   --conf spark.sql.catalog. CATALOG_NAME 
   .header.x-goog-user-project= PROJECT_ID 
   \
   --conf spark.sql.catalog. CATALOG_NAME 
   .rest.auth.type=org.apache.iceberg.gcp.auth.GoogleAuthManager \
   --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
   -f iceberg-table.sql

   Replace the following:

   • PROJECT_ID : Your Google Cloud project ID.
   • CATALOG_NAME : Iceberg catalog name.
   • BUCKET and WAREHOUSE_FOLDER : Cloud Storage bucket and folder used for the Iceberg warehouse.

4. View table metadata in BigQuery

   1. In the Google Cloud console, go to the BigQuerypage.

      Go to BigQuery Studio

   2. View Iceberg table metadata.

Zeppelin web interface

The following steps show you how to create an Iceberg table with table metadata stored in the Lakehouse REST Catalog using the Zeppelin web interface running on the master node of a Managed Service for Apache Spark cluster.

1. In the Google Cloud console, go to the Managed Service for Apache Spark Clusterspage.

   Go to the Managed Service for Apache Spark Clusters page

2. Select your cluster name to open the Cluster detailspage.

3. Click the Web Interfacestab display a list of Component Gatewaylinks to the web interfaces of default and optional components installed on the cluster.

4. Click the Zeppelinlink to open the Zeppelin web interface.

5. From the Zeppelin notebook menu, click Create new note.

6. In the Create new notedialog, input a name for the notebook, and accept the default sparkinterpreter. Click Createto open the notebook.

7. Copy the following PySpark code into your Zeppelin notebook after filling in the variables.

   %pyspark
   from  
   pyspark.sql  
   import  
   SparkSession
    project_id 
     
    = 
     
    " PROJECT_ID 
   " 
    catalog 
     
    = 
     
    " CATALOG_NAME 
   " 
    namespace 
     
    = 
     
    " NAMESPACE 
   " 
    warehouse_dir 
     
    = 
     
    "gs:// BUCKET 
   / WAREHOUSE_DIRECTORY 
   " 
   
    spark 
     
    = 
     
   SparkSession.builder  
    \ 
   .appName ( 
    "Lakehouse REST Catalog Iceberg" 
    ) 
     
    \ 
   .config ( 
   f "spark.sql.catalog.{catalog}" 
   ,  
    "org.apache.iceberg.spark.SparkCatalog" 
    ) 
     
    \ 
   .config ( 
   f "spark.sql.catalog.{catalog}.type" 
   ,  
    "rest" 
    ) 
     
    \ 
   .config ( 
   f "spark.sql.catalog.{catalog}.uri" 
   ,  
    "https://biglake.googleapis.com/iceberg/v1/restcatalog" 
    ) 
     
    \ 
   .config ( 
   f "spark.sql.catalog.{catalog}.warehouse" 
   ,  
   f "{warehouse_dir}" 
    ) 
     
    \ 
   .config ( 
   f "spark.sql.catalog.{catalog}.io-impl" 
   ,  
    "org.apache.iceberg.gcp.gcs.GCSFileIO" 
    ) 
     
    \ 
   .config ( 
   f "spark.sql.catalog.{catalog}.header.x-goog-user-project" 
   ,  
   f "{project_id}" 
    ) 
     
    \ 
   .config ( 
   f "spark.sql.catalog.{catalog}.rest.auth.type" 
   ,  
    "org.apache.iceberg.gcp.auth.GoogleAuthManager" 
    ) 
     
    \ 
   .config ( 
    "spark.sql.extensions" 
   ,  
    "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions" 
    ) 
     
    \ 
   .getOrCreate () 
   
   spark.sql ( 
   f "USE `{catalog}`;" 
    ) 
   spark.sql ( 
   f "CREATE NAMESPACE IF NOT EXISTS `{namespace}`;" 
    ) 
   spark.sql ( 
   f "USE `{namespace}`;" 
    ) 
   
    \# 
     
   Create  
   table  
   and  
   display  
   schema
   spark.sql ( 
    "DROP TABLE IF EXISTS example_iceberg_table" 
    ) 
   spark.sql ( 
    "CREATE TABLE example_iceberg_table (id int, data string) USING ICEBERG" 
    ) 
   spark.sql ( 
    "DESCRIBE example_iceberg_table;" 
    ) 
   
    \# 
     
   Insert  
   table  
   data.
   spark.sql ( 
    "INSERT INTO example_iceberg_table VALUES (1, 'first row');" 
    ) 
   
    \# 
     
   Alter  
   table,  
    then 
     
   display  
   schema.
   spark.sql ( 
    "ALTER TABLE example_iceberg_table ADD COLUMNS (newDoubleCol double);" 
    ) 
   
    \# 
     
   Select  
   and  
   display  
   the  
   contents  
   of  
   the  
   table.
   spark.sql ( 
    "SELECT * FROM example_iceberg_table" 
    ) 
   .show () 
   

   Replace the following:

   • PROJECT_ID : Your Google Cloud project ID.
   • CATALOG_NAME and NAMESPACE : The Iceberg catalog name and namespace combine to identify the Iceberg table ( catalog.namespace.table_name ).
   • BUCKET and WAREHOUSE_DIRECTORY : Cloud Storage bucket and folder used as Iceberg warehouse directory.

8. Click the run icon or press Shift-Enter to run the code.

9. View table metadata in BigQuery

   1. In the Google Cloud console, go to the BigQuerypage.

      Go to BigQuery Studio

   2. View Iceberg table metadata.