Kompatibilität der Umgebungsversion

Important

Umgebungsversionen für Lakeflow-Pipelines sind in der öffentlichen Vorschau verfügbar.

Pipelines mit einer environment-Version legen fest, Python Code über Spark Connect auszuführen. Diese Seite behandelt, was inkompatibel ist, was sich anders verhält und wie Databricks eine Pipeline nach betroffenen Mustern scannt.

Einschränkungen

Umgebungsversionen sind noch nicht mit allen Pipelinefunktionen kompatibel. Eine Pipelineausführung mit einem Umgebungsversionssatz schlägt fehl, wenn der Python Code der Pipeline einen der folgenden Aktionen ausführt:

  • Mutiert den Spark-Sitzungszustand innerhalb einer Funktion, die mit einem Pipeline-Dekorateur versehen ist. Beispiele sind : spark.conf.set(...), spark.sql("USE CATALOG ...")und createOrReplaceTempView.
  • Verwendet PySpark-APIs, die in Spark Connect nicht verfügbar sind, einschließlich SparkContext, RDD, SQLContextund alle Py4J-APIs. Erfahren Sie , was in Spark Connect unterstützt wird.

Wenn die Aktivierung einer Umgebungsversion für eine Pipeline zu einem Fehler führt, gibt das Deaktivieren der Umgebungsversion die Pipeline in den vorherigen Zustand zurück.

Verhaltensänderungen

Spark Connect weist eine kleine Anzahl von Verhaltensunterschieden von der klassischen PySpark-Laufzeit auf. Die vollständige Referenz finden Sie unter Spark Connect im Vergleich zum klassischen Spark . Der Kompatibilitätsscan erkennt diese Muster im Voraus und blockiert die Migration, bis sie adressiert sind, sodass Sie sie finden und beheben können, bevor sie die Produktionsdaten beeinflussen.

In einer Pipeline sind die häufigsten Situationen, in denen sich das Verhalten unterscheiden kann:

Interleaved DataFrame-Konstruktion und Sitzungsmutation

Wenn eine Pipeline einen DataFrame erstellt, ändert sich der Spark-Sitzungszustand (z. B. ändert den Standardkatalog oder das Standardschema, legt eine Konfiguration fest, ersetzt eine temporäre Ansicht oder registriert eine UDF erneut), und verwendet dann den DataFrame:

  • Ohne Eine Umgebungsversion verwendet DataFrame den Sitzungszustand vor der Mutation .
  • Mit einer Umgebungsversion verwendet DataFrame den Sitzungszustand nach der Mutation .

Beispiel:

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

Enthält ohne eine Umgebungsversion mytable[(1, "Original Row")]. Enthält eine Umgebungsversion mytable[(2, "Replaced Row")].

UDFs, die auf änderbare Python Zustand verweisen

Wenn eine UDF auf eine Python globale Variable verweist, deren Wert sich ändert, nachdem die UDF definiert wurde:

  • Ohne Eine Umgebungsversion verwendet die UDF den neuesten Wert der Variablen.
  • Bei einer Umgebungsversion verwendet die UDF den Wert zum Zeitpunkt der Definition der UDF.

Beispiel:

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")))

Enthält ohne eine Umgebungsversion my_mv[("alex_b",)]. Enthält eine Umgebungsversion my_mv[("alex_a",)].

Wenn eine Pipeline von beiden Mustern abhängig ist, überwachen Sie sie, bevor Sie eine Umgebungsversion aktivieren.

Kompatibilitätsscan

Der Kompatibilitätsscan findet Codemuster in deiner Pipeline, die unter einer Umgebungsversion unterschiedliche Ergebnisse liefern würden, sodass du sie vor der automatischen Migration einer Pipeline beheben kannst. Wenn der Scan in einer Pipeline aktiviert ist:

  • Jedes Update sendet pro erkanntem Muster ein Ereignis BehaviorChangeInSparkConnectWARN im Pipeline-Ereignisprotokoll.
  • Die Pipeline wird nicht auf eine Umgebungsversion migriert, und Sie können eine nicht selbst aktivieren, bis alle Kompatibilitätswarnungen des vorherigen erfolgreichen Updates behoben sind.

