Compatibiliteit met omgevingsversies

Pijplijnen met een omgevingsversie Python code uitvoeren via Spark Connect. Op deze pagina wordt beschreven wat niet compatibel is, wat zich anders gedraagt, hoe u een pijplijn scant op beïnvloede patronen en hoe u een bestaande pijplijn migreert.

Limitations

Omgevingsversies zijn nog niet compatibel met alle pijplijnfunctionaliteit. Een pijplijnuitvoering met een omgevingsversieset mislukt als de Python code van de pijplijn een van de volgende handelingen uitvoert:

• Muteert de Spark-sessiestatus binnen een functie die is ingericht met een pijplijn decorator. Voorbeelden zijn , spark.conf.set(...)spark.sql("USE CATALOG ...")en createOrReplaceTempView.
• Maakt gebruik van PySpark-API's die niet beschikbaar zijn in Spark Connect, inclusiefSparkContextRDD, SQLContexten eventuele Py4J-API's. Bekijk wat wordt ondersteund in Spark Connect.

Als het inschakelen van een omgevingsversie in een pijplijn ervoor zorgt dat deze mislukt, retourneert het uitschakelen van de omgevingsversie de pijplijn naar de vorige status.

Gedragswijzigingen

Spark Connect heeft een klein aantal gedragsverschillen ten laste van de klassieke PySpark-runtime. Zie Spark Connect versus klassieke Spark voor de volledige referentie. De compatibiliteitsscan detecteert deze patronen van tevoren en blokkeert het inschakelen totdat ze zijn opgelost, zodat u deze kunt vinden en herstellen voordat ze van invloed zijn op productiegegevens.

In een pijplijn zijn de meest voorkomende situaties waarin gedrag kan verschillen:

• Interleaved DataFrame constructie en sessie mutatie
• UDF's die verwijzen naar de onveranderbare status van Python

Interleaved DataFrame constructie en sessie mutatie

Wanneer een pijplijn een DataFrame samenstelt, muteert u vervolgens de Spark-sessiestatus (bijvoorbeeld wijzigt u de standaardcatalogus of het standaardschema, stelt u een configuratie in, vervangt u een tijdelijke weergave of registreert u een UDF opnieuw), gebruikt u vervolgens het DataFrame:

• Zonder een omgevingsversie gebruikt het DataFrame de status van de sessie vóór mutatie .
• Met een omgevingsversie gebruikt het DataFrame de sessiestatus na mutatie .

Voorbeeld:

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

Zonder een omgevingsversie bevat mytable .[(1, "Original Row")] Bevat een omgevingsversie mytable[(2, "Replaced Row")].

UDFs die verwijzen naar de status veranderlijk Python

Wanneer een UDF verwijst naar een Python globale variabele waarvan de waarde verandert nadat de UDF is gedefinieerd:

• Zonder een omgevingsversie gebruikt de UDF de meest recente waarde van de variabele.
• Met een omgevingsversie gebruikt de UDF de waarde op het moment dat de UDF is gedefinieerd.

Voorbeeld:

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

Zonder een omgevingsversie bevat my_mv .[("alex_b",)] Bevat een omgevingsversie my_mv[("alex_a",)].

Als een pijplijn afhankelijk is van een van beide patronen, controleert u deze voordat u een omgevingsversie inschakelt.

Compatibiliteitsscan

Met de compatibiliteitsscan kunt u codepatronen in uw pijplijn vinden die verschillende resultaten opleveren onder een omgevingsversie, voordat u er een inschakelt. De scan is aangemeld. Wanneer de scan is ingeschakeld voor een pijplijn:

• Elke pijplijnuitvoering verzendt één BehaviorChangeInSparkConnectWARN gebeurtenis in het gebeurtenislogboek van de pijplijn per gedetecteerd patroon.
• U kunt geen omgevingsversie inschakelen in de pijplijn totdat u alle compatibiliteitswaarschuwingen van de vorige geslaagde update hebt aangepakt.

