Databricks にて Apache Iceberg のテーブルディレクトリからデータを取得する方法の検証

## 概要

Databricks にて Hadoop Catalog の Apache Iceberg のテーブルディレクトリからデータを取得する方法の検証結果を共有します。いずれかの Databricks Runtime のバージョンで動いたものを動作確認できた方法として紹介します。エラーとなった方法や Databricks Runtime のバージョンごとの動作結果も備忘録として残しています。

## 事前準備

### データ書き込み用クラスターの準備

以下の設定を行ったクラスターを作成します。

- Databricks Runtime のバージョンを 12.2LTS 以前にすること
- アクセスモードを分離なし共有（Uniyt Catalog 利用不可のモード」）にすること
- Java 8 に対応したライブラリ（Iceberg 1.6.1 以前のライブラリ）を利用すること
- Spark 構成に Iceberg 関連の設定を実施

Spark 構成にて下記の設定を実施します。

```conf
spark.sql.catalog.spark_catalog = org.apache.iceberg.spark.SparkCatalog
spark.sql.catalog.spark_catalog.type = hadoop
spark.sql.catalog.spark_catalog.warehouse = dbfs:/Iceberg
spark.sql.extensions = org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
```

### Namespace を作成

```sql
%sql
-- データベース(Namespace)を作成する
CREATE NAMESPACE IF NOT EXISTS iceberg_read;
```

```python
wh_dir = spark.conf.get("spark.sql.catalog.spark_catalog.warehouse")
display(dbutils.fs.ls(wh_dir))
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/673d15df-b63d-4f7b-9f40-fea0cfe77ae1.png)

### テーブルの準備

```sql
%sql
CREATE OR REPLACE TABLE iceberg_read.first_table(
    current_ts TIMESTAMP
)
USING iceberg;
```

```sql
%sql
INSERT INTO iceberg_read.first_table
SELECT current_timestamp()
;
SELECT
  *
  FROM
     iceberg_read.first_table;
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b0134b80-71e4-461d-9968-6018929213f3.png)

### 読み込み用のクラスター作成

検証に利用する Databricks Runtime （12.2、13.3、14.3、 16.2）のバージョンごとに下記の設定を実施したクラスターを作成

- アクセスモードを分離なし共有にすること
- Iceberg 関連のライブラリをインストール

## 動作確認できた方法

### テーブルのディレクトリを指定する方法

Databricks Runtime 12.2 にて下記のコードで動作を確認。

```python
table_dir = "dbfs:/Iceberg/iceberg_read/first_table"
df = spark.read.format("iceberg").load(table_dir)
df.printSchema()
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fc28abc6-4521-4b99-9345-878d461fc86c.png)

Databricks Runtime 13.3 、および、Databricks Runtime 16.2 では`source`オプションを下記のように変更して動作することを確認。

```python
table_dir = "dbfs:/Iceberg/iceberg_read/first_table"
df = spark.read.format("org.apache.iceberg.spark.source.IcebergSource").load(table_dir)
df.printSchema()
df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6eea3f3a-de2c-49b8-82f8-38b35001112c.png)


### metadata file のパスを指定する方法

Databricks Runtime 12.2 にて下記のコードで動作を確認。

```python
table_dir = "dbfs:/Iceberg/iceberg_read/first_table"
v1_medata_path = f"{table_dir}/metadata/v1.metadata.json"
df_v1 = spark.read.format("iceberg").load(v1_medata_path)
df_v1.printSchema()
df_v1.display()

table_dir = "dbfs:/Iceberg/iceberg_read/first_table"
v2_medata_path = f"{table_dir}/metadata/v1.metadata.json"
df_v2 = spark.read.format("iceberg").load(v2_medata_path)
df_v2.printSchema()
df_v2.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/40498b7a-951c-41b1-858f-33385a67987f.png)

Databricks Runtime 13.3 では`source`オプションを下記のように変更して動作することを確認。

```python
table_dir = "dbfs:/Iceberg/iceberg_read/first_table"
v1_medata_path = f"{table_dir}/metadata/v1.metadata.json"
df_v1 = spark.read.format("org.apache.iceberg.spark.source.IcebergSource").load(v1_medata_path)
df_v1.printSchema()
df_v1.display()

table_dir = "dbfs:/Iceberg/iceberg_read/first_table"
v2_medata_path = f"{table_dir}/metadata/v2.metadata.json"
df_v2 = spark.read.format("org.apache.iceberg.spark.source.IcebergSource").load(v2_medata_path)
df_v2.printSchema()
df_v2.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ae5c8f57-cf96-406d-a1b0-12d836c444cd.png)

## 実行確認できていない方法

### SQL にて`iceberg`によりディレクトリを指定した方法

```sql
%sql
SELECT * FROM iceberg.`dbfs:/Iceberg/iceberg_read/first_table`;
```

> AnalysisException: Table iceberg.dbfs:/Iceberg/iceberg_read/first_table does not support batch scan.;
> Project
> +- SubqueryAlias spark_catalog.iceberg.`dbfs:/Iceberg/iceberg_read/first_table`
>    +- RelationV2[] spark_catalog.iceberg.`dbfs:/Iceberg/iceberg_read/first_table` iceberg.dbfs:/Iceberg/iceberg_read/first_table

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/607dc64f-68bb-49fa-9664-40b489adb38c.png)

### Databricks Runtime 13.3 上で PySpark にて`format`オプションに`iceberg`を指定する方法

```python
table_dir = "dbfs:/Iceberg/iceberg_read/first_table"
df = spark.read.format("iceberg").load(table_dir)
df.printSchema()
df.display()
```

> AnalysisException: Multiple sources found for iceberg (com.databricks.sql.transaction.tahoe.uniform.sources.IcebergBrowseOnlyDataSource, org.apache.iceberg.spark.source.IcebergSource), please specify the fully qualified class name.

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/bf12361e-a55f-4eba-81c8-f234ecb22164.png)

### Databricks Runtime 14.3上で PySpark にて`format`オプションに`org.apache.iceberg.spark.source.IcebergSource`を指定する方法

```python
df = spark.read.format("org.apache.iceberg.spark.source.IcebergSource").load(table_dir)
df.printSchema()
df.display()
```

> UnsupportedOperationException: Byte-buffer read unsupported by com.databricks.common.filesystem.LokiAbfsInputStream
> org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 4 times, most recent failure: Lost task 0.3 in stage 0.0 (TID 3) (10.139.64.13 executor driver): java.lang.UnsupportedOperationException: Byte-buffer read unsupported by com.databricks.common.filesystem.LokiAbfsInputStream

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/12b94717-2d39-40e0-a798-2532e1710b1b.png)

14.3 で上記エラーが発生することは Databricks による仕様であるため、 Databricks Runtime 16 以降を利用することで正常に動作することを確認。下記記事にて詳細を記述。

- [Databricks にて Iceberg 参照する際の Byte-buffer read unsupported by com.databricks.common.filesystem.LokiAbfsInputStream への対応方法 #Spark - Qiita](https://qiita.com/manabian/items/3051432f5d32b55fa1ab)