環境バージョンの互換性
パブリックプレビュー
LakeFlow Pipelinesの環境バージョンはパブリックプレビュー段階です。
環境バージョンが設定されているパイプラインは、Spark Connect を介して Python コードを実行します。このページでは、互換性のないもの、動作が異なるもの、および Databricks がパイプラインをスキャンして影響を受けるパターンを特定する方法について説明します。
制限事項
環境バージョンは、パイプラインのすべての機能とまだ互換性があるわけではありません。環境バージョンを設定したパイプラインの実行は、パイプラインのPythonコードが以下のいずれかに該当する場合に失敗します。
- パイプラインデコレータで装飾された関数内で、Sparkセッションの状態を変更します。例としては、
spark.conf.set(...)、spark.sql("USE CATALOG ...")、createOrReplaceTempViewなどがあります。 - Spark Connect では利用できないPySpark APIs使用します。これには
SparkContext、RDD、SQLContext、および Py4J APIs含まれます。 Spark Connectでサポートされている機能については、こちらをご覧ください。
パイプライン上で環境バージョンを有効にした結果、パイプラインが失敗した場合、その環境バージョンを無効にすると、パイプラインは以前の状態に戻ります。
行動の変化
Spark Connect には、従来の PySpark ランタイムと動作が異なる点がいくつかあります。詳細なリファレンスについては、「Spark Connect と従来の Spark の比較」を参照してください。互換性スキャンは、これらのパターンを事前に検出し、対処されるまで移行をブロックするため、本番運用データに影響を与える前にそれらを見つけて修正できます。
パイプラインにおいて、動作が異なる可能性のある最も一般的な状況は次のとおりです。
インターリーブされたDataFrame構築とセッションの変更
パイプラインがDataFrame構築すると、 Sparkセッション状態が変更され (たとえば、暫定カタログまたはスキーマの変更、構成の設定、一時ビューの置き換え、またはUDF再登録)、 DataFrameが使用されます。
- 環境バージョンが指定されていない場合、DataFrameは 変更前の セッション状態を使用します。
- 環境バージョンを使用すると、DataFrame は 変更後の セッション状態を使用します。
例えば：
from pyspark import pipelines as dp
spark.createDataFrame([(1, "Original Row")], ["id", "data"]) \
.createOrReplaceTempView("my_view")
df = spark.sql("SELECT * FROM my_view")
spark.createDataFrame([(2, "Replaced Row")], ["id", "data"]) \
.createOrReplaceTempView("my_view")
@dp.materialized_view
def mytable():
return df
環境バージョンがない場合、 mytableには[(1, "Original Row")]が含まれます。環境バージョンでは、 mytableには[(2, "Replaced Row")]が含まれます。
可変なPython状態を参照するUDF
UDFが、UDFの定義後に値が変化するPythonグローバル変数を参照する場合：
- 環境バージョンが指定されていない場合、UDFは変数の 最新の 値を使用します。
- 環境バージョンを使用すると、UDF は UDF が定義された時点の値 を使用します。
例えば：
from pyspark import pipelines as dp
from pyspark.sql.functions import col, udf
suffix = "a"
@udf
def my_udf(s):
return s + suffix
suffix = "b"
@dp.materialized_view
def my_mv():
return spark.createDataFrame([("alex",)], ["name"]).select(my_udf(col("name")))
環境バージョンがない場合、 my_mvには[("alex_b",)]が含まれます。環境バージョンでは、 my_mvには[("alex_a",)]が含まれます。
パイプラインがどちらかのパターンに依存している場合は、環境バージョンを有効にする前に監査を実施してください。
互換性スキャン
互換性スキャンは、環境バージョン下で異なる結果を生成する可能性のあるパイプライン内のコードパターンを検出するため、パイプラインが自動的に移行される前に修正できます。パイプラインでスキャンが有効になっている場合:
- 各更新では、検出されたパターンごとに、パイプライン イベント ログに
BehaviorChangeInSparkConnectWARNイベントが 1 つ出力されます。 - 前回の正常な更新で発生したすべての互換性警告が対処されるまで、パイプラインは環境バージョンに移行されず、手動で有効にすることもできません。
このチェックは、以前の更新がないパイプラインや、すでに環境バージョンが設定されているパイプラインには適用されません。
パイプラインでスキャンを有効にする
pipelines.environmentVersion.enableCompatibilityScan パイプライン構成を追加することで、互換性スキャンを有効にできます。パイプラインエディター UI から、またはパイプライン構成 JSON にエントリを追加することで、設定を追加できます。
UIを通して ：
- パイプラインエディタから、 [設定] をクリックします。
- パイプライン設定の 「構成」 セクションを探してください。
- クリック
設定を追加します 。 - キーとして
pipelines.environmentVersion.enableCompatibilityScan、値としてtrueを入力してください。 - パイプライン設定を保存します。
パイプライン内のJSON :
configurationブロックに以下のエントリを追加してください。
"configuration": {
"pipelines.environmentVersion.enableCompatibilityScan": "true"
}
互換性に関する警告を確認して解決する
パイプラインで環境バージョンをブロックしているパターンを見つけてクリアするには：
- パイプラインを ドライラン モードで実行し、パイプラインイベントLogsをクエリー して
BehaviorChangeInSparkConnectWARNイベントを確認します。各イベントは、検出された1つのパターンを報告します。問題コード、パターン例、推奨される修正の完全なリストについては、互換性イベントリファレンスを参照してください。 - 提案された修正に従って検出されたパターンを削除するようにパイプラインコードを更新し、パイプラインを再度実行してください。
- 更新が成功し、互換性イベントがそれ以上発行されなくなるまで繰り返します。その後、パイプラインは自動的に移行されます。また、環境バージョンを自身で有効にすることもできます。
環境バージョンを有効にすると、Databricks がパイプラインを自動的に移行する場合でも、ユーザー自身が environment_version を設定する場合でも、同じ安全チェックが実行されます。互換性の警告が未解決のパイプラインは、警告が解決されるまで環境バージョンに移行されません。移行を安全に完了できない場合、または何らかの理由で失敗した場合、データは書き込まれる前に停止し、パイプラインは以前のランタイムで実行を継続します。
これらの理由のいずれかで更新が停止した場合、パイプラインイベントlogと更新エラーメッセージに、原因と解決手順が記載されます。それらのステップに従い、パイプラインを再度実行して移行を完了してください。互換性の警告が偽陽性であると思われる場合は、フラグが立てられたパターンを解決するか、Databricks サポートにお問い合わせください。
互換性イベントのリファレンス
パイプラインで互換性スキャンが実行されると、検出されたパターンごとに パイプラインイベント Logs に BehaviorChangeInSparkConnect WARN イベントが 1 つ出力されます。前回の正常な更新でパターンが検出された場合、そのパターンが対処されるまで、パイプラインは環境バージョンに移行されません。
各イベントは、検出された内容を識別する単一の問題コードを報告します。コードを調べるには、 「問題コード」表で該当のコードを探してください。各行は、例となるパターンと推奨される修正方法が記載されているカテゴリセクションへのリンクになっています。
イベントの形状
BehaviorChangeInSparkConnect イベントは標準のパイプラインイベントログスキーマに従います。
event_typebehavior_change_in_spark_connectです。levelWARNです。detailsbehavior_change_in_spark_connectオブジェクトが含まれており、そのオブジェクトには単一のissueフィールドがあります。問題値は、以下のコードのいずれかです。message検出されたパターンを人間が読みやすい形で記述したものです。
発行コード
カテゴリー | 問題コード | 説明 |
|---|---|---|
| DataFrameの作成後にカタログが変更されました。 既存のDataFrameは、新しいデフォルトカタログを使用してテーブルを解決する可能性があります。 | |
|
| |
| DataFrameが作成された後、デフォルトのデータベースが変更されました。既存のDataFrameは、新しいデフォルトデータベースを使用してテーブルを解決する可能性があります。 | |
|
| |
| フロー関数はチェックポイントコマンドを呼び出します。 | |
| フロー関数は、DataFrameビュー（ | |
| フロー機能はリソースプロファイルを作成します。 | |
| フロー関数は、 | |
| フロー関数は、対象テーブルに対してEager | |
| フロー関数は、Spark MLの即時実行操作を実行します。 | |
| フロー関数はPythonデータソースを登録します。 | |
| フロー機能は、アクティブなストリーミングクエリハンドル上で動作します。 | |
| フロー関数は、ストリーミング クエリ リスナーを登録または削除します。 | |
| フロー関数は、ストリーミングクエリを管理するために | |
| フロー関数は、即時 | |
| フロー関数は、即時 | |
| フロー関数はストリーミングクエリ（ | |
|
| |
|
| |
|
| |
| グローバルな一時ビューは、それを参照するDataFrameが作成された後、置き換えられました。 置換は既存のDataFrameに反映される可能性があります。 | |
| 一時ビューは、それを参照するDataFrameが作成された後、置き換えられました。 置換は既存のDataFrameに反映される可能性があります。 | |
| 参照するDataFrameが作成された後、 UDF同じ名前で再登録されました。 既存のDataFrameは、新しいUDF定義を使用できる可能性があります。 | |
| UDTFを参照するDataFrameが作成された後、同じ名前でUDTFが再登録されました。既存のDataFrameは、新しいUDTF定義を使用できる。 | |
| UDFは、グローバルな可変Python変数を参照します。環境バージョンを使用する場合、UDFは呼び出し時ではなく、UDFが定義された時点の変数の値を使用します。 | |
| UDTFは、グローバルな可変Python変数を参照します。環境バージョンを使用する場合、UDTFは呼び出し時ではなく、UDTFが定義された時点での変数の値を使用します。 |
データベースとカタログの変更
これらの問題は、パイプラインコードがデフォルトのデータベースまたはカタログを変更した場合に発生します。環境バージョンを使用すると、変更前に構築されたDataFrames 、新しいデータベースまたはカタログを使用してテーブルを解決する可能性があります。
イベントをトリガーするパターンの例：
from pyspark import pipelines as dp
spark.sql("USE CATALOG marketing")
df = spark.read.table("events")
spark.sql("USE CATALOG sales") # changes the default catalog after df was created
@dp.materialized_view
def events_summary():
return df.groupBy("region").count()
環境バージョンがない場合、 df marketingカタログからeventsを解決します。環境バージョンを使用すると、 df salesカタログからeventsを解決します。
推奨される修正方法： テーブル名を完全に修飾して、解決がデフォルトのカタログやデータベースに依存しないようにし、DataFrame の作成と使用の間でデフォルトのカタログやデータベースを変更しないようにしてください。
from pyspark import pipelines as dp
df = spark.read.table("marketing.default.events")
@dp.materialized_view
def events_summary():
return df.groupBy("region").count()
Spark構成の変更
これらの問題は、パイプラインコードがSpark構成を変更し、環境バージョンによってDataFrameの動作が変わる可能性がある場合に発生します。
イベントをトリガーするパターンの例：
from pyspark import pipelines as dp
df = spark.read.table("events")
spark.conf.set("spark.sql.ansi.enabled", "true") # changes session conf after df was created
@dp.materialized_view
def events_strict():
return df.selectExpr("CAST(price AS INT) AS price")
環境バージョンが指定されていない場合、キャストはDataFrame作成時の設定値を使用します。 環境バージョンでは、キャストはspark.sql.ansi.enabled=trueを使用し、無効な入力に対して失敗する可能性があります。
推奨される修正方法： パイプラインファイルの先頭、つまりDataFrameが作成される前に、必要なSpark設定をすべて設定してください。クエリごとの設定には、パイプライン仕様のパイプラインのconfiguration設定を使用します。
一時的な表示代替
これらの問題は、一時ビューを参照するDataFrameが作成された後に、パイプライン コードが一時ビューを置き換えるときに発生します。 環境バージョンによっては、既存のDataFrameに新しいビューの内容が反映される場合があります。
イベントをトリガーするパターンの例：
from pyspark import pipelines as dp
spark.createDataFrame([(1, "Original Row")], ["id", "data"]) \
.createOrReplaceTempView("my_view")
df = spark.sql("SELECT * FROM my_view")
spark.createDataFrame([(2, "Replaced Row")], ["id", "data"]) \
.createOrReplaceTempView("my_view")
@dp.materialized_view
def mytable():
return df
環境バージョンがない場合、 mytableには[(1, "Original Row")]が含まれます。環境バージョンでは、 mytableには[(2, "Replaced Row")]が含まれます。
推奨される解決策： 各一時ビューを一度だけ作成し、置き換えないようにしてください。関連データを含む複数のビューが必要な場合は、それぞれに異なる名前を付けてください。
UDFおよびUDTF変異
これらの問題は、パイプラインコードがUDFまたはUDTFを変更し、環境バージョンによって動作が変わる場合に発生します。
イベントをトリガーするパターンの例：
from pyspark import pipelines as dp
from pyspark.sql.functions import col, udf
suffix = "a"
@udf
def my_udf(s):
return s + suffix
suffix = "b"
@dp.materialized_view
def my_mv():
return spark.createDataFrame([("alex",)], ["name"]).select(my_udf(col("name")))
環境バージョンがない場合、 my_mvには[("alex_b",)]が含まれます。環境バージョンでは、 my_mvには[("alex_a",)]が含まれます。
修正案： Pythonのグローバル変数から値を取得するのではなく、UDFに引数として値を渡すか、UDFを定義する前にグローバル変数を設定し、その後変更しないようにしてください。
from pyspark import pipelines as dp
from pyspark.sql.functions import col, lit, udf
@udf
def append_suffix(s, suffix):
return s + suffix
@dp.materialized_view
def my_mv():
return spark.createDataFrame([("alex",)], ["name"]).select(append_suffix(col("name"), lit("b")))
フロー関数内での積極的な実行
これらの問題は、パイプライン コードがパイプライン デコレーター ( @table 、 @materialized_viewなど) によって装飾された関数内で Eager Sparkコマンドを実行するときに発生します。 フロー関数はDataFrameを定義して返すことが期待されています。環境バージョンが設定されているフロー関数内では、データの書き込み、ストリーミング クエリの管理、登録実行するリソース、またはML操作を行う Eerger コマンドは許可されません。
修正案： Eager操作をフロー関数の外に移動し、フロー関数からは代わりにDataFrameを返すようにします。テーブルへの書き込みやストリーミング クエリの開始などの副作用は、パイプライン定義の範囲外に属します。パイプライン エンジンは、フロー関数によって返されたDataFrameの具体化を処理します。
イベントログで互換性イベントを検索します
次のクエリは、パイプラインのすべての互換性イベントを、最新のものから順に返します。
SELECT
timestamp,
message,
details:behavior_change_in_spark_connect:issue AS issue
FROM event_log(<pipeline-id>)
WHERE event_type = 'behavior_change_in_spark_connect'
AND level = 'WARN'
ORDER BY timestamp DESC;
最近のアップデートにおける問題コード別のイベント数をカウントするには：
SELECT
details:behavior_change_in_spark_connect:issue AS issue,
COUNT(*) AS occurrences
FROM event_log(<pipeline-id>)
WHERE event_type = 'behavior_change_in_spark_connect'
AND level = 'WARN'
GROUP BY 1
ORDER BY occurrences DESC;
イベントログのクエリ方法については、 「イベントログのクエリ」を参照してください。
その他のリソース
- パイプラインの環境バージョンの構成 — 機能の概要、自動移行、および環境バージョンを自分で有効にする方法。
- パイプライン イベント ログ スキーマ— 完全なパイプライン イベント ログ スキーマ。
- パイプライン イベント ログ — パイプラインイベント ログをクエリする方法。