Databricks ユーザーが Fabric を使う際にひっかかりがちな Delta Lake テーブル機能の互換性について

はじめに

OneLakeの実体で解説したように、Fabric は構造化フォーマットに Delta Lake を採用しています。
これは Delta Lake を開発し、主軸としている Databricks と強い相互運用性を生むことを意味しています。
双方のプラットフォームをうまく組み合わせて使うことにより、プロデータエンジニアからパワーユーザーに至るまで、すべてのデータ分析ステークホルダーが迅速かつ効率的に作業を進めることができます。

引用：https://qiita.com/yangjiayi/items/e264451e5c83c6ac90a7

本記事では、Fabric と Databricks の相互運用性を支える Delta Lake のテーブルプロトコル仕様について解説し、実際にFabric で Delta Lake 機能を使用した際に発生する問題とその解決策を紹介します。

Delta Lake テーブルプロトコル仕様とは

Delta Lake は、データのバージョン管理や変更履歴の保持、ACID トランザクションなど、データレイクでよく求められる機能を提供します。

テーブルの提供する機能は、テーブルプロトコル仕様により管理されており、これによりデータアクセスの安全性を保ちながら異なるシステムやバージョン間での相互運用が可能になります。

以下の機能が有効になり、また対応する Delta Lake バージョンに後方互換性を提供しています。
つまり、旧バージョンから提供される機能は最新バージョンでも有効化ができ、動作が保証されます。

引用：Delta Lake 機能

一方で、Delta Lake テーブルで最新の機能が有効化されている場合に、テーブルにアクセスするクライアントがその機能をサポートしていない場合、データアクセスが不可となる場合があります。

テーブルのプロトコルでサポート対象としてリストされている機能の処理方法をアプリケーションが知らない場合、そのアプリケーションはそのテーブルの読み取りまたは書き込みができません。

引用：https://docs.delta.io/latest/versioning.html#-features-by-protocol-version

Table Features

Table Features は 機能フラグによって個別に機能を有効化する Delta Lake の現仕様のことです。

テーブル機能はDelta Lake 2.3.0以前はバージョン番号でバンドル式に提供されていましたが、2.3.0以降はテーブルプロパティによる機能フラグで管理されています。

列マッピング機能までは、有効化した機能に伴い、それ以前に提供されたテーブル機能も同時に有効化されますが、以降のテーブル機能については個別に選択して有効化が可能です。

参考：Introducing Delta Lake Table Features

Protocol Version

Table Features　以前のリーダーまたはライターバージョンにバンドルされた管理では、プロトコルバージョンをアップグレードするとテーブル機能が有効化されます。

こちらも後方互換性を提供していますが、テーブル機能を個別には有効化できません。
現在はこの管理方式を使用しているのは Delta Lake 2.3.0より前のバージョンであるため、注意点は少ないのですが、プロトコルバージョン自体は生きており、 テーブルの機能を有効化すると自動的にアップグレードされ、ダウングレードはできない という点については留意する必要があります。

引用：プロトコルバージョン別の機能

Databricks と Fabric で Delta Lake テーブル機能を利用する注意点

Databricks や、 Fabric の分析エンジンがサポートする Delta Lake 機能はそれぞれ異なる場合があることを留意する必要があります。

一方が最新の Delta Lake 機能を使用し、一方がそれに対応していない機能を使用すると、テーブルプロトコル仕様が安全にデータを管理するためにデータへのアクセスが提供しない状況が発生してしまいます。

Databricks の Delta Lake テーブルプロトコルの互換性対応

Databricks では、Delta Lake の各機能はランタイムバージョンに対応しており、特定のランタイムバージョン以降でのみ利用できる機能がありますが、ランタイムバージョンを上げることはあっても下げることはほぼないので、この互換性の問題は遭遇することがほぼないかと思います。

