Delta Lake の詳細: スキーマの適用と進化

おい、ハブル！ 記事の翻訳をご紹介します 「デルタ湖へのダイビング: スキーマの強制と進化」 著者はBurak Yavuz、Brenner Heintz、Denny Lee。このコースの開始に備えて準備されました。 データエンジニア オータスから。

データは私たちの経験と同様、常に蓄積され、進化しています。 それに追いつくためには、私たちの世界のメンタル モデルが新しいデータに適応する必要があり、その中には新しい次元、つまり私たちがこれまで知らなかったものを観察する新しい方法が含まれているものもあります。 これらのメンタル モデルは、新しい情報をどのように分類して処理するかを決定するテーブル スキーマとあまり変わりません。

ここで、スキーマ管理の問題が生じます。 ビジネスの課題や要件が時間の経過とともに変化するにつれて、データの構造も変化します。 Delta Lake を使用すると、データの変化に応じて新しい測定値を簡単に導入できます。 ユーザーは、テーブル スキーマを管理するための単純なセマンティクスにアクセスできます。 これらのツールには、ユーザーがエラーや不要なデータでテーブルを意図せず汚染しないように保護する Schema Enforcement と、貴重なデータの新しい列を適切な場所に自動的に追加できるようにする Schema Evolution が含まれます。 この記事では、これらのツールの使用法について詳しく説明します。

テーブルスキーマを理解する

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

スキームの施行とは何ですか?

スキーマ強制 (スキーマ検証とも呼ばれます) は、テーブルのスキーマに一致しないレコードを拒否することでデータ品質を保証する Delta Lake のセキュリティ メカニズムです。 予約制の人気レストランのフロントデスクのホステスのように、テーブルに入力されたデータの各列が、対応する予想される列のリストに含まれているかどうか (つまり、それぞれの列に「予約」があるかどうか) をチェックします。 )、リストにない列を含むレコードは拒否されます。

スキーマの強制はどのように機能しますか?

Delta Lake は書き込み時のスキーマ チェックを使用します。これは、テーブルへのすべての新しい書き込みが、書き込み時にターゲット テーブルのスキーマとの互換性についてチェックされることを意味します。 スキーマに矛盾がある場合、Delta Lake はトランザクションを完全に中止し (データは書き込まれません)、例外を発生させてユーザーに矛盾を通知します。
Delta Lake は、次のルールを使用して、レコードがテーブルと互換性があるかどうかを判断します。 書き込み可能なデータフレーム:

• ターゲットテーブルのスキーマにない追加の列を含めることはできません。 逆に、受信データにテーブルのすべての列が完全に含まれていない場合は、すべて問題ありません。これらの列には、単純に null 値が割り当てられます。
• ターゲットテーブルの列のデータ型と異なる列データ型を持つことはできません。 ターゲット テーブルの列に StringType データが含まれているが、DataFrame 内の対応する列に IntegerType データが含まれている場合、スキーマの強制によって例外がスローされ、書き込み操作が実行されなくなります。
• 大文字と小文字のみが異なる列名を含めることはできません。 これは、同じテーブル内に「Foo」と「foo」という名前の列を定義することはできないことを意味します。 Spark は大文字と小文字を区別するモードまたは大文字と小文字を区別しない (デフォルト) モードで使用できますが、Delta Lake は大文字と小文字を保持しますが、スキーマ ストレージ内では区別されません。 Parquet は、列情報を格納および返すときに大文字と小文字を区別します。 起こり得るエラー、データ破損、またはデータ損失 (Databricks で個人的に経験したこと) を回避するために、この制限を追加することにしました。

これを説明するために、新しく生成された列をまだ受け入れるように構成されていない Delta Lake テーブルに追加しようとしたときに、以下のコードで何が起こるかを見てみましょう。

# Сгенерируем DataFrame ссуд, который мы добавим в нашу таблицу Delta Lake
loans = sql("""
            SELECT addr_state, CAST(rand(10)*count as bigint) AS count,
            CAST(rand(10) * 10000 * count AS double) AS amount
            FROM loan_by_state_delta
            """)

# Вывести исходную схему DataFrame
original_loans.printSchema()

root
  |-- addr_state: string (nullable = true)
  |-- count: integer (nullable = true)
 
# Вывести новую схему DataFrame
loans.printSchema()
 
root
  |-- addr_state: string (nullable = true)
  |-- count: integer (nullable = true)
  |-- amount: double (nullable = true) # new column
 
# Попытка добавить новый DataFrame (с новым столбцом) в существующую таблицу
loans.write.format("delta") 
           .mode("append") 
           .save(DELTALAKE_PATH)

Returns:

A schema mismatch detected when writing to the Delta table.
 
To enable schema migration, please set:
'.option("mergeSchema", "true")'
 
Table schema:
root
-- addr_state: string (nullable = true)
-- count: long (nullable = true)
 
Data schema:
root
-- addr_state: string (nullable = true)
-- count: long (nullable = true)
-- amount: double (nullable = true)
 
If Table ACLs are enabled, these options will be ignored. Please use the ALTER TABLE command for changing the schema.

新しい列を自動的に追加する代わりに、Delta Lake はスキーマを強制し、書き込みを停止します。 どの列 (または列のセット) が不一致の原因となっているかを判断するために、Spark は比較のためにスタック トレースから両方のスキーマを出力します。

