Ortam sürümü uyumluluğu

environment version kümesine sahip işlem hatları Python kodu Spark Connect aracılığıyla çalıştırır. Bu sayfada uyumlu olmayanlar, farklı davranan davranışlar, etkilenen desenler için işlem hattını tarama ve mevcut işlem hattını geçirme işlemleri anlatılır.

Sınırlamalar

Ortam sürümleri henüz tüm işlem hattı işlevleriyle uyumlu değildir. İşlem hattının Python kodu aşağıdakilerden birini yaparsa ortam sürümü kümesiyle işlem hattı çalıştırması başarısız olur:

• İşlem hatları dekoratörüyle dekore edilmiş bir işlevin içinde Spark oturum durumunu sessize alır. Örnek olarak spark.conf.set(...), spark.sql("USE CATALOG ...")ve createOrReplaceTempViewverilebilir.
• , SparkContext, RDDve tüm Py4J API'leri dahil olmak üzere SQLContextSpark Connect'te kullanılamayan PySpark API'lerini kullanır. Bkz . Spark Connect'te desteklenenler.

bir işlem hattında ortam sürümünün etkinleştirilmesi başarısız olmasına neden oluyorsa, ortam sürümünün devre dışı bırakılması işlem hattını önceki durumuna döndürür.

Davranış değişiklikleri

Spark Connect'in klasik PySpark çalışma zamanından az sayıda davranış farkı vardır. Tam başvuru için bkz . Spark Connect ve klasik Spark . Uyumluluk taraması bu desenleri önceden algılar ve ele alınana kadar etkinleştirmeyi engeller, böylece üretim verilerini etkilemeden önce bunları bulup düzeltebilirsiniz.

İşlem hattında, davranışın farklı olabileceği en yaygın durumlar şunlardır:

• Interleaved DataFrame yapısı ve oturum mutasyonu
• Değiştirilebilir Python durumuna başvuran UDFs

Interleaved DataFrame yapısı ve oturum mutasyonu

