【実験】Microsoft Fabric Spark で Iceberg テーブルを生成し、Power BI や Databricks でクエリしてみる

## はじめに

[Databricks にて Hadoop Catalog の Apache Iceberg をさわってみた](https://qiita.com/manabian/items/06399340c91cd90661b5) を参考に Fabric の Spark ノートブックで Iceberg テーブルを生成してみます。

## 環境セットアップ


### レイクハウスの作成と abfss パスの特定

レイクハウスと iceberg 用のディレクトリを作成し、abfss パスを取得します。
このディレクトリ内に iceberg テーブルを生成することになります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/ac9babf6-aada-4da4-b9db-0b5ab8f6b9b7.png)


### Spark環境の構成

spark ノートブック用の環境を作成。ランタイムバージョンは 1.3( spark 3.5) です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/cb759c27-9abf-4d2d-ba18-04f8367ea0a3.png)

[Iceberg-spark-runtime](https://mvnrepository.com/artifact/org.apache.iceberg/iceberg-spark-runtime-3.5)のjar ファイルをダウンロードして、環境にアップします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/c2dc0687-90a5-4294-a8b8-aa46ca17f533.png)


次に、 https://iceberg.apache.org/docs/latest/spark-configuration/#catalog-configuration にしたがって

>Iceberg also supports a directory-based catalog in HDFS that can be configured using type=hadoop:
>spark.sql.catalog.hadoop_prod = org.apache.iceberg.spark.SparkCatalog
>spark.sql.catalog.hadoop_prod.type = hadoop
>spark.sql.catalog.hadoop_prod.warehouse = hdfs://nn:8020/warehouse/path

と、必要に応じて [SQL Extensions](https://iceberg.apache.org/docs/latest/spark-configuration/#sql-extensions)

> spark.sql.extensions = org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions 

も設定します。

spark.sql.catalog.hadoop_prod.warehouse には最初に取得したabfss パスを設定してください。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/c1b30a80-bd4b-4451-ad42-d174f01b888c.png)

:::note info

spark 上で iceberg を操作する分には Files/・・・などの相対パスでも可能ですが、[Iceberg ショートカット](https://learn.microsoft.com/ja-jp/fabric/onelake/onelake-iceberg-tables) では絶対パスでのファイル参照でメタデータが構成される必要があるため、abfss パスを使用します。

:::

spark 環境を発行します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/e7b273c7-2fcc-43d8-8514-3ea7d36ee447.png)


## テーブル操作

### テーブル作成とデータ登録

対象のレイクハウスにアタッチされたノートブックを作成し、

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/40cbf092-7480-4877-be96-be2349a11ed3.png)

環境を先ほど作成したものに設定します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/f534b5ad-7672-4798-a97f-2ecf38e2ff3e.png)

#### pyspark での既定のカタログへの書き込み

spark でテーブルを作成してみます。

```python:pyspark

from pyspark.sql.types import *

table_name = "people"

records = [
   (1, 'John', 25, 'NYC', '2023-09-28 00:00:00'),
   (2, 'Emily', 30, 'SFO', '2023-09-28 00:00:00'),
   (3, 'Michael', 35, 'ORD', '2023-09-28 00:00:00'),
   (4, 'Andrew', 40, 'NYC', '2023-10-28 00:00:00'),
   (5, 'Bob', 28, 'SEA', '2023-09-23 00:00:00'),
   (6, 'Charlie', 31, 'DFW', '2023-08-29 00:00:00')
]

schema = StructType([
   StructField("id", IntegerType(), True),
   StructField("name", StringType(), True),
   StructField("age", IntegerType(), True),
   StructField("city", StringType(), True),
   StructField("create_ts", StringType(), True)
])

df = spark.createDataFrame(records, schema)

(
   df.write
   .format("iceberg")
   .saveAsTable(table_name)
)

```

iceberg フォーマットのディレクトリが生成されました。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/0684f789-4604-4238-b8b5-ab86d846eae8.png)

テーブルは既定では、<レイクハウスの名称>のNamespaceに作成されることがわかります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/19e9a97b-d798-4dea-a8c1-2c8f9c4c0531.png)


#### SQL での書き込み

Namespace の生成をします。

```sql:sql

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

```

Namespace を作成すると、ディレクトリが生成されていることがわかります。このあたりはカタログをどのように構成したかで変わるようです。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/d3f3339a-7c8d-4be6-adde-eac324632bc4.png)

