Compatibilité des versions de l’environnement

Les pipelines avec une version environment définissent l’exécution Python code via Spark Connect. Cette page explique ce qui est incompatible, ce qui se comporte différemment, et comment Databricks analyse un pipeline à la recherche de motifs affectés.

Limitations

Les versions d’environnement ne sont pas encore compatibles avec toutes les fonctionnalités de pipeline. Une exécution de pipeline avec un jeu de versions d'environnement échoue si le code Python du pipeline effectue l'une des opérations suivantes :

• Mute l’état de session Spark à l’intérieur d’une fonction décorée avec un décorateur de pipelines. Exemples : , spark.conf.set(...)spark.sql("USE CATALOG ...")et createOrReplaceTempView.
• Utilise des API PySpark qui ne sont pas disponibles dans Spark Connect, y compris SparkContext, RDD, SQLContextet toutes les API Py4J. Voir ce qui est pris en charge dans Spark Connect.

Si vous activez une version d’environnement sur un pipeline, la désactivation de la version de l’environnement retourne le pipeline à son état précédent.

Changements de comportement

Spark Connect présente un petit nombre de différences de comportement par rapport au runtime PySpark classique. Consultez Spark Connect et Spark Classic pour obtenir la référence complète. Le scan de compatibilité détecte ces schémas à l’avance et bloque la migration jusqu’à ce qu’ils soient traités, afin que vous puissiez les trouver et les corriger avant qu’ils n’affectent les données de production.

Dans un pipeline, les situations les plus courantes où le comportement peut différer sont les suivants :

• Mutation de la construction et de la session entrelacées du DataFrame
• UDF qui font référence à l’état Python mutable

Mutation de la construction et de la session entrelacées du DataFrame

Lorsqu’un pipeline construit un DataFrame, mute l’état de session Spark (par exemple, modifie le catalogue ou le schéma par défaut, définit une configuration, remplace une vue temporaire ou réinscrit une fonction UDF), puis utilise le DataFrame :

• Sans version d’environnement, le DataFrame utilise l’état de session de pré-mutation .
• Avec une version d’environnement, le DataFrame utilise l’état de session post-mutation .

Par exemple:

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

Sans version d’environnement, mytable contient [(1, "Original Row")]. Avec une version d’environnement, mytable contient [(2, "Replaced Row")].

UDF qui référencent l’état Python mutable

Lorsqu’une fonction UDF fait référence à une variable globale Python dont la valeur change après la définition de la fonction UDF :

• Sans version d’environnement, l’UDF utilise la dernière valeur de la variable.
• Avec une version d’environnement, l’UDF utilise la valeur au moment où l’UDF a été définie.

Par exemple:

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

Sans version d’environnement, my_mv contient [("alex_b",)]. Avec une version d’environnement, my_mv contient [("alex_a",)].

Si un pipeline s’appuie sur l’un ou l’autre modèle, auditez-le avant d’activer une version d’environnement.

Analyse de compatibilité

Le scan de compatibilité trouve des motifs de code dans votre pipeline qui produiraient des résultats différents sous une version d’environnement, vous pouvez donc les corriger avant qu’un pipeline ne soit migré automatiquement. Lorsque l’analyse est activée sur un pipeline :

• Chaque mise à jour émet un BehaviorChangeInSparkConnectWARN événement dans le journal des événements du pipeline par motif détecté.
• Le pipeline n’est pas migré vers une version d’environnement, et vous ne pouvez pas en activer une vous-même tant que tous les avertissements de compatibilité de la mise à jour réussie précédente ne sont pas traités.

Cette vérification ne s’applique pas à un pipeline qui n’a pas de mise à jour précédente, ou qui a déjà une version d’environnement définie.

Activer l’analyse sur un pipeline

Vous pouvez activer l’analyse de compatibilité en ajoutant la configuration du pipelines.environmentVersion.enableCompatibilityScan pipeline. Vous pouvez ajouter une configuration via l’interface utilisateur de l’éditeur de pipeline ou en ajoutant une entrée au json de configuration de pipeline.

Via l’interface utilisateur :

1. Dans l’éditeur de pipeline, cliquez sur Paramètres.
2. Recherchez la section Configuration dans les paramètres du pipeline.
3. Cliquez sur Ajoutez une configuration.
4. Entrez pipelines.environmentVersion.enableCompatibilityScan la clé et true la valeur.
5. Enregistrez les paramètres du pipeline.

Dans le json du pipeline :

Ajoutez l’entrée configuration suivante au bloc :

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

Révisez et résolvez les avertissements de compatibilité

Pour trouver et effacer les motifs qui bloquent une version d’environnement sur votre pipeline :

1. Exécutez le pipeline en mode exécution à sec , puis interrogez le journal d’événements du pipeline pour les BehaviorChangeInSparkConnectWARN événements. Chaque événement signale un modèle détecté. Consultez la référence des événements de compatibilité pour obtenir la liste complète des codes de problème, des exemples de modèles et des correctifs suggérés.
2. Mettez à jour le code du pipeline pour supprimer les motifs détectés suite à la correction suggérée, puis relancez le pipeline.
3. Répétez jusqu’à ce qu’une mise à jour réussie n’émette plus d’événements de compatibilité. Le pipeline peut alors être migré automatiquement, et vous pouvez aussi activer vous-même une version de l’environnement.

Activer une version d’environnement exécute les mêmes vérifications de sécurité pour savoir si Databricks migre automatiquement le pipeline ou si c’est vous qui le définissez environment_version vous-même. Un pipeline avec des avertissements de compatibilité non résolus ne passe pas à une version d’environnement tant que ces avertissements ne sont pas résolus. Si la migration ne peut pas être réalisée en toute sécurité, ou échoue pour une raison quelconque, elle s’arrête avant d’écrire des données et le pipeline continue de fonctionner à son exécution précédente.

