Delta Lake におけるマージ操作でのスキーマ進化と運用メトリクス
Delta Lake 0.6.0では、マージとテーブル履歴の運用メトリクスにおけるスキーマ進化とパフォーマンスの向上が導入されました
<em>Databricksの新しいebook</em><a href="https://www.databricks.com/resources/ebook/delta-lake-running-oreilly?itm_data=operationsmetricsdl-blog-oreillydlupandrunning" target="_blank"><em>O'Reilly版</em></a><em>で、Delta Lakeの利用開始に必要なステップバイステップのガイダンスを入手しましょう。</em></p><blockquote><p>以下の手順を再現するには、この<a href="https://github.com/databricks/tech-talks/blob/master/samples/Schema%20Evolution%20in%20Merge%20Operations.ipynb" rel="noopener noreferrer" target="_blank">ノートブック</a>をお試しください。</p></blockquote><p>先日、Delta Lake 0.6.0をリリースしました。このバージョンでは、マージ操作におけるスキーマ進化とパフォーマンスの向上、およびテーブル履歴における運用メトリクスが導入されています。このリリースにおける主な機能は以下の通りです。</p><ul><li>マージ操作でのスキーマ進化のサポート（<a href="https://github.com/delta-io/delta/issues/170">#170</a>） - マージ操作でテーブルのスキーマを自動的に進化させることができるようになりました。これは、テーブルに変更データをupsert（更新または挿入）したい場合で、データのスキーマが時間とともに変化する場合に便利です。upsert前にスキーマ変更を検出して適用する代わりに、マージ操作でスキーマを同時に進化させ、変更をupsertできます。</li><li><strong>自動パーティショニングによるマージパフォーマンスの向上</strong>（<a href="https://github.com/delta-io/delta/issues/349">#349</a>） - パーティションテーブルにマージする場合、テーブルに書き込む前にパーティション列でデータを自動的に再パーティション化することを選択できます。パーティションテーブルでのマージ操作が、多数の小さなファイルを生成するために遅くなる場合（<a href="https://github.com/delta-io/delta/issues/345">#345</a>）、自動再パーティション化（spark.delta.merge.repartitionBeforeWrite）を有効にするとパフォーマンスが向上する可能性があります。</li><li><strong>INSERT句がない場合でもパフォーマンスが向上</strong>（<a href="https://github.com/delta-io/delta/issues/342">#342</a>） - INSERT句がないマージ操作では、パフォーマンスが向上します。</li><li><strong>DESCRIBE HISTORYでの操作メトリクス</strong>（<a href="https://github.com/delta-io/delta/issues/312">#312</a>） - Deltaテーブルの履歴で、すべての書き込み、更新、削除に対する操作メトリクス（例：変更されたファイル数と行数）を確認できるようになりました。</li><li><strong>任意のファイルシステムからのDeltaテーブルの読み取りサポート</strong>（<a href="https://github.com/delta-io/delta/issues/347">#347</a>） - Hadoop FileSystem実装を持つ任意のストレージシステム上のDeltaテーブルを読み取ることができるようになりました。ただし、Deltaテーブルへの書き込みには、ストレージシステムで必要な保証を提供するLogStore実装の設定が必要です。</li></ul><h2>マージ操作におけるスキーマ進化</h2><p>Delta Lakeの以前のリリースで述べたように、Delta Lakeには、<a href="https://www.databricks.com/blog/2019/10/03/simple-reliable-upserts-and-deletes-on-delta-lake-tables-using-python-apis.html" rel="noopener noreferrer" target="_blank">マージ操作を実行</a>して、<a href="https://www.youtube.com/watch?v=7ewmcdrylsA">挿入/更新/削除操作を単一のトランザクションで</a>簡素化する機能が含まれています。また、<a href="https://www.databricks.com/blog/2019/09/24/diving-into-delta-lake-schema-enforcement-evolution.html" rel="noopener noreferrer" target="_blank">スキーマの強制と進化</a>を行う機能も含まれています（詳細については、この<a href="https://www.youtube.com/watch?v=tjb10n5wVs8">技術トーク</a>でも確認できます）。 Delta Lake 0.6.0のリリースにより、マージ操作内でスキーマを進化させることができるようになりました。</p><p>これを実演するために、タイムリーな例を使用します。元のコードサンプルは<a href="https://github.com/databricks/tech-talks/blob/master/samples/Schema%20Evolution%20in%20Merge%20Operations.ipynb">このノートブック</a>で見つけることができます。 まず、ジョンズ・ホプキンスCSSEによる<a href="https://github.com/CSSEGISandData/COVID-19">2019 Novel Coronavirus COVID-19 (2019-nCoV) データリポジトリ</a>の小規模なサブセットを使用します。これは、<a href="https://docs.databricks.com/data/databricks-datasets.html">/databricks-datasets</a>で利用可能にしました。 これは、研究者やアナリストが世界中のCOVID-19の症例数に関する洞察を得るためによく使用するデータセットです。このデータの問題点の1つは、スキーマが時間とともに変化することです。</p><p>例えば、3月1日から3月21日までのCOVID-19症例を表すファイル（2020年4月30日現在）のスキーマは次のとおりです。</p><pre># Import old_data old_data = (spark.read.option("inferSchema", True).option("header", True)... .csv(/databricks-datasets/COVID/.../03-21-2020.csv)) old_data.printSchema() root |-- Province/State: string (nullable = true) |-- Country/Region: string (nullable = true) |-- Last Update: timestamp (nullable = true) |-- Confirmed: integer (nullable = true) |-- Deaths: integer (nullable = true) |-- Recovered: integer (nullable = true) |-- Latitude: double (nullable = true) |-- Longitude: double (nullable = true) </pre><p>しかし、3月22日以降のファイル（2020年4月30日現在）には、FIPS、Admin2、Active、Combined_Keyなどの追加列がありました。</p><pre>new_data = (spark.read.option("inferSchema", True).option("header", True)... .csv(/databricks-datasets/COVID/.../04-21-2020.csv)) new_data.printSchema() root |-- FIPS: integer (nullable = true) |-- Admin2: string (nullable = true) |-- Province_State: string (nullable = true) |-- Country_Region: string (nullable = true) |-- Last_Update: string (nullable = true) |-- Lat: double (nullable = true) |-- Long_: double (nullable = true) |-- Confirmed: integer (nullable = true) |-- Deaths: integer (nullable = true) |-- Recovered: integer (nullable = true) |-- Active: integer (nullable = true) |-- Combined_Key: string (nullable = true) </pre><p><em><i class="text-center">サンプルコードでは、一部の列名を変更しました（例：Long_ -> Longitude、Province/State -> Province_Stateなど）。これらは意味的には同じです。 テーブルスキーマを進化させる代わりに、単に列名を変更しました。</i></em></p><p>スキーマをマージすることだけが懸念事項であれば、<code>DataFrame.write()</code>の「mergeSchema」オプションを使用してDelta Lakeのスキーマ進化機能を使用できます。これは、次のステートメントで示されています。</p><pre>new_data.write.option("mergeSchema", "true").mode("append").save(path) </pre><p>しかし、既存の値を更新し、同時にスキーマをマージする必要がある場合はどうなるでしょうか？ Delta Lake 0.6.0では、<em>マージ操作のためのスキーマ進化</em>を使用してこれを実現できます。これを視覚化するために、まずold_data（<em>1行</em>）を確認することから始めましょう。</p><pre>old_data.select("process_date", "Province_State", "Country_Region", "Last_Update", "Confirmed").show() +------------+--------------+--------------+-------------------+---------+ |process_date|Province_State|Country_Region| Last_Update|Confirmed| +------------+--------------+--------------+-------------------+---------+ | 2020-03-21| Washington| US|2020-03-21 22:43:04| 1793| +------------+--------------+--------------+-------------------+---------+ </pre><p>次に、new_dataのスキーマに従う更新エントリをシミュレートします。</p><pre># Simulate an Updated Entry items = [(53, '', 'Washington', 'US', '2020-04-27T19:00:00', 47.4009, -121.4905, 1793, 94, 0, '', '', '2020-03-21', 2)] cols = ['FIPS', 'Admin2', 'Province_State', 'Country_Region', 'Last_Update', 'Latitude', 'Longitude', 'Confirmed', 'Deaths', 'Recovered', 'Active', 'Combined_Key', 'process_date', 'level'] simulated_update = spark.createDataFrame(items, cols) </pre><p>そして、simulated_updateとnew_dataを合わせて<em>40行</em>にします。</p><pre>new_data.select("process_date", "FIPS", "Province_State", "Country_Region", "Last_Update", "Confirmed").sort(col("FIPS")).show(5) +------------+-----+--------------+--------------+-------------------+---------+ |process_date| FIPS|Province_State|Country_Region| Last_Update|Confirmed| +------------+-----+--------------+--------------+-------------------+---------+ | 2020-03-21| 53| Washington| US|2020-04-27T19:00:00| 1793| | 2020-04-11|53001| Washington| US|2020-04-11 22:45:33| 30| | 2020-04-11|53003| Washington| US|2020-04-11 22:45:33| 4| | 2020-04-11|53005| Washington| US|2020-04-11 22:45:33| 244| | 2020-04-11|53007| Washington| US|2020-04-11 22:45:33| 53| +------------+-----+--------------+--------------+-------------------+---------+ </pre><p><em><a href="https://docs.delta.io/latest/delta-update.html#automatic-schema-evolution">自動スキーマ進化</a></em>のために環境を設定するパラメータを以下に示します。</p><pre># Enable automatic schema evolution spark.sql("SET spark.databricks.delta.schema.autoMerge.enabled = true") </pre><p>これで、値（2020/3/21から）を更新し、新しいスキーマをマージする単一のトランザクション操作を次のステートメントで実行できます。</p><pre>from delta.tables import * deltaTable = DeltaTable.forPath(spark, DELTA_PATH) # Schema Evolution with a Merge Operation deltaTable.alias("t").merge( new_data.alias("s"), "s.process_date = t.process_date AND s.province_state = t.province_state AND s.country_region = t.country_region AND s.level = t.level" ).whenMatchedUpdateAll( ).whenNotMatchedInsertAll( ).execute() </pre><p>次のステートメントでDelta Lakeテーブルを確認しましょう。</p><pre># Load the data spark.read.format("delta").load(DELTA_PATH) .select("process_date", "FIPS", "Province_State", "Country_Region", "Last_Update", "Confirmed", "Admin2") .sort(col("FIPS")) .show() +------------+-----+--------------+--------------+-------------------+---------+ |process_date| FIPS|Province_State|Country_Region| Last_Update|Confirmed|Admin| +------------+-----+--------------+--------------+-------------------+---------+-----+ | 2020-03-21| 53| Washington| US|2020-04-27T19:00:00| 1793| | | 2020-04-11|53001| Washington| US|2020-04-11 22:45:33| 30| Adams | | 2020-04-11|53003| Washington| US|2020-04-11 22:45:33| 4| Asotin | | 2020-04-11|53005| Washington| US|2020-04-11 22:45:33| 244| Benton | | 2020-04-11|53007| Washington| US|2020-04-11 22:45:33| 53| Chelan | +------------+-----+--------------+--------------+-------------------+---------+-----+ </pre><h2>運用指標</h2><p>Spark UI で Delta Lake テーブル履歴（operationMetrics 列）を確認することで、運用指標をさらに詳しく調べることができます。次のステートメントを実行してください。</p><pre>deltaTable.history().show() </pre><p>以下は、前のコマンドの省略された出力です。</p><pre>+-------+------+---------+--------------------+ |version|userId|operation| operationMetrics| +-------+------+---------+--------------------+ | 1|100802| MERGE|[numTargetRowsCop...| | 0|100802| WRITE|[numFiles -> 1, n...| +-------+------+---------+--------------------+ </pre><p>テーブルには 2 つのバージョンがあることに気づくでしょう。1 つは古いスキーマ用、もう 1 つは新しいスキーマ用です。運用指標を確認すると、39 行が挿入され、1 行が更新されたことがわかります。</p><pre>{ "numTargetRowsCopied":"0", "numTargetRowsDeleted":"0",{ "numTargetRowsCopied":"0", "numTargetRowsDeleted":"0", "numTargetFilesAdded":"3", "numTargetRowsInserted":"39", "numTargetRowsUpdated":"1", "numOutputRows":"40", "numSourceRows":"40", "numTargetFilesRemoved":"1" } "numTargetFilesAdded":"3", "numTargetRowsInserted":"39", "numTargetRowsUpdated":"1", "numOutputRows":"40", "numSourceRows":"40", "numTargetFilesRemoved":"1" } </pre><p>これらの運用指標の背後にある詳細については、Spark UI の SQL タブをご覧ください。</p><p style="max-width:450px;"><a href="https://www.databricks.com/wp-content/uploads/2020/05/blog-delta-lake-6-og3.gif" data-lightbox=" "><img class="aligncenter size-full wp-image-109752" style="width:500px;" src="https://www.databricks.com/wp-content/uploads/2020/05/blog-delta-lake-6-og3.gif" alt="An example of the operational metrics now available for review in the Spark UI through Delta Lake 0.6.0" height="410"></a></p><p>アニメーション GIF は、確認できる Spark UI の主要コンポーネントを示しています。</p><ol><li>新しいスキーマで 4/11/2020 の 1 つのファイルからの初期行 39 行（新しい new_data DataFrame を作成しました）。</li><li>新しい new_data DataFrame と結合される 1 つのシミュレートされた更新行が生成されました。</li><li>古いスキーマで 3/21/2020 の 1 つのファイルからの 1 行（古い old_data DataFrame を作成しました）。</li><li>Delta Lake テーブルに永続化するために 2 つの DataFrame を結合するために使用された SortMergeJoin。</li></ol><p>これらの運用指標の解釈方法についてさらに詳しく知りたい場合は、<a href="https://www.youtube.com/watch?v=7ewmcdrylsA">Diving into Delta Lake Part 3: How do DELETE, UPDATE, and MERGE work tech talk</a> を参照してください。</p><p><a class="lightbox-trigger" href="https://www.youtube.com/watch?v=7ewmcdrylsA" data-lightbox=" "><img class="aligncenter size-full wp-image-97884" src="https://www.databricks.com/wp-content/uploads/2020/05/blog-delta-lake-6-2.png" alt="ADiving into Delta Lake Part 3: How do DELETE, UPDATE, and MERGE work tech talk." height="630"></a></p>
ガートナー®: Databricks、クラウドデータベースのリーダー
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 のマイルストーンで、今後のリリースと計画された機能すべてを 追跡できます。また、無料アカウントで Databricks 上のマネージド 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、lyso716、Mahmoud Mahdi、Maryann Xue
(このブログ記事はAI翻訳ツールを使用して翻訳されています) 原文記事