Diese Überprüfung gilt nicht für eine Pipeline, die kein vorheriges Update hat oder die bereits eine Umgebungsversion gesetzt hat.

Aktivieren des Scans in einer Pipeline

Sie können die Kompatibilitätsüberprüfung aktivieren, indem Sie die pipelines.environmentVersion.enableCompatibilityScan Pipelinekonfiguration hinzufügen. Sie können die Konfiguration über die Pipeline-Editor-Benutzeroberfläche hinzufügen oder einen Eintrag zur Pipelinekonfigurations-JSON hinzufügen.

Über die Benutzeroberfläche:

  1. Klicken Sie im Pipeline-Editor auf "Einstellungen".
  2. Suchen Sie den Abschnitt "Konfiguration" in den Pipelineeinstellungen.
  3. Klicken Sie auf das Plussymbol.Konfiguration hinzufügen.
  4. Geben Sie pipelines.environmentVersion.enableCompatibilityScan als Schlüssel und true als Wert ein.
  5. Speichern Sie die Pipelineeinstellungen.

In der Pipeline JSON:

Fügen Sie dem Block den folgenden Eintrag hinzu configuration :

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

Überprüfen und beheben Sie Kompatibilitätswarnungen

Um die Muster zu finden und zu beseitigen, die eine Umgebungsversion in Ihrer Pipeline blockieren:

  1. Führen Sie die Pipeline im Trockenlaufmodus aus und fragen Sie dann das Pipeline-Ereignisprotokoll nach Ereignissen ab BehaviorChangeInSparkConnectWARN . Jedes Ereignis meldet ein erkanntes Muster. Eine vollständige Liste der Problemcodes, Beispielmuster und vorgeschlagene Korrekturen finden Sie in der Referenz zu Kompatibilitätsereignissen .
  2. Aktualisieren Sie den Pipeline-Code, um die erkannten Muster gemäß der vorgeschlagenen Lösung zu entfernen, und führen Sie die Pipeline erneut aus.
  3. Wiederholen Sie das, bis ein erfolgreiches Update keine weiteren Kompatibilitätsereignisse mehr auslöst. Die Pipeline kann dann automatisch migriert werden, und du kannst auch selbst eine Umgebungsversion aktivieren.

Das Aktivieren einer Umgebungsversion führt dieselben Sicherheitskontrollen aus, ob Databricks die Pipeline automatisch migriert oder man selbst einsetzt environment_version . Eine Pipeline mit ungelösten Kompatibilitätswarnungen wechselt erst in eine Umgebungsversion, wenn die Warnungen behoben sind. Wenn die Migration nicht sicher abgeschlossen werden kann oder aus irgendeinem Grund fehlschlägt, stoppt sie vor dem Schreiben von Daten und die Pipeline läuft weiter mit ihrer vorherigen Laufzeit.

Wenn ein Update aus einem dieser Gründe gestoppt wird, beschreiben das Pipeline-Ereignisprotokoll und die Aktualisierungsfehlermeldung die Ursache und die Schritte zur Behebung. Befolge diese Schritte und führe die Pipeline erneut durch, um die Migration abzuschließen. Wenn Sie glauben, dass eine Kompatibilitätswarnung ein Fehlalarm ist, beheben Sie das markierte Muster oder kontaktieren Sie den Support von Azure Databricks.

Referenz zu Kompatibilitätsereignissen

Wenn der Kompatibilitätsscan auf einer Pipeline ausgeführt wird, sendet er pro erkanntem Muster ein Ereignis BehaviorChangeInSparkConnectWARN im Pipeline-Ereignislog. Wenn das vorherige erfolgreiche Update Muster erkannt hat, wird die Pipeline erst in eine Umgebungsversion migriert, wenn die Muster adressiert sind.

Jedes Ereignis meldet einen einzelnen Problemcode, der angibt, was erkannt wurde. Um einen Code nachzuschlagen, suchen Sie ihn in der Tabelle "Problemcodes ", wobei jede Zeile mit dem Kategorieabschnitt verknüpft ist, der ein Beispielmuster und die vorgeschlagene Korrektur enthält.

