Miljöversionskompatibilitet

Pipelines med en environment version ställa in kör Python kod via Spark Connect. Denna sida täcker vad som är inkompatibelt, vad som beter sig annorlunda och hur Databricks skannar en pipeline efter påverkade mönster.

Limitations

Miljöversioner är ännu inte kompatibla med alla pipelinefunktioner. En pipelinekörning med en miljöversionsuppsättning misslyckas om pipelinens Python kod gör något av följande:

• Muterar Spark-sessionstillståndet i en funktion som är dekorerad med en pipelinesdekoratör. Exempel är spark.conf.set(...), spark.sql("USE CATALOG ...")och createOrReplaceTempView.
• Använder PySpark-API:er som inte är tillgängliga i Spark Connect, inklusive SparkContext, RDD, SQLContextoch eventuella Py4J-API:er. Se Vad stöds i Spark Connect.

Om det inte går att aktivera en miljöversion på en pipeline returnerar inaktivering av miljöversionen pipelinen till dess tidigare tillstånd.

Beteendeförändringar

Spark Connect har ett litet antal beteendeskillnader jämfört med den klassiska PySpark-körningen. Se Spark Connect jämfört med klassiska Spark för den fullständiga referensen. Kompatibilitetsskanningen upptäcker dessa mönster i förväg och blockerar migrering tills de åtgärdas, så du kan hitta och åtgärda dem innan de påverkar produktionsdata.

I en pipeline är de vanligaste situationerna där beteendet kan skilja sig åt:

• Interleaved DataFrame konstruktion och session mutation
• UDF:er som refererar till föränderligt Python tillstånd

Interleaved DataFrame konstruktion och session mutation

När en pipeline skapar en DataFrame, muterar sedan Spark-sessionstillståndet (till exempel ändrar standardkatalogen eller schemat, anger en konfiguration, ersätter en temporär vy eller omregistrerar en UDF) använder sedan DataFrame:

• Utan en miljöversion använder DataFrame sessionstillståndet före mutationen .
• Med en miljöversion använder DataFrame sessionstillståndet efter mutationen .

Ett exempel:

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

Utan en miljöversion mytable innehåller [(1, "Original Row")]. Med en miljöversion mytable innehåller [(2, "Replaced Row")].

UDF:er som refererar till föränderligt Python tillstånd

När en UDF refererar till en Python global variabel vars värde ändras efter att UDF har definierats:

• Utan en miljöversion använder UDF det senaste värdet för variabeln.
• Med en miljöversion använder UDF värdet vid den tidpunkt då UDF definierades.

Ett exempel:

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

Utan en miljöversion my_mv innehåller [("alex_b",)]. Med en miljöversion my_mv innehåller [("alex_a",)].

Om en pipeline förlitar sig på något av mönstret granskar du det innan du aktiverar en miljöversion.

Kompatibilitetsgenomsökning

Kompatibilitetsskanningen hittar kodmönster i din pipeline som skulle ge olika resultat under en miljöversion, så du kan åtgärda dem innan en pipeline migreras automatiskt. När genomsökningen är aktiverad på en pipeline:

• Varje uppdatering sänder en BehaviorChangeInSparkConnectWARN händelse i pipelinens händelselogg per upptäckt mönster.
• Pipelinen migreras inte till en miljöversion, och du kan inte aktivera en själv förrän alla kompatibilitetsvarningar från föregående lyckade uppdatering är åtgärdade.

Denna kontroll gäller inte för en pipeline som inte har någon tidigare uppdatering, eller som redan har en miljöversion satt.

Aktivera genomsökningen på en pipeline

Du kan aktivera kompatibilitetsgenomsökningen pipelines.environmentVersion.enableCompatibilityScan genom att lägga till pipelinekonfigurationen. Du kan lägga till konfiguration via pipelineredigerarens användargränssnitt eller genom att lägga till en post i pipelinekonfigurationens JSON.

Via användargränssnittet:

1. Klicka på Inställningar i pipelineredigeraren.
2. Leta reda på avsnittet Konfiguration i pipelineinställningar.
3. Klicka på Lägg till konfiguration.
4. Ange pipelines.environmentVersion.enableCompatibilityScan som nyckel och true som värde.
5. Spara pipelineinställningarna.

I pipelinens JSON:

Lägg till följande post i configuration blocket:

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

Granska och lösa kompatibilitetsvarningar

För att hitta och rensa de mönster som blockerar en miljöversion i din pipeline:

1. Kör pipelinen i torrkörningsläge och sök sedan pipelinens händelselogg efter BehaviorChangeInSparkConnectWARN händelser. Varje händelse rapporterar ett identifierat mönster. Se Referens för kompatibilitetshändelser för den fullständiga listan över problemkoder, exempelmönster och föreslagna korrigeringar.
2. Uppdatera pipelinekoden för att ta bort de upptäckta mönstren enligt den föreslagna lösningen och kör pipelinen igen.
3. Upprepa tills en lyckad uppdatering inte ger några fler kompatibilitetshändelser. Pipelinen kan då migreras automatiskt, och du kan också aktivera en miljöversion själv.

Att aktivera en miljöversion kör samma säkerhetskontroller oavsett om Databricks migrerar pipelinen automatiskt eller om du ställer in environment_version dig själv. En pipeline med olösta kompatibilitetsvarningar flyttar inte till en miljöversion förrän varningarna har lösts. Om migreringen inte kan slutföras säkert, eller misslyckas av någon anledning, stannar den innan någon data skrivs och pipelinen fortsätter köra på sin tidigare runtime.

