Delta Lake におけるマージ操作でのスキーマ進化と運用メトリクス

Databricksの最新ebook『O'Reilly Delta Lake: Running On All Cylinders』のプレビューをいち早く入手して、Delta Lakeの利用開始に必要なステップバイステップのガイダンスを手に入れましょう。

以下の手順を再現するには、こちらのノートブックをお試しください。

先日、Delta Lake 0.6.0のリリースを発表しました。このバージョンでは、マージ操作におけるスキーマ進化とパフォーマンスの向上、およびテーブル履歴における運用メトリクスが導入されています。このリリースにおける主な機能は以下の通りです。

• マージ操作におけるスキーマ進化のサポート（#170） - マージ操作により、テーブルのスキーマを自動的に進化させることができます。これは、データのスキーマが時間とともに変化する場合に、変更データをテーブルにupsert（更新または挿入）したいシナリオで役立ちます。upsertを実行する前にスキーマの変更を検出し適用する代わりに、マージはスキーマを進化させながら変更をupsertできます。
• 自動パーティショニングによるマージパフォーマンスの向上（#349） - パーティションテーブルにマージする場合、テーブルに書き込む前にパーティション列でデータを自動的に再パーティショニングすることを選択できます。パーティションテーブルでのマージ操作が、多数の小さなファイルを生成するために遅くなる場合（#345）、自動再パーティショニング（spark.delta.merge.repartitionBeforeWrite）を有効にするとパフォーマンスが向上する可能性があります。
• INSERT句がない場合でもパフォーマンスが向上（#342） - INSERT句がないマージ操作では、パフォーマンスが向上します。
• DESCRIBE HISTORYでの操作メトリクス（#312） - Deltaテーブルに対するすべての書き込み、更新、削除の操作メトリクス（例：変更されたファイル数と行数）をテーブル履歴で確認できるようになりました。
• 任意のファイルシステムからのDeltaテーブル読み込みサポート（#347） - Hadoop FileSystem実装を持つ任意のストレージシステム上のDeltaテーブルを読み込めるようになりました。ただし、Deltaテーブルへの書き込みには、ストレージシステムに必要な保証を提供するLogStore実装の設定が必要です。

マージ操作におけるスキーマ進化

Delta Lakeの以前のリリースでも述べたように、Delta Lakeはマージ操作を実行して、単一のトランザクションでの挿入/更新/削除操作を簡素化し、さらにスキーマの強制と進化を可能にします（詳細については、こちらのテクニカルトークでも説明しています）。  Delta Lake 0.6.0のリリースにより、マージ操作内でスキーマを進化させることができるようになりました。

これを例で示しましょう。元のコードサンプルはこちらのノートブックで見つけることができます。  まず、ジョンズ・ホプキンス大学CSSEが提供する2019 Novel Coronavirus COVID-19 (2019-nCoV) データリポジトリの小規模なサブセットを使用します。これは/databricks-datasetsで利用可能にしています。  これは、世界中のCOVID-19の症例数を把握するために、研究者やアナリストが一般的に使用するデータセットです。このデータの問題点の1つは、スキーマが時間とともに変化することです。

例えば、3月1日から3月21日までのCOVID-19症例を表すファイル（2020年4月30日現在）のスキーマは以下のようになっています。

しかし、3月22日以降のファイル（2020年4月30日現在）には、FIPS、Admin2、Active、Combined_Keyなどの追加列がありました。

サンプルコードでは、意味的に同じであるため、一部の列名を変更しました（例：Long_ -> Longitude、Province/State -> Province_Stateなど）。  テーブルスキーマを進化させる代わりに、単に列名を変更しました。

スキーマを結合することだけが懸念事項であれば、DataFrame.write()の「mergeSchema」オプションを使用して、Delta Lakeのスキーマ進化機能を使用できます。これは次のステートメントで示されています。

しかし、既存の値を更新し、同時にスキーマをマージする必要がある場合はどうなるでしょうか？ Delta Lake 0.6.0では、マージ操作のためのスキーマ進化を使用してこれを達成できます。これを視覚化するために、まず1行のold_dataを確認することから始めましょう。

次に、new_dataのスキーマに従う更新エントリをシミュレートしましょう。

そして、simulated_updateとnew_dataを合計40行で結合します。

自動スキーマ進化のために環境を設定する次のパラメータを設定します。

これで、値（2020/3/21）を更新し、新しいスキーマをマージするための単一のトランザクション操作を次のステートメントで実行できます。

次のステートメントでDelta Lakeテーブルを確認しましょう。

運用指標

Spark UIでDelta Lakeテーブル履歴（operationMetrics列）を確認することで、運用指標をさらに詳しく調べることができます。以下のステートメントを実行してください。

以下は、前のコマンドの省略された出力です。

テーブルには2つのバージョンがあり、1つは古いスキーマ用、もう1つは新しいスキーマ用です。以下の運用指標を確認すると、39行が挿入され、1行が更新されたことがわかります。

これらの運用指標の詳細については、Spark UIのSQLタブをご覧ください。

アニメーションGIFは、確認のためのSpark UIの主要コンポーネントを示しています。

1. 新しいスキーマ（2020年4月11日）の1つのファイルからの初期39行が、初期のnew_data DataFrameを作成しました。
2. new_data DataFrameとunionされるシミュレートされた更新行が1行生成されました。
3. 古いスキーマ（2020年3月21日）の1つのファイルからの1行が、old_data DataFrameを作成しました。
4. Delta Lakeテーブルに永続化するために、2つのDataFrameを結合するために使用されたSortMergeJoin。

これらの運用指標の解釈方法についてさらに詳しく知りたい場合は、Diving into Delta Lake Part 3: How do DELETE, UPDATE, and MERGE work tech talkをご覧ください。

Delta Lake 0.6.0 を使い始める

Apache Spark 2.4.5（以降）インスタンス（DatabricksではDBR 6.6以降で試してください）で、上記のコードスニペットを使用してDelta Lakeを試してみてください。Delta Lakeは、データレイクが新しいものを作成する場合でも、既存のデータレイクを移行する場合でも、データレイクをより信頼性の高いものにします。詳細については、https://delta.io/を参照し、SlackおよびGoogle Groupを介してDelta Lakeコミュニティに参加してください。GitHubのmilestonesで、今後のリリースと計画されている機能すべてを追跡できます。無料アカウントでDatabricks上のManaged Delta Lakeを試すこともできます。

クレジット

Delta Lake 0.6.0の更新、ドキュメント変更、および貢献をしていただいた以下の貢献者に感謝いたします。Ali Afroozeh, Andrew Fogarty, Anurag870, Burak Yavuz, Erik LaBianca, Gengliang Wang, IonutBoicuAms, Jakub Orłowski, Jose Torres, KevinKarlBob, Michael Armbrust, Pranav Anand, Rahul Govind, Rahul Mahadev, Shixiong Zhu, Steve Suh, Tathagata Das, Timothy Zhang, Tom van Bussel, Wesley Hoffman, Xiao Li, chet, Eugene Koifman, Herman van Hovell, hongdd, lswyyy, lys0716, Mahmoud Mahdi, Maryann Xue

(このブログ記事はAI翻訳ツールを使用して翻訳されています) 原文記事