Ereignis-Shape

BehaviorChangeInSparkConnect Ereignisse folgen dem Standardmäßigen Pipelineereignisprotokollschema:

  • event_type ist behavior_change_in_spark_connect.
  • level ist WARN.
  • details enthält das behavior_change_in_spark_connect Objekt, das über ein einzelnes issue Feld verfügt. Der Ausgabewert ist einer der unten aufgeführten Codes.
  • message ist eine lesbare Beschreibung des erkannten Musters.

Problemcodes

Kategorie Problemcode Description
Datenbank- und Katalogmutationen USE_CATALOG_OUTSIDE_QUERY_FUNCTION_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR Der Standardkatalog wurde geändert, nachdem ein DataFrame erstellt wurde. Der vorhandene DataFrame kann Tabellen mithilfe des neuen Standardkatalogs auflösen.
Datenbank- und Katalogmutationen USE_CATALOG_OUTSIDE_QUERY_FUNCTION_COULD_CHANGE_BEHAVIOR USE CATALOG wurde außerhalb einer Funktion aufgerufen, die von einem Rohrleitungsdekoror verziert wurde. Der Standardkatalog kann sich bei nachfolgenden Vorgängen unerwartet ändern.
Datenbank- und Katalogmutationen USE_DATABASE_OUTSIDE_QUERY_FUNCTION_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR Die Standarddatenbank wurde geändert, nachdem ein DataFrame erstellt wurde. Der vorhandene DataFrame kann Tabellen mithilfe der neuen Standarddatenbank auflösen.
Datenbank- und Katalogmutationen USE_DATABASE_OUTSIDE_QUERY_FUNCTION_COULD_CHANGE_BEHAVIOR USE DATABASE wurde außerhalb einer Funktion aufgerufen, die von einem Rohrleitungsdekoror verziert wurde. Die Standarddatenbank kann sich bei nachfolgenden Vorgängen unerwartet ändern.
Eiferige Ausführung innerhalb von Flussfunktionen CHECKPOINT_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED Die Flussfunktion ruft einen Prüfpunktbefehl auf.
Eiferige Ausführung innerhalb von Flussfunktionen CREATE_DATAFRAME_VIEW_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED Die Flussfunktion erstellt eine DataFrame-Ansicht (createOrReplaceTempView oder ähnlich).
Eiferige Ausführung innerhalb von Flussfunktionen CREATE_RESOURCE_PROFILE_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED Die Flussfunktion erstellt ein Ressourcenprofil.
Eiferige Ausführung innerhalb von Flussfunktionen GET_RESOURCES_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED Die Ablauffunktion ruft oder spark.resources eine zugehörige Ressourcen-API auf.
Eiferige Ausführung innerhalb von Flussfunktionen MERGE_INTO_TABLE_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED Die Flussfunktion führt eine zieltabelle aus MERGE INTO .
Eiferige Ausführung innerhalb von Flussfunktionen ML_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED Die Flussfunktion führt einen eifrigen Spark ML-Vorgang aus.
Eiferige Ausführung innerhalb von Flussfunktionen REGISTER_DATA_SOURCE_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED Die Flussfunktion registriert eine Python Datenquelle.
Eiferige Ausführung innerhalb von Flussfunktionen STREAMING_QUERY_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED Die Flussfunktion arbeitet mit einem aktiven Streamingabfragehandle.
Eiferige Ausführung innerhalb von Flussfunktionen STREAMING_QUERY_LISTENER_BUS_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED Die Ablauffunktion registriert oder entfernt einen Streamingabfragelistener.
Eiferige Ausführung innerhalb von Flussfunktionen STREAMING_QUERY_MANAGER_COMMAND_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED Die Ablauffunktion ruft zum Verwalten von Streamingabfragen auf spark.streams .
Eiferige Ausführung innerhalb von Flussfunktionen WRITE_OPERATION_V2_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED Die Flussfunktion führt einen eifrigen DataFrameWriterV2 Vorgang aus.
Eiferige Ausführung innerhalb von Flussfunktionen WRITE_OPERATION_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED Die Flussfunktion führt einen eifrigen DataFrame.write Vorgang aus.
Eiferige Ausführung innerhalb von Flussfunktionen WRITE_STREAM_OPERATION_START_WITHIN_QUERY_FUNCTION_NOT_SUPPORTED Die Flussfunktion startet eine Streamingabfrage (writeStream.start()).
Spark-Konfigurationsmutationen CHANGE_CONF_INSIDE_QUERY_FUNCTION_NOT_SUPPORTED spark.conf.set() oder spark.conf.unset() wurde in einer Funktion aufgerufen, die von einem Pipelines-Dekorateur verziert wurde. Dies wird mit einer Umgebungsversion nicht unterstützt.
Spark-Konfigurationsmutationen SET_CONF_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR spark.conf.set() wurde außerhalb einer Funktion aufgerufen, die von einem Pipelines-Dekorateur nach der Erstellung eines DataFrames eingerichtet wurde. Die Konfigurationsänderung kann sich auf den vorhandenen DataFrame zur Ausführungszeit auswirken.
Spark-Konfigurationsmutationen UNSET_CONF_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR spark.conf.unset() wurde außerhalb einer Funktion aufgerufen, die von einem Pipelines-Dekorateur nach der Erstellung eines DataFrames eingerichtet wurde. Die Konfigurationsänderung kann sich auf den vorhandenen DataFrame zur Ausführungszeit auswirken.
Temporäre Ansichtsersetzungen REPLACE_GLOBAL_TEMP_VIEW_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR Eine globale temporäre Ansicht wurde ersetzt, nachdem ein DataFrame darauf verweist. Der Ersatz kann im vorhandenen DataFrame widerspiegelt werden.
Temporäre Ansichtsersetzungen REPLACE_TEMP_VIEW_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR Eine temporäre Ansicht wurde ersetzt, nachdem ein DataFrame darauf verweist. Der Ersatz kann im vorhandenen DataFrame widerspiegelt werden.
UDF- und UDTF-Mutationen OVERWRITE_SESSION_UDF_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR Eine UDF wurde mit demselben Namen erneut registriert, nachdem ein DataFrame darauf verweist, dass es erstellt wurde. Der vorhandene DataFrame kann die neue UDF-Definition verwenden.
UDF- und UDTF-Mutationen OVERWRITE_SESSION_UDTF_AFTER_DATAFRAME_COULD_CHANGE_BEHAVIOR Ein UDTF wurde mit demselben Namen erneut registriert, nachdem ein DataFrame darauf verweist, dass es erstellt wurde. Der vorhandene DataFrame kann die neue UDTF-Definition verwenden.
UDF- und UDTF-Mutationen UDF_REFERENCES_GLOBAL_VARIABLE_COULD_CHANGE_BEHAVIOR Eine UDF verweist auf eine globale veränderbare Python Variable. Bei einer Umgebungsversion verwendet die UDF den Wert der Variablen zum Zeitpunkt der Definition der UDF, nicht zum Aufrufzeitpunkt.
UDF- und UDTF-Mutationen UDTF_REFERENCES_GLOBAL_VARIABLE_COULD_CHANGE_BEHAVIOR Ein UDTF verweist auf eine globale veränderbare Python Variable. Bei einer Umgebungsversion verwendet udTF den Wert der Variablen zum Zeitpunkt der DEFINITION des UDTF, nicht zum Aufrufzeitpunkt.