Als de scan niet is ingeschakeld, worden er geen gebeurtenissen verzonden en environment_version wordt het inschakelen niet geblokkeerd. Databricks raadt u aan om de scan in te schakelen en gedetecteerde patronen op te lossen voordat u een omgevingsversie in de pijplijn inschakelt.

De scan inschakelen voor een pijplijn

U kunt de compatibiliteitsscan inschakelen door de pipelines.environmentVersion.enableCompatibilityScan pijplijnconfiguratie toe te voegen. U kunt configuratie toevoegen via de gebruikersinterface van de pijplijneditor of door een vermelding toe te voegen aan de JSON voor de pijplijnconfiguratie.

Via de gebruikersinterface:

1. Klik in de pijplijneditor op Instellingen.
2. Zoek de sectie Configuratie in de pijplijninstellingen.
3. Klik op Configuratie toevoegen.
4. Voer pipelines.environmentVersion.enableCompatibilityScan in als de sleutel en true als de waarde.
5. Sla de pijplijninstellingen op.

In de pijplijn-JSON:

Voeg de volgende vermelding toe aan het configuration blok:

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

Aanbevolen werkstroom

1. Schakel de scan in op de pijplijn.
2. Een pijplijnuitvoering activeren.
3. Voer een query uit op het gebeurtenislogboek van de pijplijn voor BehaviorChangeInSparkConnectWARN gebeurtenissen. Zie naslaginformatie over compatibiliteitsevenementen voor de volledige lijst met foutcodes, voorbeeldpatronen en voorgestelde oplossingen.
4. Werk de pijplijncode bij om de gedetecteerde patronen te verwijderen en voer de pijplijn opnieuw uit totdat er geen gebeurtenissen meer worden verzonden.
5. Voeg environment_version toe aan de pijplijn met behulp van een van de methoden in Een omgevingsversie inschakelen in een pijplijn.

Als u denkt dat een compatibiliteitswaarschuwing een fout-positief is en toch wilt inschakelen environment_version , verwijdert u de pipelines.environmentVersion.enableCompatibilityScan vermelding uit de pijplijnconfiguratie om de controle te omzeilen. (Als u de waarde false instelt op is niet toegestaan, moet u de vermelding helemaal verwijderen.)

De voorbereidende controle wordt niet uitgevoerd op pijplijnen die geen eerdere update hebben of op pijplijnen waarvoor al een omgevingsversie is ingesteld.

Een bestaande pijplijn migreren naar omgevingsversies

Als u een bestaande pijplijn wilt migreren die nog geen omgevingsversie gebruikt, volgt u deze end-to-end-werkstroom. U wordt begeleid bij het vinden van codepatronen die zich mogelijk anders gedragen onder Spark Connect, het herstellen ervan en het veilig implementeren van de omgevingsversie.

1. Schakel de compatibiliteitsscan voor de pijplijn in. Schakel de scan in op de pijplijn, zoals beschreven in de compatibiliteitsscan. Dit zorgt ervoor dat gedetecteerde patronen worden weergegeven in het gebeurtenislogboek en wat de voorbereidende controle inschakelt die uw activeringspoging bewaakt.

2. Een pijplijnuitvoering activeren en compatibiliteitsevenementen controleren. Een normale pijplijnupdate activeren. Nadat dit is voltooid, voert u een query uit op het gebeurtenislogboek van de pijplijn voor BehaviorChangeInSparkConnectWARN gebeurtenissen. Elke gebeurtenis rapporteert één gedetecteerd patroon. Zie naslaginformatie over compatibiliteitsevenementen voor de volledige lijst met foutcodes, voorbeeldpatronen en voorgestelde oplossingen.

