Spark から Hadoop Catalog の Apache Iceberg の version-hint.text の要否検証

## 概要

Spark から Hadoop Catalog の Apache Iceberg で処理する際に生成される`version-hint.text`の必要性に疑問があったため検証しました。 Spark においては、`version-hint.text`の内容に基づかずに、最新のバージョンのデータを取得するような動作となりました。検証コードと結果を共有します。

## 事前準備

### Google Colab 上にノートブックを作成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/2f45c986-d9fb-4e9d-88e7-ea8d5ccf38c5.png)

### Spark のバージョンを確認

```bash
!pyspark --version
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/cec14e8d-07bf-4db4-b6f7-4a9b79ca3870.png)

### SparkSession を定義

```python
from pyspark.sql import SparkSession

spark = (SparkSession.builder
    .appName("IcebergExample")
    .config("spark.jars.packages", "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.0")
    .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", "/iceberg")
    .getOrCreate()
)
spark
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3a92b742-da14-4d2e-98c9-32b1a702910f.png)

### Namespace を作成

```python
spark.sql("CREATE NAMESPACE IF NOT EXISTS  iceberf_gc ;")
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/35f95593-3445-43b1-b4a9-86b0cb6802ec.png)


### テーブルの作成とデータ挿入

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

```python
spark.sql("""
INSERT INTO iceberf_gc.first_table 
SELECT 1, 'a'
;""")

spark.sql("""
INSERT INTO iceberf_gc.first_table 
SELECT 2, 'b'
;""")

spark.sql("""
INSERT INTO iceberf_gc.first_table 
SELECT 3, 'c'
;""")
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/f1607785-798e-435e-bae6-18f5c892f7bd.png)

### テーブルのデータの確認

```python
display(
    spark.read
    .table("iceberf_gc.first_table")
    .orderBy("id")
    .toPandas()
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/00c316e9-e6e6-4f0b-b37c-56ef0ba10ce9.png)

### テーブルの`metadata`ディレクトリを確認

```bash
!ls /iceberg/iceberf_gc/first_table/metadata
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/84d47e80-df17-44c1-a49e-26fd911a14d8.png)


### version-hint.text の内容を確認

```python
!cat /iceberg/iceberf_gc/first_table/metadata/version-hint.text
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/0318a474-909e-462a-8b9a-04930fea995d.png)

## version-hint.text の検証

### version-hint.text の内容を`2`に修正

`version-hint.text`ファイルの内容を`2`に上書きしてクエリを実行したところ、最新のテーブルのデータを取得したことを確認。

```bash
!echo "2" > /iceberg/iceberf_gc/first_table/metadata/version-hint.text
!cat /iceberg/iceberf_gc/first_table/metadata/version-hint.text
```

```python
from pyspark.sql import SparkSession

if spark:
    spark.stop()
    print("SparkSession を停止しました。")

spark = (SparkSession.builder
    .appName("IcebergExample")
    .config("spark.jars.packages", "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.0")
    .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", "/iceberg")
    .getOrCreate()
)
spark
```

```python
display(
    spark.read
    .table("iceberf_gc.first_table")
    .orderBy("id")
    .toPandas()
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ea336022-1dcf-444b-b677-61e0c818d00d.png)


### version-hint.text を削除

`version-hint.text`ファイルを削除後にクエリを実行したところ、最新のテーブルのデータを取得したことを確認。

```bash
!rm /iceberg/iceberf_gc/first_table/metadata/version-hint.text
!ls /iceberg/iceberf_gc/first_table/metadata
```

```python
from pyspark.sql import SparkSession

if spark:
    spark.stop()
    print("SparkSession を停止しました。")

spark = (SparkSession.builder
    .appName("IcebergExample")
    .config("spark.jars.packages", "org.apache.iceberg:iceberg-spark-runtime-3.5_2.12:1.8.0")
    .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", "/iceberg")
    .getOrCreate()
)
spark
```

```python
display(
    spark.read
    .table("iceberf_gc.first_table")
    .orderBy("id")
    .toPandas()
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1a6b84f7-15c7-416b-bbcd-95280521f172.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d2561ad4-f521-49fb-bace-2edafb79dc7b.png)