Datenbank- und Katalogmutationen

Diese Probleme werden ausgegeben, wenn Pipelinecode die Standarddatenbank oder den Standardkatalog stummschaltet. Mit einer Umgebungsversion erstellt DataFrames, bevor die Mutation Tabellen mithilfe der neuen Datenbank oder des neuen Katalogs auflösen kann.

Beispielmuster, das ein Ereignis auslöst:

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()

Ohne Eine Umgebungsversion dfevents wird aus dem marketing Katalog aufgelöst. Bei einer Umgebungsversion df wird events der sales Katalog aufgelöst.

Vorgeschlagene Lösung: Vollqualifizierte Tabellennamen, sodass die Auflösung nicht vom Standardkatalog oder der Standarddatenbank abhängt, und vermeiden Sie das Ändern des Standardkatalogs oder der Datenbank zwischen der Erstellung und Verwendung von 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-Konfigurationsmutationen

Diese Probleme werden ausgegeben, wenn der Pipelinecode spark-Konfiguration so stummschaltet, dass das DataFrame-Verhalten unter einer Umgebungsversion geändert werden kann.

Beispielmuster, das ein Ereignis auslöst:

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")

Ohne Eine Umgebungsversion verwendet die Umwandlung den Konf-Wert zur Erstellungszeit von DataFrame. Bei einer Umgebungsversion wird die Umwandlung verwendet spark.sql.ansi.enabled=true und kann bei ungültiger Eingabe fehlschlagen.