3. Werk uw pijplijncode bij om gedetecteerde patronen te verhelpen. Werk voor elk gedetecteerd patroon de pijplijncode bij volgens de voorgestelde oplossing. Na elke wijziging activeert u een andere pijplijnupdate en controleert u of de bijbehorende gebeurtenissen niet meer worden weergegeven. Herhaal dit totdat in het gebeurtenislogboek geen compatibiliteitsevenementen meer worden weergegeven voor een geslaagde update.

4. Schakel de omgevingsversie in de pijplijn in. Nadat de meest recente geslaagde update geen compatibiliteitsgebeurtenissen heeft, voegt u deze toe environment_version aan de pijplijn met behulp van de gebruikersinterface, API of bundel, zoals beschreven in een omgevingsversie inschakelen in een pijplijn. De volgende update wordt uitgevoerd met Spark Connect en de vastgemaakte Python taalversie en vooraf geïnstalleerde bibliotheken.

   Als de update mislukt omdat er nog compatibiliteitswaarschuwingen bestaan, verwijdert u de optie , gaat u terug naar stap 2 en lost u de resterende waarschuwingen op voordat u het environment_versionopnieuw probeert.

5. Controleer de migratie. Nadat de eerste update met de omgevingsversie is voltooid, controleert u:

   • De create_update gebeurtenis in het gebeurtenislogboek wordt environment_version ingesteld op de verwachte waarde.
   • De pijplijn produceert de verwachte gegevens en er worden geen nieuwe fout gebeurtenissen weergegeven.
   • Spot-check downstream-tabellen op eventuele subtiele gedragsverschillen die worden beschreven in gedragswijzigingen.

Terugdraaien

Als de pijplijn zich na de migratie misdraagt, verwijdert u de environment_version uit de pijplijninstellingen. De volgende update wordt uitgevoerd met de vorige Python runtimeconfiguratie. Gebruik de teruggedraaide uitvoering om fouten op te sporen en herhaal vervolgens de migratie van stap 2 nadat u het probleem hebt geïdentificeerd en opgelost.

Naslaginformatie over compatibiliteitsevenementen

Wanneer de compatibiliteitsscan is ingeschakeld voor een pijplijn, verzendt SDP één BehaviorChangeInSparkConnectWARN gebeurtenis in het gebeurtenislogboek van de pijplijn per gedetecteerd patroon. Wanneer de scan is ingeschakeld en de vorige geslaagde update patronen heeft gedetecteerd, blokkeert environment_version SDP ook het inschakelen totdat de patronen zijn opgelost.

Elke gebeurtenis rapporteert één probleemcode die aangeeft wat er is gedetecteerd. Als u een code wilt opzoeken, zoekt u deze in de tabel Probleemcodes : elke rij wordt gekoppeld aan de categoriesectie met een voorbeeldpatroon en de voorgestelde oplossing.

Gebeurtenisshape

BehaviorChangeInSparkConnect gebeurtenissen volgen het standaardschema van het gebeurtenislogboek van de pijplijn:

• event_type is behavior_change_in_spark_connect.
• level is WARN.
• details bevat het behavior_change_in_spark_connect object, dat één issue veld heeft. De waarde van het probleem is een van de onderstaande codes.
• message is een door mensen leesbare beschrijving van het gedetecteerde patroon.

Codes voor probleem

Database- en catalogusmutaties

Deze problemen worden verzonden wanneer pijplijncode de standaarddatabase of -catalogus muteert. Met een omgevingsversie kunnen DataFrames die zijn samengesteld voordat de mutatie tabellen oplost met behulp van de nieuwe database of catalogus.

Voorbeeldpatroon waarmee een gebeurtenis wordt geactiveerd:

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

Zonder een omgevingsversie eventsmarketing wordt df de catalogus omgezet. Met een omgevingsversie events wordt df deze omgezet vanuit de sales catalogus.

Voorgestelde oplossing: Volledig in aanmerking komende tabelnamen, zodat oplossing niet afhankelijk is van de standaardcatalogus of database, en vermijd het wijzigen van de standaardcatalogus of database tussen het maken en gebruiken van 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-configuratiemutaties

