Delta Lakeを深く掘り下げる: スキーマ適用と進化

Databricks でこのノートブックシリーズをお試しください

データは、私たちの経験と同様に、常に進化し、蓄積されています。それに遅れないようにするには、世界のメンタルモデルを新しいデータに適応させる必要があります。その中には、新しい次元、つまり、これまで考えたことのなかった物事を見る新しい方法が含まれています。これらのメンタルモデルは、テーブルのスキーマと似ており、新しい情報をどのように分類し、処理するかを定義します。

これにより、スキーマ管理について説明します。ビジネス上の問題と要件が時間の経過とともに進化するにつれて、データの構造も進化します。Delta Lakeを使用すると、データの変更に合わせて、新しい次元を簡単に組み込むことができます。ユーザーは、テーブルのスキーマを制御するためのシンプルなセマンティクスにアクセスできます。これらのツールには、ユーザーが誤ってテーブルを誤りやガベージデータで汚染するのを防ぐスキーマ適用や、リッチデータに属する新しい列を自動的に追加できるスキーマ進化などがあります。このブログでは、これらのツールの使用方法について詳しく説明します。

テーブルスキーマについて

Apache Spark™ のすべての DataFrame には、データ型や列、メタデータなど、データの形状を定義するブループリントであるスキーマが含まれています。Delta Lake では、テーブルのスキーマはトランザクションログ内に JSON 形式で保存されます。

スキーマ適用とは

スキーマ検証とも呼ばれるスキーマ適用は、テーブルのスキーマと一致しないテーブルへの書き込みを拒否することにより、データ品質を保証する Delta Lake のセーフガードです。予約のみを受け付ける忙しいレストランのフロントデスクマネージャーのように、テーブルに挿入されたデータの各列が、予期される列のリストにあるかどうか (つまり、それぞれに「予約」があるかどうか) を確認し、リストにない列を含む書き込みを拒否します。

スキーマ適用の仕組み

Delta Lake は、スキーマ検証を書き込み時に使用します。つまり、テーブルへのすべての新しい書き込みは、書き込み時にターゲットテーブルのスキーマとの互換性がチェックされます。スキーマに互換性がない場合、Delta Lake はトランザクションを完全にキャンセルし (データは書き込まれません)、例外を発生させて、不一致をユーザーに通知します。

テーブルへの書き込みに互換性があるかどうかを判断するために、Delta Lake は次のルールを使用します。書き込まれる DataFrame:

• ターゲットテーブルのスキーマに存在しない追加の列を含めることはできません。逆に、入力データにテーブル内のすべての列が含まれていなくても問題ありません。これらの列には単純に null 値が割り当てられます。
• 列のデータ型が、ターゲットテーブルの列のデータ型と異なるデータ型を持つことはできません。ターゲットテーブルの列に StringType データが含まれているが、DataFrame の対応する列に IntegerType データが含まれている場合、スキーマ適用は例外を発生させ、書き込み操作が行われないようにします。
• 大文字と小文字のみが異なる列名を含めることはできません。これは、同じテーブルに「Foo」や「foo」などの列を定義できないことを意味します。Spark は大文字と小文字を区別するモードまたは区別しないモード (デフォルト) で使用できますが、Delta Lake はスキーマを保存するときに大文字と小文字を保持しますが、区別しません。Parquet は、列情報を保存して返すときに大文字と小文字を区別します。潜在的な間違い、データの破損、または損失の問題 (Databricks で個人的に経験した) を回避するために、この制限を追加することにしました。

例として、まだ受け入れるように設定されていない Delta Lake テーブルに、新しく計算された列を追加しようとした場合に、以下のコードで何が起こるかを見てみましょう。

Delta Lake は、新しい列を自動的に追加するのではなく、スキーマを適用し、書き込みが発生しないようにします。どの列が不一致の原因になったかを特定するために、Spark は比較のためにスタックトレースに両方のスキーマを出力します。

スキーマ適用はどのように役立ちますか

スキーマ適用は非常に厳格なチェックであるため、本番環境または消費の準備ができた、クリーンで完全に変換されたデータセットのゲートキーパーとして使用するのに最適なツールです。通常、直接供給するテーブルに適用されます。

• 機械学習アルゴリズム
• BI ダッシュボード
• データ分析および視覚化ツール
• 高度に構造化され、厳密に型指定されたセマンティックスキーマを必要とする本番システム

この最終的なハードルに備えてデータを準備するために、多くのユーザーは、テーブルに構造を段階的に追加するシンプルな「マルチホップ」アーキテクチャを採用しています。詳細については、「Delta Lake を使用した機械学習の本番環境化」というタイトルの投稿をご覧ください。

もちろん、スキーマ適用はパイプラインのどこでも使用できますが、たとえば、入力データに 1 つの列を追加したことを忘れたために、テーブルへのストリーミング書き込みが失敗すると、少しイライラする可能性があることに注意してください。

データの希釈の防止

この時点で、あなたは自分自身に尋ねているかもしれません、一体何が騒ぎなのですか?結局のところ、特に Delta Lake を初めて使用する場合は、予期しない「スキーマの不一致」エラーがワークフローでつまずくことがあります。DataFrame を何があっても書き込めるように、スキーマを必要なように変更させないのはなぜですか?

