Compatibilidade de versão do ambiente

Pipelines com uma versão environment definir Python código de execução por meio do Spark Connect. Esta página aborda o que é incompatível, o que se comporta de forma diferente e como o Databricks escaneia um pipeline em busca de padrões afetados.

Limitações

As versões de ambiente ainda não são compatíveis com todas as funcionalidades de pipeline. Uma execução de pipeline com um conjunto de versões de ambiente falhará se o código de Python do pipeline fizer o seguinte:

• Modifica o estado da sessão spark dentro de uma função decorada com um decorador de pipelines. Exemplos incluem spark.conf.set(...), spark.sql("USE CATALOG ...")e createOrReplaceTempView.
• Usa APIs do PySpark que não estão disponíveis no Spark Connect, incluindo SparkContext, RDDe SQLContextqualquer APIs Py4J. Veja o que tem suporte no Spark Connect.

Se a habilitação de uma versão de ambiente em um pipeline causar falha, desabilitar a versão do ambiente retornará o pipeline ao estado anterior.

Alterações de comportamento

O Spark Connect tem um pequeno número de diferenças de comportamento em relação ao runtime clássico do PySpark. Consulte Spark Connect vs. Spark clássico para obter a referência completa. A varredura de compatibilidade detecta esses padrões antecipadamente e bloqueia a migração até que sejam resolvidos, para que você possa encontrá-los e corrigi-los antes que afetem os dados de produção.

Em um pipeline, as situações mais comuns em que o comportamento pode ser diferente são:

• Construção de DataFrame intercalada e mutação de sessão
• UDFs que fazem referência ao estado Python mutável

Construção de DataFrame intercalada e mutação de sessão

Quando um pipeline constrói um DataFrame, depois altera o estado da sessão do Spark (por exemplo, altera o catálogo ou esquema padrão, define uma configuração, substitui uma exibição temporária ou registra novamente um UDF) e usa o DataFrame:

• Sem uma versão de ambiente, o DataFrame usa o estado da sessão de pré-mutação .
• Com uma versão de ambiente, o DataFrame usa o estado da sessão pós-mutação .

Por exemplo:

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

Sem uma versão de ambiente, mytable contém [(1, "Original Row")]. Com uma versão de ambiente, mytable contém [(2, "Replaced Row")].

UDFs que fazem referência ao estado de Python mutável

Quando uma UDF faz referência a uma variável global Python cujo valor é alterado após a definição da UDF:

• Sem uma versão de ambiente, a UDF usa o valor mais recente da variável.
• Com uma versão de ambiente, a UDF usa o valor no momento em que a UDF foi definida.

Por exemplo:

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

Sem uma versão de ambiente, my_mv contém [("alex_b",)]. Com uma versão de ambiente, my_mv contém [("alex_a",)].

Se um pipeline depender de qualquer padrão, audite-o antes de habilitar uma versão do ambiente.

Verificação de compatibilidade

A varredura de compatibilidade encontra padrões de código no seu pipeline que produziriam resultados diferentes em uma versão do ambiente, para que você possa corrigi-los antes que o pipeline seja migrado automaticamente. Quando a verificação está habilitada em um pipeline:

• Cada atualização emite um BehaviorChangeInSparkConnectWARN evento no registro de eventos do pipeline por padrão detectado.
• O pipeline não é migrado para uma versão do ambiente, e você não pode habilitar uma por conta própria, até que todos os avisos de compatibilidade da atualização anterior bem-sucedida sejam resolvidos.

Essa verificação não se aplica a um pipeline que não tem atualização prévia, ou que já tem uma versão do ambiente definida.

Habilitar a verificação em um pipeline

Você pode habilitar a verificação de compatibilidade adicionando a configuração do pipelines.environmentVersion.enableCompatibilityScan pipeline. Você pode adicionar a configuração por meio da interface do usuário do editor de pipeline ou adicionando uma entrada ao JSON de configuração de pipeline.

Por meio da interface do usuário:

1. No editor de pipeline, clique em Configurações.
2. Localize a seção Configuração nas configurações do pipeline.
3. Clique Adicionar configuração.
4. Insira pipelines.environmentVersion.enableCompatibilityScan como a chave e true como o valor.
5. Salve as configurações de pipeline.

No JSON do pipeline:

Adicione a seguinte entrada ao configuration bloco:

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

Revise e resolva avisos de compatibilidade

Para encontrar e limpar os padrões que bloqueiam uma versão do ambiente no seu pipeline:

1. Execute o pipeline em modo de execução a seco e depois consulte o log de eventos do pipeline para BehaviorChangeInSparkConnectWARN eventos. Cada evento relata um padrão detectado. Consulte a referência de eventos de compatibilidade para a lista completa de códigos de problema, padrões de exemplo e correções sugeridas.
2. Atualize o código do pipeline para remover os padrões detectados após a correção sugerida e execute o pipeline novamente.
3. Repita até que uma atualização bem-sucedida não emita mais eventos de compatibilidade. O pipeline pode então ser migrado automaticamente, e você também pode ativar uma versão do ambiente por conta própria.

Ativar uma versão do ambiente executa as mesmas verificações de segurança, seja o Databricks migrar o pipeline automaticamente ou se você mesmo configura environment_version . Um pipeline com avisos de compatibilidade não resolvidos não migra para uma versão do ambiente até que os avisos sejam resolvidos. Se a migração não puder ser concluída com segurança, ou falhar por qualquer motivo, ela para antes de gravar qualquer dado e o pipeline continua rodando em seu tempo de execução anterior.