スキーマを強制する利点は何ですか?

スキーマの適用はかなり厳格なチェックであるため、運用または使用の準備ができているクリーンで完全に変換されたデータ セットへのゲートキーパーとして使用するのに最適なツールです。 通常は、データを直接フィードするテーブルに適用されます。

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

この最後のハードルに向けてデータを準備するために、多くのユーザーは、テーブルに徐々に構造を導入する単純な「マルチホップ」アーキテクチャを使用します。 これについて詳しくは、記事をご覧ください。 Delta Lake を使用した実稼働グレードの機械学習。

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

データの希薄化を防ぐ

ここまでで、一体何が大騒ぎしているのかと疑問に思っているかもしれません。 結局のところ、特に Delta Lake を初めて使用する場合は、予期しない「スキーマの不一致」エラーによってワークフローがつまづいてしまうことがあります。 必要に応じてスキーマを変更して、何があっても DataFrame を記述できるようにしてはどうでしょうか?

古いことわざにあるように、「XNUMX オンスの予防は XNUMX ポンドの治療に匹敵します。」 スキーマの適用に注意しないと、ある時点で、データ型の互換性の問題が厄介な問題として浮上します。一見均質な生データ ソースには、エッジ ケース、破損した列、不正なマッピング、またはその他の恐ろしい事態が含まれている可能性があります。悪夢。 最善のアプローチは、スキーマの強制を使用して、これらの敵を門前で阻止し、後で本番コードの暗い深さに潜み始めたときに対処するのではなく、明るいうちに対処することです。

スキーマを強制すると、変更を承認しない限りテーブルのスキーマが変更されないことが保証されます。 これにより、新しい列が頻繁に追加され、データの氾濫によって以前は貴重で圧縮されたテーブルの意味や有用性が失われる場合に発生する可能性のあるデータの希薄化が防止されます。 意図的に行動し、高い基準を設定し、高品質を期待することを奨励することで、スキーマの適用は設計どおりに機能し、誠実さを保ち、スプレッドシートをクリーンに保つのに役立ちます。

さらに検討した結果、本当にそう判断した場合は、 必要 新しい列を追加します。問題ありません。以下は XNUMX 行の修正です。 解決策は回路の進化！

スキーマ進化とは何ですか?

スキーマ進化は、時間の経過とともに変化するデータに応じて現在のテーブル スキーマを簡単に変更できる機能です。 これは、追加または書き換え操作を実行して XNUMX つ以上の新しい列を含むようにスキーマを自動的に調整するときに最もよく使用されます。

スキーマ進化はどのように機能するのでしょうか?

前のセクションの例に従うと、開発者はスキーマの進化を使用して、スキーマの不整合のために以前に拒否された新しい列を簡単に追加できます。 追加することで回路進化が発動します .option('mergeSchema', 'true') Spark チームへ .write или .writeStream.

# Добавьте параметр mergeSchema
loans.write.format("delta") 
           .option("mergeSchema", "true") 
           .mode("append") 
           .save(DELTALAKE_SILVER_PATH)

グラフを表示するには、次の Spark SQL クエリを実行します。

# Создайте график с новым столбцом, чтобы подтвердить, что запись прошла успешно
%sql
SELECT addr_state, sum(`amount`) AS amount
FROM loan_by_state_delta
GROUP BY addr_state
ORDER BY sum(`amount`)
DESC LIMIT 10

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

リクエストにパラメータを含めることにより、 mergeSchema、データフレームには存在するがターゲットテーブルには存在しないすべての列は、書き込みトランザクションの一部としてスキーマの最後に自動的に追加されます。 ネストされたフィールドを追加することもでき、これらも対応する構造列の末尾に追加されます。

データ エンジニアとデータ サイエンティストは、このオプションを使用して、古い列に基づく既存のモデルを壊すことなく、既存の機械学習実稼働テーブルに新しい列 (おそらく、最近追跡された指標または今月の販売実績列) を追加できます。

次のタイプのスキーマ変更は、テーブルの追加または書き換え中のスキーマ進化の一環として許可されます。

• 新しい列の追加 (これが最も一般的なシナリオ)
• データ型を NullType -> 他の型に変更するか、ByteType -> ShortType -> IntegerType に昇格します。

スキーマの進化内で許可されていないその他の変更では、スキーマとデータを追加して書き直す必要があります。 .option("overwriteSchema", "true")。 たとえば、列「Foo」が元々整数で、新しいスキーマが文字列データ型だった場合、すべての Parquet(data) ファイルを書き直す必要があります。 このような変更には次のものが含まれます。

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

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

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

回路の進化の利点は何ですか?

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

まとめ

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

一方、スキーマの進化は、スキーマを簡素化することで施行を補完します。 想定された 自動スキーマ変更。 結局のところ、列を追加するのは難しくないはずです。

スキームの強制適用は陽であり、スキームの進化は陰です。 これらの機能を併用すると、ノイズ抑制と信号調整がこれまでより簡単になります。

この記事への貢献については、Mukul Murthy 氏と Pranav Anand 氏にも感謝いたします。

このシリーズの他の記事:

Delta Lake の詳細: トランザクション ログの開梱

関連記事

Delta Lake を使用した本番グレードの機械学習

データレイクとは何ですか?

コースについて詳しく見る

出所： habr.com