När en uppdatering stoppas av någon av dessa skäl beskriver pipeline-händelseloggen och uppdateringsfelmeddelandet orsaken och stegen för att lösa den. Följ dessa steg och kör pipelinen igen för att slutföra migreringen. Om du tror att en kompatibilitetsvarning är en falsk positiv, lös det flaggade mönstret eller kontakta Azure Databricks support.

Referens för kompatibilitetshändelser

När kompatibilitetsskanningen körs på en pipeline sänder den en BehaviorChangeInSparkConnectWARN händelse i pipelinens händelselogg per detekterat mönster. När den föregående lyckade uppdateringen upptäckte några mönster migreras inte pipelinen till en miljöversion förrän mönstren är adresserade.

Varje händelse rapporterar en enda problemkod som identifierar vad som har identifierats. Om du vill leta upp en kod hittar du den i tabellen Problemkoder – varje rad länkar till kategoriavsnittet som innehåller ett exempelmönster och den föreslagna korrigeringen.

Händelseform

BehaviorChangeInSparkConnect händelser följer standardschemat för händelseloggen för pipeline:

• event_type är behavior_change_in_spark_connect.
• level är WARN.
• details innehåller objektet behavior_change_in_spark_connect , som har ett enda issue fält. Problemvärdet är en av de koder som anges nedan.
• message är en läsbar beskrivning av det identifierade mönstret.

Problemkoder

Databas- och katalogmutationer

Dessa problem genereras när pipelinekoden muterar standarddatabasen eller katalogen. Med en miljöversion kan DataFrames som skapats före mutationen lösa tabeller med hjälp av den nya databasen eller katalogen.

Exempelmönster som utlöser en händelse:

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

Utan en miljöversion df löses events från marketing katalogen. Med en miljöversion df löses events från sales katalogen.

Föreslagen korrigering: Fullständigt kvalificerade tabellnamn så att lösningen inte är beroende av standardkatalogen eller databasen, och undvik att ändra standardkatalogen eller databasen mellan skapande och användning av 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-konfigurationsmutationer

Dessa problem genereras när pipelinekoden muterar Spark-konfigurationen på sätt som kan ändra DataFrame-beteendet under en miljöversion.

Exempelmönster som utlöser en händelse:

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

Utan en miljöversion använder rollbesättningen konfigurationsvärdet vid skapandetiden för DataFrame. Med en miljöversion använder spark.sql.ansi.enabled=true casten och kan misslyckas med ogiltiga indata.

Föreslagen korrigering: Ställ in alla nödvändiga Spark-konfigurationer överst i pipelinefilen innan någon DataFrame skapas. För konfiguration per fråga använder du pipelinens configuration inställning i pipelinespecifikationen.

Temporära vyersättningar

Dessa problem genereras när pipelinekoden ersätter en tillfällig vy efter att en DataFrame-referens har skapats. Med en miljöversion kan den befintliga DataFrame återspegla det nya visningsinnehållet.

Exempelmönster som utlöser en händelse:

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

Utan en miljöversion mytable innehåller [(1, "Original Row")]. Med en miljöversion mytable innehåller [(2, "Replaced Row")].

Föreslagen korrigering: Skapa varje temporär vy en gång och ersätt den inte. Om du behöver flera vyer med relaterade data ger du var och en ett distinkt namn.

UDF- och UDTF-mutationer

Dessa problem genereras när pipelinekod muterar en UDF eller UDTF på sätt som ändrar beteende under en miljöversion.

Exempelmönster som utlöser en händelse:

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

Utan en miljöversion my_mv innehåller [("alex_b",)]. Med en miljöversion my_mv innehåller [("alex_a",)].

Suggested fix: Skicka värden till UDF som argument i stället för att samla in dem från Python globaler, eller ange det globala innan du definierar UDF och mutera det inte efteråt.

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

Ivrig körning i flödesfunktioner

Dessa problem genereras när pipelinekoden utför ett ivrigt Spark-kommando i en funktion som är dekorerad av en pipelines-dekoratör (@table, @materialized_viewosv.). Flödesfunktioner förväntas definiera och returnera en DataFrame. ivriga kommandon som skriver data, hanterar strömmande frågor, registrerar resurser eller kör ML-åtgärder tillåts inte i en flödesfunktion med en miljöversionsuppsättning.

Föreslagen korrigering: Flytta den ivriga åtgärden utanför flödesfunktionen och returnera en DataFrame från flödesfunktionen i stället. Biverkningar som att skriva till en tabell eller starta en direktuppspelningsfråga hör utanför pipelinedefinitionen. pipelinemotorn hanterar materialisering av dataramen som returneras av flödesfunktionen.

Hitta kompatibilitetshändelser i händelseloggen

Följande fråga returnerar alla kompatibilitetshändelser för en pipeline, ordnade senast först:

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;

Så här räknar du händelser efter problemkod för de senaste uppdateringarna:

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;

Information om hur du kör frågor mot händelseloggen finns i Fråga efter händelseloggen.

Ytterligare resurser

• Konfigurera miljöversioner för pipelines — funktionsöversikt, automatisk migrering och hur du aktiverar en miljöversion själv.
• Schema för pipelinehändelselogg – fullständigt schema för pipelinehändelseloggar.
• Händelselogg för pipeline – så här kör du frågor mot händelseloggen för pipelinen.