Google Colab による Google Cloud Storage上でのApache Icebergデータ操作方法

## 概要

Google Colab から Google Cloud Storage に Apace Iceberg 形式のデータを書き込む方法を紹介します。

Google Colab の Spark にて Apache Iceberg 形式のデータの基本操作については下記の記事で紹介しています。

- [Getting started with PyIceberg と Iceberg で作成されるファイル郡の確認を Google Colab でやってみた #Python - Qiita](https://qiita.com/manabian/items/c7d8204a565e63a99764)
- [Google Colab の Spark にて Hadoop Catalog の Apache Iceberg の動作検証 #iceberg - Qiita](https://qiita.com/manabian/items/99aa3396bc15a28f9e10)

## 事前準備

下記の記事を参考にして Google Colab と GCS を接続してください。

- [Google Colab の Spark にて Google Cloud Storage と連携する方法 #GoogleCloud - Qiita](https://qiita.com/manabian/items/3ef03f927a0230a57f05)

## Apache Icebeg 形式のデータの操作方法

### Service Account のキーファイルをストレージに配置後、Google Drive をマウント

```python
from google.colab import drive
drive.mount('/content/drive')
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b1e69a86-78f4-4258-9d56-4a72f8b1bd21.png)

### SparkSession の定義

```python
gcs_iceberg_path = "gs://manabianqiita/iceberg"
gcs_key_path = "/content/drive/MyDrive/Colab Notebooks/bigquery_iceberg/gcs_key.json"
```

```python
from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
        .appName("gcs_test")
        .config("spark.jars", "https://storage.googleapis.com/hadoop-lib/gcs/gcs-connector-hadoop3-latest.jar")
        .config('spark.jars.packages',
            'org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.1,'
            'org.apache.iceberg:iceberg-gcp-bundle:1.8.1')
        .config('spark.sql.extensions', 'org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions')
        .config('spark.sql.catalog.spark_catalog', 'org.apache.iceberg.spark.SparkCatalog')
        .config("spark.sql.catalog.spark_catalog.type", "hadoop")
        .config("spark.sql.catalog.spark_catalog.warehouse", gcs_iceberg_path)
        .config("spark.hadoop.fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem")
        .config("google.cloud.auth.service.account.enable", "true")
        .config("spark.hadoop.google.cloud.auth.service.account.enable", "true")
        .config("spark.hadoop.google.cloud.auth.service.account.json.keyfile", gcs_key_path)
        .getOrCreate()
)
spark
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/60752c90-f95a-47b4-bcf7-762089ed8575.png)

SpakSession を定義済みの場合には、`ランタイムを接続解除して削除`を選択して最初からやり直してください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/4e5e6d46-dc3d-4ad7-ad82-6d7484b34bcd.png)

### Apache Iceberg テーブルを作成

```python
spark.sql(f"""
CREATE OR REPLACE TABLE default.first_table(
  id    INT,
  name  STRING,
  status STRING
)
USING iceberg;
""")
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d175d7d8-e683-4cba-8b19-fc10c0f97f39.png)

GCS 上にテーブルのディレクトリが作成されていることを確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0d4ea191-35ac-4cb4-8b4b-863d10cbee02.png)

### テーブルにデータを挿入

```python
spark.sql("""
INSERT INTO
    default.first_table (id, name, status)
VALUES
    (1, 'Alice', 'active'),
    (2, 'Bob', 'active'),
    (3, 'Charlie', 'active'),
    (4, 'Daisy', 'active'),
    (5, 'Ethan', 'active');
""").show()
```

GCS 上にテーブルのディレクトリ内に parquet ファイルが書き込まれていることを確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f62a9ea1-ace7-408a-b696-e2a838251200.png)

### テーブルのデータを取得

```python
spark.table("default.first_table").show()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b6ea79d7-c21d-476a-b425-0f1ba221e1e9.png)