多くの Azure Databricks の最適化では、テーブルで Delta Lake 機能を有効にする必要があります。 Delta Lake 機能は常に下位互換性があるため、低次の Databricks Runtime バージョンで書き込まれたテーブルは、高次の Databricks Runtime バージョンで常に読み取り/書き込みできます。 一部の機能を有効にすると、Databricks Runtime のバージョンが低い状態で実行されているワークロードへの上位互換性がなくなります。 上位互換性を損なう機能については、アップグレードされたテーブルを参照するすべてのワークロードを更新して、準拠している Databricks Runtime バージョンを使用する必要があります。

引用：Azure Databricks で Delta Lake 機能の互換性を管理する方法

Databricks では OSS Delta Lake に実装されていないエッジ機能も含まれており、より快適に Delta Lake を使用できる環境となっています

Microsoft Fabric の Delta Lake テーブルプロトコルの互換性対応

Microsoft Fabric では、異なるエンジンが Delta Lake という標準フォーマットを共有して動作することで、様々なユースケースにデータコピーなしに対応できることが強みとなっています。

ただし、分析機能により異なるエンジンが動作し、そのするため、それぞれがサポートする DeltaLake テーブル機能も異なることに注意が必要です。

すべての Fabric のエクスペリエンスが Delta Lake の機能と Fabric の機能に合わせて調整されます。Delta Lake テーブルにのみ書き込むことができるエクスペリエンスもあれば、Delta Lake テーブルから読み取ることができるエクスペリエンスもあります。

引用：Delta Lake テーブル形式の相互運用性

Microsoft Fabric は V-order と呼ばれるデータ書き込み時に動作する最適化機能が既定で機能します。
データの並びが最適化されるためクエリを高速化するメリットがありますが、独自の技術であり、Fabric の エッジ機能と考えられます。

ただし、V-order 書き込みされたからといって、Databrikcs から読み書きができないということにはなりませんのでご安心ください（Fabric以外からも書き込めるが、データの並びがv-order と同一の最適化状態にならないだけ）

大まかな考え方としては、Fabric Spark ランタイムは Delta Lake のバージョンも明示されるためはまりポイントは少ないが、他の機能では対応する Delta Lake のバージョンが明示されないため注意が必要という点を抑えると良いかと思います。

なお、ウェアハウスなど、一部の機能でテーブルを作成や更新した場合には、Delta 閲覧者/作成者のバージョンと既定のテーブル機能 に記載のように基本機能だけでなく、特定のテーブル機能が有効化された状態となる点にも注意が必要です。

Fabric Spark Runtime

Delta Lake をもっとも柔軟に使用できるのが Spark Runtime ですが、動作バージョンには注意が必要です。

以下の表の通り、 Runtime 1.3 から Delta Lake 3.2 が提供されているため、例えば、最新の Delta Lake 3.1 から提供されたリキッドクラスタリングを有効化にすると、Delta Lake 2.4.0 で動作するRuntime 1.2 からは書きこみができなくなります。

引用:Fabric の Apache Spark Runtime

実例

互換性の問題がどのように起きるか実例を見ていきましょう。

リキッドクラスタリングを有効化する

Databricks ユーザーとしては積極的に使っている人も多いかと思われるリキッドクラスタリングです。
これを Fabric で使用するとどうなるでしょうか。

1. 通常のテーブルを作成し、テーブルプロトコルを確認します。

   sparksql

   %%sql
   CREATE TABLE sales_created_by_notebook_basic (
       SalesOrderNumber string,
       SalesOrderLineNumber string,
       OrderDate string,
       CustomerName string,
       EmailAddress string,
       Item string,
       Quantity string ,
       UnitPrice string ,
       TaxAmount string
   )
   USING DELTA;
   
   SHOW TBLPROPERTIES sales_created_by_notebook_basic;
   

   

2. このテーブルはデータパイプラインから書き込みが可能です。
   
   

