環境バージョンの互換性

environment バージョンが設定されたパイプラインは、Spark Connect を介してPythonコードを実行します。 このページでは、何が互換性がないのか、何が異なる挙動をするのか、そしてDatabricksがどのようにパイプラインから影響を受けたパターンをスキャンするかを扱っています。

制限事項

環境バージョンは、すべてのパイプライン機能とまだ互換性がありません。 パイプラインのPython コードが次のいずれかの処理を行うと、環境バージョンが設定されたパイプラインの実行が失敗します。

• パイプライン デコレーターで修飾された関数内の Spark セッション状態を変更します。 たとえば、 spark.conf.set(...)、 spark.sql("USE CATALOG ...")、 createOrReplaceTempViewなどがあります。
• SparkContext、RDD、SQLContext、Py4J API など、Spark Connect で使用できない PySpark API を使用します。 Spark Connect でサポートされている内容を参照してください。

パイプラインで環境バージョンを有効にすると失敗する場合、環境バージョンを無効にすると、パイプラインは以前の状態に戻ります。

動作の変更

Spark Connect には、従来の PySpark ランタイムと少数の動作の違いがあります。 完全なリファレンスについては、 Spark Connect とクラシック Spark を参照してください。 互換性スキャンはこれらのパターンを事前に検出し、対処されるまで移行をブロックするため、本番データに影響を与える前に発見・修正できます。

パイプラインでは、動作が異なる可能性がある最も一般的な状況は次のとおりです。

• インターリーブされた DataFrame の構築とセッションの変更
• 変更可能なPython状態を参照する UDF

インターリーブされた 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 が定義された後に値が変化するPythonグローバル変数を UDF が参照する場合:

• 環境バージョンがない場合、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",)]が含まれます。

パイプラインがいずれかのパターンに依存している場合は、環境バージョンを有効にする前に監査します。

互換性スキャン

互換性スキャンは、環境バージョンによって異なる結果が出るコードパターンをパイプライン内で見つけ出し、 パイプラインが自動的に移行される前に修正できます。 パイプラインでスキャンが有効になっている場合:

• 各更新は検出されたパターンごとにパイプラインイベントログに1つの BehaviorChangeInSparkConnectWARN イベントを発行します。
• パイプラインは環境バージョンに移行せず、前回の成功したアップデートによる互換性警告がすべて解決されるまでは、自分で環境バージョンを有効にすることもできません。

このチェックは、以前のアップデートがないパイプラインや、すでに環境バージョンセットが設定されているパイプラインには適用されません。

パイプラインでスキャンを有効にする

互換性スキャンを有効にするには、 pipelines.environmentVersion.enableCompatibilityScan パイプライン構成を追加します。 パイプライン エディター UI を使用するか、パイプライン構成 JSON にエントリを追加して、構成を追加できます。

UI を使用して次の操作を行います。

1. パイプライン エディターで、[ 設定] をクリックします。
2. パイプライン設定の [構成] セクションを見つけます。
3. [ 構成を追加します。
4. キーとして pipelines.environmentVersion.enableCompatibilityScan を入力し、値として true します。
5. パイプラインの設定を保存します。

パイプライン JSON で次の手順を実行します。

configuration ブロックに次のエントリを追加します。

"configuration": {
  "pipelines.environmentVersion.enableCompatibilityScan": "true"
}

互換性警告の確認と解決

パイプライン上の環境バージョンを妨げるパターンを見つけて除去するには:

1. パイプラインを ドライラン モードで実行し、パイプライン イベントログで イベント BehaviorChangeInSparkConnectWARN 検索します。 各イベントは、検出された 1 つのパターンを報告します。 問題コード、サンプル パターン、推奨される修正プログラムの完全な一覧については、 互換性イベント リファレンスを参照してください。
2. 提案された修正後に検出されたパターンを除去するためにパイプラインコードを更新し、再度パイプラインを実行してください。
3. この繰り返しを、成功したアップデートで互換性イベントが出なくなるまで続けます。 パイプラインは自動的に移行でき、 環境バージョンも自分で有効にできます。

環境バージョンを有効にすると、Databricksがパイプラインを自動的に移行するか、あなたが自分で設定するかの安全チェック environment_version 同じです。 未解決の互換性警告があるパイプラインは、警告が解決されるまで環境バージョンに移行しません。 移行が安全に完了できない場合や何らかの理由で失敗した場合、データを書き込む前に停止し、パイプラインは以前の実行時で実行を続けます。

これらの理由で更新が停止した場合、パイプラインイベントログと更新エラーメッセージが原因と解決手順を説明します。 その手順に従い、再度パイプラインを実行して移行を完了してください。 互換性警告が誤検知だと思われる場合は、フラグが立てられたパターンを解決するか、Azure Databricksサポートに連絡してください。

互換性イベントリファレンス

互換スキャンがパイプライン上で実行されると、検出されたパターンごとにパイプラインイベントログに1つのBehaviorChangeInSparkConnectWARNイベントを出力します。 前回の成功した更新でパターンが検出された場合、パターンが処理されるまでパイプラインは環境バージョンに移行されません。

各イベントは、検出された内容を識別する 1 つの問題コードを報告します。 コードを検索するには、[ 問題コード ] テーブルでコードを見つけます。各行は、パターンの例と修正候補を含むカテゴリ セクションにリンクします。

イベント図形

BehaviorChangeInSparkConnect イベントは、標準の パイプライン イベント ログ スキーマに従います。

• event_type は behavior_change_in_spark_connectです。
• level は WARNです。
• detailsには、単一のbehavior_change_in_spark_connect フィールドを持つissue オブジェクトが含まれています。 問題の値は、以下に示すコードの 1 つです。
• message は、検出されたパターンの人間が判読できる説明です。

問題コード

データベースとカタログの変更

これらの問題は、パイプライン コードが既定のデータベースまたはカタログを変更するときに生成されます。 環境バージョンでは、変更の前に構築された DataFrame は、新しいデータベースまたはカタログを使用してテーブルを解決できます。

イベントをトリガーするパターンの例:

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はevents カタログからのmarketingを解決します。 環境バージョンでは、dfはevents カタログからのsalesを解決します。

推奨される修正: テーブル名を完全修飾して、解決が既定のカタログまたはデータベースに依存しないようにし、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 構成の変更

これらの問題は、パイプライン コードが環境バージョンで DataFrame の動作を変更できる方法で Spark 構成を変更するときに生成されます。

イベントをトリガーするパターンの例:

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 作成時に conf 値を使用します。 環境バージョンでは、キャストは 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")]が含まれます。

推奨される修正: 各一時ビューを 1 回作成し、置き換えないでください。 関連データを含む複数のビューが必要な場合は、それぞれに個別の名前を付けます。

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",)]が含まれます。

Suggested fix: 値を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 など) によって装飾された関数内で一括 Spark コマンドを実行すると生成されます。 フロー関数は、DataFrame を定義して返す必要があります。データの書き込み、ストリーミング クエリの管理、リソースの登録、または ML 操作の実行を行う一括コマンドは、環境バージョン セットを持つフロー関数内では許可されません。

推奨される修正: 一括操作をフロー関数の外部に移動し、代わりにフロー関数から 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;

イベント ログのクエリを実行する方法については、イベント ログのクエリを参照してください。

その他のリソース

• パイプラインの環境バージョンを設定する — 機能概要、自動移行、そして環境バージョンを自分で有効化する方法。
• パイプライン イベント ログ スキーマ — パイプライン イベント ログ スキーマ全体。
• パイプライン イベント ログ — パイプライン イベント ログに対してクエリを実行する方法。