Lorsqu’une mise à jour s’arrête pour l’une de ces raisons, le journal des événements du pipeline et le message d’erreur de mise à jour décrivent la cause et les étapes pour la résoudre. Suivez ces étapes et relancez le pipeline pour terminer la migration. Si vous pensez qu’un avertissement de compatibilité est un faux positif, résolvez le motif signalé ou contactez le support Azure Databricks.

Informations de référence sur les événements de compatibilité

Lorsque le balayage de compatibilité s’exécute sur un pipeline, il émet un BehaviorChangeInSparkConnectWARN événement dans le journal d’événements du pipeline par motif détecté. Lorsque la mise à jour réussie précédente détectait des patterns, le pipeline n’était pas migré vers une version de l’environnement tant que les patterns ne sont pas traités.

Chaque événement signale un code de problème unique qui identifie ce qui a été détecté. Pour rechercher un code, recherchez-le dans la table Codes de problème : chaque ligne est liée à la section catégorie qui contient un exemple de modèle et le correctif suggéré.

Forme d’événement

BehaviorChangeInSparkConnect les événements suivent le schéma standard du journal des événements du pipeline :

• event_type a la valeur behavior_change_in_spark_connect.
• level a la valeur WARN.
• details contient l’objet behavior_change_in_spark_connect , qui a un seul issue champ. La valeur du problème est l’un des codes répertoriés ci-dessous.
• message est une description lisible par l’homme du modèle détecté.

Codes de problème

Mutations de base de données et de catalogue

Ces problèmes sont émis lorsque le code de pipeline mute la base de données ou le catalogue par défaut. Avec une version d’environnement, les DataFrames construits avant la mutation peuvent résoudre les tables à l’aide de la nouvelle base de données ou catalogue.

Exemple de modèle qui déclenche un événement :

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

Sans version d’environnement, df résout events à partir du marketing catalogue. Avec une version d’environnement, df résout events à partir du sales catalogue.

Correctif suggéré : Qualifiez entièrement les noms de tables afin que la résolution ne dépende pas du catalogue ou de la base de données par défaut, et évitez de modifier le catalogue ou la base de données par défaut entre la création et l’utilisation de 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()

Mutations de configuration Spark

Ces problèmes sont émis lorsque le code de pipeline mute la configuration Spark de manière à modifier le comportement du DataFrame sous une version d’environnement.

Exemple de modèle qui déclenche un événement :

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

Sans version d’environnement, le cast utilise la valeur conf au moment de la création du DataFrame. Avec une version d’environnement, le cast utilise spark.sql.ansi.enabled=true et peut échouer sur une entrée non valide.

Correctif suggéré : Définissez toutes les configurations Spark requises en haut du fichier de pipeline, avant la création d’un DataFrame. Pour la configuration par requête, utilisez le paramètre du configuration pipeline dans la spécification du pipeline.

Remplacements de vues temporaires

Ces problèmes sont émis lorsque le code du pipeline remplace une vue temporaire après la création d’un DataFrame. Avec une version d’environnement, le DataFrame existant peut refléter le nouveau contenu de l’affichage.

Exemple de modèle qui déclenche un événement :

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

Sans version d’environnement, mytable contient [(1, "Original Row")]. Avec une version d’environnement, mytable contient [(2, "Replaced Row")].

Correctif suggéré : Créez chaque affichage temporaire une seule fois et ne le remplacez pas. Si vous avez besoin de plusieurs vues avec des données associées, attribuez un nom distinct.

Mutations UDF et UDTF

Ces problèmes sont émis lorsque le code de pipeline mute une fonction UDF ou UDTF de manière à modifier le comportement dans une version d’environnement.

Exemple de modèle qui déclenche un événement :

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

Sans version d’environnement, my_mv contient [("alex_b",)]. Avec une version d’environnement, my_mv contient [("alex_a",)].

Suggested fix : Passez des valeurs dans la fonction UDF en tant qu’arguments au lieu de les capturer à partir de Python globals, ou définissez le global avant de définir la fonction UDF et ne la mutez pas ultérieurement.

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

Exécution impatiente dans les fonctions de flux

Ces problèmes sont émis lorsque le code du pipeline exécute une commande Spark impatiente à l’intérieur d’une fonction décorée par un décorateur de pipelines (@table, @materialized_viewetc.). Les fonctions de flux sont censées définir et retourner un DataFrame ; Les commandes qui écrivent des données, gèrent les requêtes de diffusion en continu, inscrivent des ressources ou exécutent des opérations ML ne sont pas autorisées à l’intérieur d’une fonction de flux avec un jeu de versions d’environnement.

Correctif suggéré : Déplacez l’opération impatiente en dehors de la fonction de flux et retournez un DataFrame à partir de la fonction de flux à la place. Les effets secondaires tels que l’écriture dans une table ou le démarrage d’une requête de diffusion en continu appartiennent en dehors de la définition du pipeline ; le moteur de pipeline gère la matérialisation du DataFrame retourné par la fonction de flux.

Rechercher des événements de compatibilité dans le journal des événements

La requête suivante retourne tous les événements de compatibilité d’un pipeline, classés d’abord les plus récents :

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;

Pour compter les événements par code de problème dans les mises à jour récentes :

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;

Pour savoir comment interroger le journal des événements, consultez Interroger le journal des événements.

Ressources additionnelles

• Configurez les versions de l’environnement pour les pipelines — aperçu des fonctionnalités, migration automatique, et comment activer vous-même une version d’environnement.
• Schéma du journal des événements de pipeline : schéma complet du journal des événements de pipeline.
• Journal des événements de pipeline : comment interroger le journal des événements du pipeline.