Deze problemen worden gegenereerd wanneer de Spark-configuratie door pijplijncode wordt gedempt op manieren die het gedrag van DataFrame onder een omgevingsversie kunnen wijzigen.

Voorbeeldpatroon waarmee een gebeurtenis wordt geactiveerd:

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

Zonder een omgevingsversie gebruikt de cast de conf-waarde bij het maken van DataFrame. Met een omgevingsversie gebruikt spark.sql.ansi.enabled=true de cast mogelijk ongeldige invoer.

Voorgestelde oplossing: Stel alle vereiste Spark-configuraties boven aan het pijplijnbestand in voordat een DataFrame wordt gemaakt. Gebruik voor configuratie per query de instelling van configuration de pijplijn in de pijplijnspecificatie.

Tijdelijke weergavevervangingen

Deze problemen worden verzonden wanneer pijplijncode een tijdelijke weergave vervangt nadat een DataFrame naar deze is gemaakt. Met een omgevingsversie kan het bestaande DataFrame de inhoud van de nieuwe weergave weerspiegelen.

Voorbeeldpatroon waarmee een gebeurtenis wordt geactiveerd:

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

Zonder een omgevingsversie bevat mytable .[(1, "Original Row")] Bevat een omgevingsversie mytable[(2, "Replaced Row")].

Voorgestelde oplossing: Maak elke tijdelijke weergave één keer en vervang deze niet. Als u meerdere weergaven met gerelateerde gegevens nodig hebt, geeft u elk een afzonderlijke naam.

UDF- en UDTF-mutaties

Deze problemen worden verzonden wanneer pijplijncode een UDF of UDTF muteert op manieren die gedrag wijzigen onder een omgevingsversie.

Voorbeeldpatroon waarmee een gebeurtenis wordt geactiveerd:

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

Zonder een omgevingsversie bevat my_mv .[("alex_b",)] Bevat een omgevingsversie my_mv[("alex_a",)].

Suggested fix: geef waarden door aan de UDF als argumenten in plaats van ze vast te leggen uit Python globals, of stel de globale waarde in voordat u de UDF definieert en deze later niet dempt.

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

Gretige uitvoering binnen stroomfuncties

Deze problemen worden gegenereerd wanneer pijplijncode een gretige Spark-opdracht uitvoert in een functie die is ingericht door een pijplijn decorator (@table, @materialized_viewenzovoort). Flow-functies zullen naar verwachting een DataFrame definiëren en retourneren; gretige opdrachten die gegevens schrijven, streamingquery's beheren, resources registreren of ML-bewerkingen uitvoeren, zijn niet toegestaan in een stroomfunctie met een omgevingsversieset.

Voorgestelde oplossing: Verplaats de gretige bewerking buiten de stroomfunctie en retourneer in plaats daarvan een DataFrame van de stroomfunctie. Neveneffecten zoals schrijven naar een tabel of het starten van een streamingquery horen buiten de pijplijndefinitie; de pijplijnengine verwerkt de materialisatie van het DataFrame dat door de stroomfunctie wordt geretourneerd.

Compatibiliteitsevenementen zoeken in het gebeurtenislogboek

De volgende query retourneert alle compatibiliteitsevenementen voor een pijplijn, die als eerste de meest recente is gesorteerd:

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;

Gebeurtenissen tellen op uitgiftecode voor recente 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;

Zie Query's uitvoeren op het gebeurtenislogboek voor informatie over het uitvoeren van query's op het gebeurtenislogboek.

Zie ook

• Omgevingsversies voor pijplijnen configureren : overzicht van functies, hoe u een omgevingsversie inschakelt.
• Schema voor gebeurtenislogboek van pijplijn: volledig schema voor gebeurtenislogboeken van pijplijn.
• Gebeurtenislogboek van pijplijn : het uitvoeren van query's op het gebeurtenislogboek van de pijplijn.