İşlem hattı bir DataFrame oluştururken Spark oturum durumunu sessize alır (örneğin, varsayılan kataloğu veya şemayı değiştirir, yapılandırma ayarlar, geçici görünümü değiştirir veya bir UDF'yi yeniden kaydeder), ardından DataFrame'i kullanır:

• Ortam sürümü olmadan DataFrame , mutasyon öncesi oturum durumunu kullanır.
• Ortam sürümüyle DataFrame , mutasyon sonrası oturum durumunu kullanır.

Örneğin:

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

Ortam sürümü olmadan içerir mytable[(1, "Original Row")]. Ortam sürümü ile içerir mytable[(2, "Replaced Row")].

Değiştirilebilir Python durumuna başvuran UDF'ler

UDF, UDF tanımlandıktan sonra değeri değişen bir Python genel değişkenine başvurduğunda:

• Ortam sürümü olmadan UDF, değişkenin en son değerini kullanır.
• Ortam sürümüyle UDF, UDF tanımlandığı sırada değerini kullanır.

Örneğin:

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

Ortam sürümü olmadan içerir my_mv[("alex_b",)]. Ortam sürümü ile içerir my_mv[("alex_a",)].

İşlem hattı iki desenden birini kullanıyorsa, ortam sürümünü etkinleştirmeden önce bunu denetleyin.

Uyumluluk taraması

Uyumluluk taraması, işlem hattınızda ortam sürümü altında farklı sonuçlar üretebilecek kod desenlerini etkinleştirmeden önce bulmanıza yardımcı olur. Tarama kabul edilir. İşlem hattında tarama etkinleştirildiğinde:

• Her işlem hattı çalıştırması, algılanan desen başına işlem hattı olay günlüğünde bir BehaviorChangeInSparkConnectWARN olay yayar.
• Önceki başarılı güncelleştirmeden gelen tüm uyumluluk uyarılarını ele alıncaya kadar işlem hattında bir ortam sürümünü etkinleştiremezsiniz.

Tarama etkin değilse hiçbir olay gösterilmez ve environment_version etkinleştirme engellenmez. Databricks, işlem hattında ortam sürümünü etkinleştirmeden önce taramanın etkinleştirilmesini ve algılanan desenlerin çözülmesini önerir.

İşlem hattında taramayı etkinleştirme

İşlem hattı yapılandırmasını ekleyerek uyumluluk taramasını pipelines.environmentVersion.enableCompatibilityScan etkinleştirebilirsiniz. İşlem hattı düzenleyicisi kullanıcı arabirimi aracılığıyla veya işlem hattı yapılandırması JSON'sine bir giriş ekleyerek yapılandırma ekleyebilirsiniz.

Kullanıcı arabirimi aracılığıyla:

1. İşlem hattı düzenleyicisinde Ayarlar'a tıklayın.
2. İşlem hattı ayarlarında Yapılandırma bölümünü bulun.
3. Yapılandırma ekleyin.
4. Anahtar olarak ve pipelines.environmentVersion.enableCompatibilityScan değer olarak girintrue.
5. İşlem hattı ayarlarını kaydedin.

İşlem hattı JSON'unda:

Aşağıdaki girdiyi bloğuna configuration ekleyin:

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

Önerilen iş akışı

1. İşlem hattında taramayı etkinleştirin.
2. İşlem hattını çalıştırmayı tetikleyin.
3. Olaylar için BehaviorChangeInSparkConnectWARN. Sorun kodlarının, örnek desenlerin ve önerilen düzeltmelerin tam listesi için uyumluluk olayları başvurusuna bakın.
4. Algılanan desenleri kaldırmak için işlem hattı kodunu güncelleştirin ve başka olay yayılmayana kadar işlem hattını yeniden çalıştırın.
5. İşlem hattında environment_version yöntemlerden birini kullanarak işlem hattına ekleyin.

Uyumluluk uyarısının hatalı pozitif olduğunu düşünüyorsanız ve yine de etkinleştirmek environment_version istiyorsanız, denetimi atlamak için girişi işlem hattı yapılandırmasından kaldırın pipelines.environmentVersion.enableCompatibilityScan . (Değerin false olarak ayarlanmasına izin verilmiyor; girdiyi tamamen kaldırmanız gerekir.)

Denetim öncesi denetimi, önceden güncelleştirme yapılmamış işlem hatlarında veya zaten bir ortam sürümü ayarlanmış olan işlem hatlarında çalışmaz.

Mevcut işlem hattını ortam sürümlerine geçirme

Henüz bir ortam sürümü kullanmayan mevcut bir işlem hattını geçirmek için bu uçtan uca iş akışını izleyin. Spark Connect altında farklı davranabilecek kod desenlerini bulma, düzeltme ve ortam sürümünü güvenli bir şekilde kullanıma sağlama konusunda size yol gösterir.

1. İşlem hattında uyumluluk taramasını etkinleştirin. Uyumluluk taraması bölümünde açıklandığı gibi işlem hattında taramayı etkinleştirin. Algılanan desenlerin olay günlüğünde ortaya çıkmasına neden olan ve etkinleştirme girişiminizi koruyan denetim öncesi denetimi sağlayan şey budur.

2. İşlem hattı çalıştırması tetikleme ve uyumluluk olaylarını gözden geçirme. Normal bir işlem hattı güncelleştirmesi tetikleyin. İşlem başarıyla tamamlandıktan sonra olaylar için BehaviorChangeInSparkConnectWARN işlem hattı olay günlüğünü sorgulayın. Her olay algılanan bir desen bildirir. Sorun kodlarının, örnek desenlerin ve önerilen düzeltmelerin tam listesi için uyumluluk olayları başvurusuna bakın.

3. İşlem hattı kodunuzu algılanan desenleri ele almak için güncelleştirin. Algılanan her düzen için, önerilen düzeltmeyi izleyerek işlem hattı kodunuzu güncelleştirin. Her değişiklik sonrasında başka bir işlem hattı güncelleştirmesi tetikleyin ve karşılık gelen olayların artık görünmediğinden emin olun. Başarılı bir güncelleştirme için olay günlüğü artık herhangi bir uyumluluk olayı ortaya çıkarıncaya kadar tekrarlayın.

4. İşlem hattında ortam sürümünü etkinleştirin. En son başarılı güncelleştirme hiçbir uyumluluk olayına sahip olduktan sonra, İşlem hattında ortam sürümünü etkinleştirme bölümünde açıklandığı gibi kullanıcı arabirimini, API'yi veya paketi kullanarak environment_version ekleyin. Sonraki güncelleştirme Spark Connect ve sabitlenmiş Python dil sürümü ve önceden yüklenmiş kitaplıklarla çalışır.

   Uyumluluk uyarıları hala mevcut olduğundan güncelleştirme başarısız olursa, değerini bırakın environment_version, 2. adıma dönün ve yeniden denemeden önce kalan uyarıları çözün.

5. Geçişi doğrulayın. Ortam sürümüyle ilk güncelleştirme tamamlandıktan sonra şunları doğrulayın:

   • Olay create_update günlüğündeki olay, beklenen değere ayarlanmış olarak gösterilir environment_version .
   • İşlem hattı beklenen verileri üretir ve yeni hata olayı görüntülenmez.
   • Davranış değişiklikleri bölümünde açıklanan küçük davranış farklılıkları için aşağı akış tablolarını spot-check yapın.

Geri alma

İşlem hattı geçişten sonra yanlış davranıyorsa işlem hattı ayarlarından öğesini kaldırın environment_version . Sonraki güncelleştirme, önceki Python çalışma zamanı yapılandırmasıyla çalışır. Hata ayıklamak için geri alınan çalıştırmayı kullanın, ardından sorunu tanımlayıp düzelttiğinizden sonra 2. adımdan geçişi yineleyin.

Uyumluluk olayları başvurusu

İşlem hattında uyumluluk taraması etkinleştirildiğinde, algılanan desen başına işlem hattı olay günlüğünde bir BehaviorChangeInSparkConnectWARN olay yayar. Tarama etkinleştirildiğinde ve önceki başarılı güncelleştirme herhangi bir desen algıladığında, işlem hattı desenler giderilene kadar etkinleştirmeyi de engeller environment_version .

Her olay, algılananları tanımlayan tek bir sorun kodu bildirir. Bir kodu aramak için Sorun kodları tablosunda bulabilirsiniz; her satır örnek desen ve önerilen düzeltmeyi içeren kategori bölümüne bağlanır.

Olay şekli

BehaviorChangeInSparkConnect olaylar standart işlem hattı olay günlüğü şemasını izler:

• event_type, behavior_change_in_spark_connect'e eşittir.
• level, WARN'e eşittir.
• details behavior_change_in_spark_connect tek bir issue alanı olan nesnesini içerir. Sorun değeri, aşağıda listelenen kodlardan biridir.
• message algılanan desenin okunabilir bir açıklamasıdır.

Sorun kodları

Veritabanı ve katalog mutasyonları

İşlem hattı kodu varsayılan veritabanını veya kataloğu sessize aldığınızda bu sorunlar yayılır. Ortam sürümüyle, mutasyondan önce inşa edilen DataFrame'ler yeni veritabanını veya kataloğu kullanarak tabloları çözebilir.

Bir olayı tetikleyen örnek desen:

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

Ortam sürümü olmadan, df katalogdan events çözümlermarketing. Ortam sürümüyle, df katalogdan events çözümlersales.

Önerilen düzeltme: Çözümlemenin varsayılan kataloğa veya veritabanına bağımlı olmaması için tablo adlarını tam olarak niteleyin ve DataFrame oluşturma ve kullanma arasında varsayılan kataloğu veya veritabanını değiştirmekten kaçının.

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 yapılandırma mutasyonları

İşlem hattı kodu Spark yapılandırmasını ortam sürümü altında DataFrame davranışını değiştirebilecek şekilde sessize aldığınızda bu sorunlar yayılır.

Bir olayı tetikleyen örnek desen:

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

Ortam sürümü olmadan, atama DataFrame oluşturma zamanında conf değerini kullanır. Ortam sürümüyle, atama işlemi kullanır spark.sql.ansi.enabled=true ve geçersiz girişte başarısız olabilir.

Önerilen düzeltme: Herhangi bir DataFrame oluşturulmadan önce işlem hattı dosyasının en üstünde tüm gerekli Spark yapılandırmalarını ayarlayın. Sorgu başına yapılandırma için işlem hattı belirtimindeki configuration işlem hattının ayarını kullanın.

Geçici görünüm değiştirmeleri

Bu sorunlar, işlem hattı kodu oluşturulduktan sonra dataframe'e başvuran geçici bir görünümün yerini alırsa ortaya çıkar. Ortam sürümüyle, mevcut DataFrame yeni görünüm içeriğini yansıtabilir.

Bir olayı tetikleyen örnek desen:

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

Ortam sürümü olmadan içerir mytable[(1, "Original Row")]. Ortam sürümü ile içerir mytable[(2, "Replaced Row")].

Önerilen düzeltme: Her geçici görünümü tek bir kez oluşturun ve değiştirmeyin. İlgili verilerle birden çok görünüme ihtiyacınız varsa, her birine ayrı bir ad verin.

UDF ve UDTF mutasyonları

bu sorunlar, işlem hattı kodu bir UDF veya UDTF'yi ortam sürümü altında davranışı değiştirecek şekilde sessize alırsa yayılır.

Bir olayı tetikleyen örnek desen:

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

Ortam sürümü olmadan içerir my_mv[("alex_b",)]. Ortam sürümü ile içerir my_mv[("alex_a",)].

Suggested fix: Değerleri Python genel değerlerden yakalamak yerine UDF'ye bağımsız değişken olarak geçirin veya UDF'yi tanımlamadan önce geneli ayarlayın ve daha sonra sessize alma.

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

Akış işlevleri içinde istekli yürütme

İşlem hattı kodu, bir işlem hattı dekoratörü (@table, vb.) tarafından dekore edilmiş bir işlev içinde istekli bir Spark komutu gerçekleştirdiğinde bu @materialized_viewsorunlar yayılır. Akış işlevlerinin bir DataFrame tanımlaması ve döndürmesi beklenir; ortam sürümü ayarlanmış bir akış işlevi içinde veri yazan, akış sorgularını yöneten, kaynakları kaydeden veya ML işlemlerini çalıştıran istekli komutlara izin verilmez.

Önerilen düzeltme: İstekli işlemi akış işlevinin dışına taşıyın ve bunun yerine akış işlevinden bir DataFrame döndürür. Tabloya yazma veya akış sorgusu başlatma gibi yan etkiler işlem hattı tanımının dışındadır; işlem hattı altyapısı, akış işlevi tarafından döndürülen DataFrame'in gerçekleştirilmesini işler.

Olay günlüğünde uyumluluk olaylarını bulma

Aşağıdaki sorgu, bir işlem hattı için en son sıralanmış tüm uyumluluk olaylarını döndürür:

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;

Son güncelleştirmelerde sorun koduna göre olayları saymak için:

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;

Olay günlüğünü sorgulamak için bkz. Olay günlüğünü sorgulama.

Ek kaynaklar

• İşlem hatları için ortam sürümlerini yapılandırma — özelliğe genel bakış, ortam sürümünü etkinleştirme.
• İşlem hattı olay günlüğü şeması — tam işlem hattı olay günlüğü şeması.
• İşlem hattı olay günlüğü — işlem hattı olay günlüğünü sorgulama.