古いことわざにあるように、「予防は治療に勝る」。スキーマを適用しないと、データ型の互換性の問題が醜い頭をもたげます。一見均質な生のデータのソースには、エッジケース、破損した列、誤ったマッピング、または夜にぶつかるその他の恐ろしいものが含まれている可能性があります。はるかに優れたアプローチは、スキーマ適用を使用して、これらの敵をゲートで阻止し、後で本番コードの影の奥深くに潜んでいるときにではなく、日中にそれらに対処することです。

スキーマ適用は、肯定的な選択をして変更しない限り、テーブルのスキーマが変更されないという安心感を提供します。新しい列が頻繁に追加されると、以前はリッチで簡潔なテーブルがデータの氾濫により意味と有用性を失う可能性がある場合に発生する可能性のあるデータの「希釈」を防ぎます。意図的になり、高い基準を設定し、高品質を期待するように促すことで、スキーマ適用はまさに設計されたとおりに機能し、正直さを保ち、テーブルをきれいに保ちます。

さらに検討した結果、本当にその新しい列を追加するつもりだったと判断した場合は、以下で説明するように、簡単な 1 行の修正です。解決策はスキーマ進化です!

スキーマ進化とは

スキーマ進化は、時間の経過とともに変化するデータに対応するために、テーブルの現在のスキーマを簡単に変更できる機能です。最も一般的には、追加または上書き操作を実行するときに、スキーマを自動的に適応させて 1 つ以上の新しい列を含めるために使用されます。

スキーマ進化の仕組み

前のセクションの例に続いて、開発者はスキーマ進化を簡単に使用して、以前にスキーマの不一致が原因で拒否された新しい列を追加できます。スキーマ進化は、 .option('mergeSchema', 'true') を .write または .writeStream Spark コマンドに追加することでアクティブ化されます。

プロットを表示するには、次の Spark SQL ステートメントを実行します。

または、spark.databricks.delta.schema.autoMerge = True を Spark 構成に追加して、Spark セッション全体に対してこのオプションを設定することもできます。スキーマ適用は、意図しないスキーマの不一致について警告しなくなるため、注意して使用してください。

クエリに mergeSchema オプションを含めることにより、DataFrame に存在するがターゲットテーブルに存在しない列は、書き込みトランザクションの一部としてスキーマの最後に追加されます。ネストされたフィールドも追加でき、これらのフィールドはそれぞれの struct 列の最後にも追加されます。

データエンジニアと科学者は、このオプションを使用して、新しい列 (おそらく新しく追跡されたメトリック、または今月の売上高の列) を既存の機械学習本番テーブルに追加して、古い列に依存する既存のモデルを壊すことなく追加できます。

次のタイプのスキーマ変更は、テーブルの追加または上書き中にスキーマ進化の対象となります。

• 新しい列の追加 (これが最も一般的なシナリオです)
• データ型の NullType -> 他の型への変更、または ByteType -> ShortType -> IntegerType からのアップキャスト

スキーマ進化の対象とならないその他の変更では、.option("overwriteSchema", "true") を追加して、スキーマとデータを上書きする必要があります。たとえば、列「Foo」が元々 integer データ型であり、新しいスキーマが文字列データ型である場合、すべての Parquet (データ) ファイルを書き換える必要があります。これらの変更には、次のものが含まれます。

• 列の削除
• 既存の列のデータ型の (インプレース) 変更
• 大文字と小文字のみが異なる列名の名前変更 (例: 「Foo」と「foo」)

最後に、今後の Spark 3.0 のリリースでは、明示的な DDL (ALTER TABLE を使用) が完全にサポートされ、ユーザーはテーブルスキーマに対して次のアクションを実行できるようになります。

• 列の追加
• 列のコメントの変更
• トランザクションログの保持期間の設定など、テーブルの動作を定義するテーブルプロパティの設定

スキーマ進化はどのように役立ちますか

スキーマ進化は、テーブルのスキーマを変更する (DataFrame に存在すべきではない列を誤って追加した場合とは対照的に) 意図している場合はいつでも使用できます。正しい列名とデータ型を明示的に宣言する必要なく、自動的に追加するため、スキーマを移行する最も簡単な方法です。

まとめ

スキーマ適用は、テーブルと互換性のない新しい列またはその他のスキーマ変更を拒否します。これらの高い基準を設定し、維持することにより、アナリストとエンジニアは、データが最高レベルの整合性を持ち、明確に推論できることを信頼でき、より良いビジネス上の意思決定を行うことができます。

コインの裏側では、スキーマ進化は、意図されたスキーマ変更を自動的に簡単に行えるようにすることで、適用を補完します。結局のところ、列の追加は難しくないはずです。

スキーマ適用は、スキーマ進化の陰です。これらの機能を一緒に使用すると、ノイズを遮断し、信号に同調することがこれまで以上に簡単になります。

また、このブログにご協力いただいた Mukul Murthy と Pranav Anand に感謝します。

関連情報

このシリーズの記事:
Delta Lake を掘り下げる #1: トランザクションログのアンパック
Delta Lake を掘り下げる #2: スキーマ適用と進化
Delta Lake を掘り下げる #3: DML 内部 (更新、削除、マージ)

Delta Lake を使用した機械学習の本番環境化
データレイクとは

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