Quando uma atualização para por um desses motivos, o log de eventos do pipeline e a mensagem de erro de atualização descrevem a causa e os passos para resolvê-la. Siga esses passos e execute o pipeline novamente para completar a migração. Se você acredita que um aviso de compatibilidade é um falso positivo, resolva o padrão sinalizado ou entre em contato com o suporte do Azure Databricks.

Referência de eventos de compatibilidade

Quando a varredura de compatibilidade roda em um pipeline, ela emite um BehaviorChangeInSparkConnectWARN evento no log de eventos do pipeline por padrão detectado. Quando a atualização bem-sucedida anterior detectou algum padrão, o pipeline não é migrado para uma versão do ambiente até que os padrões sejam corrigidos.

Cada evento relata um único código de problema que identifica o que foi detectado. Para pesquisar um código, localize-o na tabela Códigos de problema – cada linha vincula à seção de categoria que contém um padrão de exemplo e a correção sugerida.

Forma do evento

BehaviorChangeInSparkConnect os eventos seguem o esquema de log de eventos de pipeline padrão:

• event_type é behavior_change_in_spark_connect.
• level é WARN.
• details contém o behavior_change_in_spark_connect objeto, que tem um único issue campo. O valor do problema é um dos códigos listados abaixo.
• message é uma descrição legível pelo ser humano do padrão detectado.

Emitir códigos

Mutações de banco de dados e catálogo

Esses problemas são emitidos quando o código do pipeline altera o banco de dados ou catálogo padrão. Com uma versão de ambiente, dataframes construídos antes da mutação podem resolver tabelas usando o novo banco de dados ou catálogo.

Padrão de exemplo que dispara um evento:

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

Sem uma versão de ambiente, df resolve events do marketing catálogo. Com uma versão de ambiente, df é resolvida events a partir do sales catálogo.

Correção sugerida: Qualifique totalmente os nomes de tabela para que a resolução não dependa do catálogo ou banco de dados padrão e evite alterar o catálogo ou o banco de dados padrão entre a criação e o uso do 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()

Mutações de configuração do Spark

Esses problemas são emitidos quando o código de pipeline altera a configuração do Spark de maneiras que podem alterar o comportamento do DataFrame em uma versão de ambiente.

Padrão de exemplo que dispara um evento:

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

Sem uma versão de ambiente, a conversão usa o valor de configuração na hora de criação do DataFrame. Com uma versão de ambiente, a conversão usa spark.sql.ansi.enabled=true e pode falhar na entrada inválida.

Correção sugerida: Defina todas as configurações necessárias do Spark na parte superior do arquivo de pipeline, antes que qualquer DataFrame seja criado. Para a configuração por consulta, use a configuração do configuration pipeline na especificação do pipeline.

Substituições de exibição temporária

Esses problemas são emitidos quando o código de pipeline substitui uma exibição temporária depois que um DataFrame que faz referência a ele foi criado. Com uma versão de ambiente, o DataFrame existente pode refletir o novo conteúdo de exibição.

Padrão de exemplo que dispara um evento:

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

Sem uma versão de ambiente, mytable contém [(1, "Original Row")]. Com uma versão de ambiente, mytable contém [(2, "Replaced Row")].

Correção sugerida: Crie cada exibição temporária uma única vez e não a substitua. Se você precisar de várias exibições com dados relacionados, dê a cada um um nome distinto.

Mutações UDF e UDTF

Esses problemas são emitidos quando o código de pipeline altera uma UDF ou UDTF de maneiras que alteram o comportamento em uma versão de ambiente.

Padrão de exemplo que dispara um evento:

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

Sem uma versão de ambiente, my_mv contém [("alex_b",)]. Com uma versão de ambiente, my_mv contém [("alex_a",)].

Suggested fix: Pass values into the UDF as arguments instead of capture them from Python globals, or set the global before de define the UDF and do not mutate it afterward.

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

Execução ansiosa dentro de funções de fluxo

Esses problemas são emitidos quando o código de pipeline executa um comando Spark ansioso dentro de uma função decorada por um decorador de pipelines (@table, @materialized_viewetc.). Espera-se que as funções de fluxo definam e retornem um DataFrame; Comandos ansiosos que gravam dados, gerenciam consultas de streaming, registram recursos ou executam operações de ML não são permitidos dentro de uma função de fluxo com um conjunto de versões de ambiente.

Correção sugerida: Mova a operação ansiosa para fora da função de fluxo e retorne um DataFrame da função de fluxo. Efeitos colaterais, como gravar em uma tabela ou iniciar uma consulta de streaming, pertencem fora da definição do pipeline; o mecanismo de pipeline manipula a materialização do DataFrame retornado pela função de fluxo.

Localizar eventos de compatibilidade no log de eventos

A consulta a seguir retorna todos os eventos de compatibilidade de um pipeline, ordenados primeiro:

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;

Para contar eventos por código de emissão em atualizações recentes:

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;

Para saber como consultar o log de eventos, consulte Consultar o log de eventos.

Recursos adicionais

• Configure versões do ambiente para pipelines — visão geral de recursos, migração automática e como habilitar uma versão do ambiente por conta própria.
• Esquema de log de eventos de pipeline – esquema de log de eventos de pipeline completo.
• Log de eventos do pipeline – como consultar o log de eventos do pipeline.