Vorgeschlagene Lösung: Legen Sie alle erforderlichen Spark-Konfigurationen oben in der Pipelinedatei fest, bevor DataFrame erstellt wird. Verwenden Sie für die Konfiguration pro Abfrage die Einstellung der Pipeline configuration in der Pipelinespezifikation.

Temporäre Ansichtsersetzungen

Diese Probleme werden ausgegeben, wenn Pipelinecode eine temporäre Ansicht ersetzt, nachdem ein DataFrame darauf verweist, dass er erstellt wurde. Mit einer Umgebungsversion kann der vorhandene DataFrame den neuen Ansichtsinhalt widerspiegeln.

Beispielmuster, das ein Ereignis auslöst:

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

Enthält ohne eine Umgebungsversion mytable[(1, "Original Row")]. Enthält eine Umgebungsversion mytable[(2, "Replaced Row")].

Vorgeschlagene Lösung: Erstellen Sie jede temporäre Ansicht einzeln, und ersetzen Sie sie nicht. Wenn Sie mehrere Ansichten mit verwandten Daten benötigen, geben Sie jedem einen eindeutigen Namen.

UDF- und UDTF-Mutationen

Diese Probleme werden ausgegeben, wenn Pipelinecode eine UDF- oder UDTF-Datei so stummschaltet, dass sich das Verhalten unter einer Umgebungsversion ändert.

Beispielmuster, das ein Ereignis auslöst:

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")))

Enthält ohne eine Umgebungsversion my_mv[("alex_b",)]. Enthält eine Umgebungsversion my_mv[("alex_a",)].

Suggested fix: Übergeben Sie Werte als Argumente an die UDF, anstatt sie aus Python Globalen zu erfassen, oder legen Sie die Globale fest, bevor Sie die UDF definieren und sie danach nicht stummschalten.

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")))

Eiferige Ausführung innerhalb von Flussfunktionen

Diese Probleme werden ausgegeben, wenn pipelinecode einen eifrigen Spark-Befehl in einer Funktion ausführt, die von einem Pipelinedekoror (@table, @materialized_viewusw.) eingerichtet wurde. Flussfunktionen werden erwartet, dass ein DataFrame definiert und zurückgegeben wird; Eifrige Befehle, die Daten schreiben, Streamingabfragen verwalten, Ressourcen registrieren oder ML-Vorgänge ausführen, sind in einer Flussfunktion mit einem Umgebungsversionssatz nicht zulässig.

Vorgeschlagene Lösung: Verschieben Sie den eifrigen Vorgang außerhalb der Flussfunktion, und geben Sie stattdessen einen DataFrame aus der Flussfunktion zurück. Nebeneffekte wie das Schreiben in eine Tabelle oder das Starten einer Streamingabfrage gehören außerhalb der Pipelinedefinition; das Pipelinemodul verarbeitet die Materialisierung des DataFrames, der von der Flussfunktion zurückgegeben wird.

Suchen nach Kompatibilitätsereignissen im Ereignisprotokoll

Die folgende Abfrage gibt alle Kompatibilitätsereignisse für eine Pipeline zurück, sortiert zuerst:

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;

So zählen Sie Ereignisse nach Problemcode in den letzten Updates:

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;

Informationen zum Abfragen des Ereignisprotokolls finden Sie unter Abfragen des Ereignisprotokolls.

Weitere Ressourcen

  • Last updated on