3. 次にリキッドクラスタリングを有効化したテーブルを作成し、テーブルプロトコルを確認します

   sparksql

   %%sql
   CREATE TABLE sales_created_by_notebook (
       SalesOrderNumber string,
       SalesOrderLineNumber string,
       OrderDate string,
       CustomerName string,
       EmailAddress string,
       Item string,
       Quantity string ,
       UnitPrice string ,
       TaxAmount string
   )
   USING DELTA
   CLUSTER BY (OrderDate);
   
   SHOW TBLPROPERTIES sales_created_by_notebook;
   

   

   domainMetadata なる機能も有効化されていますが、これはリキッドクラスタリングの要件に含まれた機能です。
   参考：https://docs.delta.io/latest/delta-clustering.html

4. テーブル機能の状態はdelta_log からも確認が可能です。
   

5. Spark からデータの書き込みが可能です

   sparksql

   %%sql
   
   SET `spark.microsoft.delta.optimizeWrite.enabled` = true ; 
   INSERT INTO sales_created_by_notebook
   SELECT * FROM sales_created_by_notebook_basic ; 
   SELECT COUNT(1) FROM sales_created_by_notebook ; 
       
   

   

6. しかしこのテーブルはデータパイプラインから書き込むことができません。
   
   

   Error messge詳細

   ErrorCode=DeltaNotSupportedAction,'Type=Microsoft.DataTransfer.Common.Shared.HybridDeliveryException,Message=The supported Delta actions in Delta log file are Metadata, Add, Remove, CDC, Transaction, Protocol and CommitInfo.,Source=Microsoft.DataTransfer.ClientLibrary,'

このことから、リキッドクラスタリングを有効化したテーブルにデータパイプラインは完全な互換性を提供していないということがわかります。

テーブルプロトコルバージョンがダウングレードをしようとしてしまう

不意にテーブルプロトコルバージョンがダウングレードされてしまう使い方があります。

1. まず、データパイプラインでレイクハウス上にテーブルを作成します。
   
   
   

2. 現在のテーブルプロトコルバージョンを確認します。
   

3. 対象テーブルのテーブルプロトコルバージョンをアップグレードします。 CHECK constraints　が有効となる WriteVersion = 3 の状態です

   pyspark

   %%sql
   
   ALTER TABLE sales_created_by_pipeline SET TBLPROPERTIES (
      'delta.minReaderVersion' = '1',
      'delta.minWriterVersion' = '3');
   
   SHOW TBLPROPERTIES sales_created_by_pipeline 
   

   

4. delta_log にてプロトコルバージョンがアップグレードされたことがわかります。

   アップグレード前
   

   

   アップグレード後
   

5. データパイプラインで作成したテーブルをFabric ノートブックで読み書きが可能です。

   pyspark

   df = spark.sql("SELECT * FROM LH_table_protocol.sales_created_by_pipeline")
   print(f"befor {df.count()}")
   
   df.write.mode("append").saveAsTable("LH_table_protocol.sales_created_by_pipeline")
   
   df_after = spark.sql("SELECT * FROM LH_table_protocol.sales_created_by_pipeline")
   
   print(f"after {df_after.count()}")
   

   

6. データパイプラインでも読み書き(追加)が引き続き可能です。
   

7. しかし、データパイプラインの動作を上書きにするとエラーとなります。
   

   エラーメッセージからはテーブルプロトコルに関するメッセージが出力されます。
   

   Error messge詳細

   Failure happened on 'destination' side. 'Type=System.InvalidOperationException,Message=Cannot downgrade the minimum writer version,Source=Microsoft.DataTransfer.ClientLibrary,'

これはデータパイプラインの作成するテーブルプロトコルは基本機能であるライターバージョンが2、リーダーバージョンが1であり、データパイプラインの上書き操作によりダウングレードを起こしてしまうことで起きています。

delta_log からはREPLACE_TABLEが実行されていることがわかります。

まとめ

Fabric と Databricks は共通のストレージフォーマットにより強い相互運用性を実現できる一方、それを活かすためには、Delta Lake のテーブルプロトコル仕様を理解することが重要というお話でした。

その他参考

• https://milescole.dev/integration/2024/03/22/Decoding-Delta-Lake-Compatibility-Between-Fabric-and-Databricks.html