テーブルも作成します。

```sql:sql

CREATE TABLE iceberg_demo.sample (
    id bigint NOT NULL COMMENT 'unique id',
    data string)
USING iceberg;

```

```sql:sql

INSERT INTO iceberg_demo.sample (id, data) VALUES
    (1, 'first entry'),
    (2, 'second entry'),
    (3, 'third entry');
    
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/ed0c79ae-910b-4781-9a3f-a4eeb4c2a771.png)

指定したNamespace のディレクトリにデータが生成されました。

### Spark SQL での確認

SQL アクセスを確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/18af2b54-760e-487a-9659-fdfd69e0849b.png)

### pyspark によるフォーマット直接アクセスによる確認

相対パスを使用して、カタログを使用せず、ファイルメタデータだけでアクセスします。


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/10de9baf-5de8-4ee2-9c07-c5ddccd48b42.png)

```python:pyspark

df = spark.read.format("iceberg").load("Files/hadoop/iceberg_demo/sample")
display(df)

```


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/9037e773-ee0d-4f87-8c25-d10a7fa2e8e6.png)


### 過去バージョンへのクエリ

データの追加を行い、バージョンの履歴を確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/ff463052-0076-46ba-b956-5c1e4e369ed6.png)


```sql:sparksql

SELECT snapshot_id, committed_at FROM iceberg_demo.sample.snapshots

```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/ff6ae25a-eea8-42c2-b657-dddacc57801f.png)


```sql:sparksql


SELECT * FROM iceberg_demo.sample FOR SYSTEM_VERSION AS OF 1621798013750238808;

```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/ad688f1b-22f9-45d7-a472-2feacb446ba1.png)


過去バージョンも取得可能できています。



## Delta Lake 相互運用

OneLake 上に Iceberg がある場合、Iceberg ショートカットを使用することで Delta Lake として読み取ることができます。

### Iceberg ショートカットの生成

Tables にてショートカットを作成

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/2f2df550-5422-4b21-8d2e-8b0b92af92ba.png)

対象のテーブルディレクトリを選択します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/c95b549c-9fc1-47a1-aa00-0e0c18c23c78.png)

iceberg データを含むディレクトリをショートカットするとTables内で_delta_log が生成されます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/a73c6b14-993b-4050-a52e-2c24483d7343.png)



### Power BI での確認

Power BI でデータの確認をしてみます。

セマンティックモデルを作成します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/cb3a30e2-26ce-458e-8b41-cb9c081f7255.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/93139b03-3957-40de-b0d2-047ae05691c7.png)

レポートで表示を確認します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/c3633952-6c65-461e-bc5f-866662b88d89.png)

### Databricks でのアクセス

Tables フォルダでは、delta_log が生成されているので、Databricksから参照することができます。

パススルークラスターでアクセスするのが楽です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/22d27ed9-7b6a-4075-b3bc-8378b7e21339.png)

:::note warn

このクラスター構成は現在非推奨です。詳しくは [Azure Databricks から OneLake 上のデータにアクセスする方法 2024/12 版](https://qiita.com/ryoma-nagata/items/39fd52ab81015e3c9527)　で

:::

Tablesのabfss パスを取得します。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/d4baef52-00c6-45fa-b862-38e58ed5726d.png)


Databricks 上で以下実行していきます。

```python:pyspark

table_path = "abfss://<取得したパス>"

```



```python:pyspark

# SQL APIでアクセス

sql_df = spark.sql(f"select * from delta.`{table_path}`")
display(sql_df)

```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/8fb4cd82-827c-467f-941e-9d11b2fcc18d.png)


```python:pyspark

# pyspark APIでアクセス

pyspark_df = spark.read.format("delta").load(table_path)
display(pyspark_df)

```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/085b46eb-49a6-42f3-98e5-0ff0b78c2e77.png)

また、タイムトラベルも機能することがわかります。(対象テーブルは sample に変えています)

```python:pyspark

sql_df = spark.sql(f"select * from delta.`{table_path}` version as of 0 ")
display(sql_df)

```


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/84e0ed21-a48a-4bbb-ba0c-3eedecb565e0.png)

```python:pyspark

sql_df = spark.sql(f"select * from delta.`{table_path}` version as of 1 ")
display(sql_df)

```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/281819/17b8725c-347b-4370-a7eb-8e6f35ee1392.png)