Azure Databricks'te şema evrimi

Şema evrimi , sistemin zaman içinde veri yapısındaki değişikliklere uyum sağlayabilmesini ifade eder. Bu değişiklikler, yeni alanların eklendiği, veri türlerinin değiştiği veya iç içe yerleştirilmiş yapıların geliştiği yarı yapılandırılmış verilerle, olay akışlarıyla veya üçüncü taraf kaynaklarla çalışırken yaygındır.

Yaygın değişiklikler şunlardır:

  • Yeni sütunlar: Daha önce tanımlanmamış, bazen özel bir dolum değeri olan ek alanlar.
  • Sütun yeniden adlandırma: Bir sütun adını ( örneğin, yerine) namefull_namedeğiştirme.
  • Bırakılan sütunlar: Tablo şemasından sütunları kaldırma.
  • Tür genişletmesi: Bir sütunun türünü daha geniş bir türle değiştirme. Örneğin, bir INT alan haline geliyor DOUBLE.
  • Diğer tür değişiklikleri: Bir sütunun türünü değiştirmek. Örneğin, bir INT alan haline geliyor STRING.

Şema evrimi desteklenmesi, sık sık el ile yapılan güncelleştirmeler olmadan değişen verilere uyum sağlayabilecek dayanıklı ve uzun süre çalışan işlem hatları oluşturmak için kritik öneme sahiptir.

Components

Azure Databricks şemasının evrimi dört ana bileşen kategorisi içerir ve her işleme şeması birbirinden bağımsız olarak değişir:

  1. Bağlayıcılar: Dış kaynaklardan veri alınan bileşenler. Bunlar Otomatik Yükleyici, Kafka, Kinesis ve Lakeflow bağlayıcılarıdır.
  2. Biçim ayrıştırıcıları: , from_json, from_avrove from_xmlgibi from_protobufham biçimlerin kodunu çözen işlevler.
  3. Altyapılar: Yapılandırılmış Akış da dahil olmak üzere sorguları yürüten altyapıları işleme.
  4. Veri kümeleri: Akış tabloları, kalıcı görüntüler, Delta tabloları ve verileri kalıcı hale getiren ve sunan görünümler.

Şema evrimi

Veri mühendisliği mimarisinin şema evrimindeki her bileşen bağımsızdır. Veri işleme akışınızda istenen davranışı elde etmek için tek tek bileşenlerde şema evrimini yapılandırmak sizin sorumluluğunuzdadır.

Örneğin, Verileri Delta tablosuna almak için Otomatik Yükleyici kullanılırken iki kalıcı şema vardır: biri şema konumunda Otomatik Yükleyici tarafından yönetilir ve diğeri hedef Delta tablosunun şemasıdır. İstikrarlı bir durumda, bu ikisi aynı. Otomatik Yükleyici gelen verilere göre şemasını geliştirdiğinde Delta tablosunun da şemasını geliştirmesi gerekir, aksi durumda sorgu başarısız olur. Bu durumda, (a) şema evrimini etkinleştirerek veya doğrudan bir DDL komutu kullanarak hedef Delta tablosu şemasını güncelleştirebilir veya (b) hedef Delta tablosunun tam yeniden yazmasını yapabilirsiniz.

Bağlayıcıya göre şema evrimi desteği

Aşağıdaki bölümlerde her Azure Databricks bileşeninin farklı şema değişiklikleri türlerini nasıl işlediği açıklanır.

Otomatik Yükleyici

Otomatik Yükleyici sütun değişikliklerini ve tür genişletmeyi destekler. cloudFiles.schemaEvolutionMode ve rescuedDataColumn ile otomatik şema evrimini yapılandırın. El ile schemaHints veya sabit bir schema ayarlayabilirsiniz. Şema otomatik olarak geliştirildiğinde, akış başlangıçta başarısız olur. Yeniden başlatıldığında, geliştirilen şema kullanılır. Bkz. Otomatik Yükleyici şemasının evrimi nasıl çalışır?.

  • Yeni sütunlar: Desteklenmesi, yapılan seçime bağlıdır schemaEvolutionMode. Şemaya yeni sütunlar eklemek için gerekli el ile yeniden başlatma işlemiyle başarısız olur.
  • Sütun yeniden adlandırma: Seçilen elemana schemaEvolutionMode bağlı olarak desteklenir. Yeniden adlandırılan sütun, eklenen yeni bir sütun olarak değerlendirilir ve yeni satırlar için eski sütun ile NULL doldurulur. Şemayı güncelleştirmek için manuel bir yeniden başlatma gerektiren bir hata oluşur.
  • Silinmiş sütunlar: Desteklenir. tr-TR: Geçici silme olarak değerlendirilir; silinen sütun için yeni satırlar NULL olarak ayarlanır.
  • Tür genişletme: Databricks Runtime 16.4 ve üzerinde schemaEvolutionMode için ayarlanmış olarak addNewColumnsWithTypeWideningdesteklenir. Desteklenen veri türü değişiklikleri otomatik olarak genişletilir. Desteklenmeyen tür değişiklikleri rescuedDataColumn içinde yakalanır. Daha fazla bilgi için bkz. Otomatik Yükleyici ile Otomatik Tür Genişletme.
  • Diğer tür değişiklikleri: Desteklenmez. Tür değişiklikleri, rescuedDataColumn içinde, rescueDataColumn ayarlandığında ve schemaEvolutionModerescue olarak ayarlandığında yakalanır. Aksi takdirde el ile şema değişikliği gerekir.

Delta bağlayıcısı

Delta bağlayıcısı şema evrimini destekleyebilir. Sütun eşlemesi ve schemaTrackingLocation'un etkinleştirildiği bir Delta tablosundan okuma, sütun yeniden adlandırma ve bırakılan sütunlar için şema evrimi destekler. Akışı durdurmadan şemayı geliştirmek için ilgili değişikliklerin her biri için doğru Spark yapılandırmasını ayarlamanız gerekir. Aksi takdirde akış, bir değişiklik algılandığında izlenen şemasını günceller ve ardından durur. Ardından işlemeye devam etmek için akış sorgusunu el ile yeniden başlatmanız gerekir.

  • Yeni sütunlar: Desteklenir. Etkinleştirildiğinde mergeSchema , yeni sütunlar otomatik olarak eklenir. Aksi takdirde sorgu başarısız olur ve şemaya yeni sütunlar eklemek için akışı yeniden başlatmanız gerekir, ancak Delta tablosu yeniden yazma gerektirmez.
  • Sütun yeniden adlandırma: Desteklenir. Spark yapılandırması spark.databricks.delta.streaming.allowSourceColumnRenameile bir akış sorgusu içinde şema geliştirebilirsiniz.
  • Silinmiş sütunlar: Desteklenir. Spark yapılandırması spark.databricks.delta.streaming.allowSourceColumnDropile bir akış sorgusu içinde şema geliştirebilirsiniz.
  • Tür genişletme: Databricks Runtime 16.4 LTS ve üzerinde desteklenir. mergeSchema etkinleştirildiğinde ve tür genişletme hedef tabloda etkin olduğunda, tür değişiklikleri otomatik olarak işlenir. delta.enableTypeWidening tablo özelliğiyle tür genişletmeyi etkinleştirebilirsiniz. Bkz. Tür genişletme.
  • Diğer tür değişiklikleri: Desteklenmez.

SaaS ve CDC bağlayıcıları

SaaS ve CDC bağlayıcıları, sütunlar değiştiğinde şemayı otomatik olarak geliştirmektedir. Bu, bir değişiklik algılandığında otomatik yeniden başlatma yoluyla işlenir. Tür değişiklikleri tam yenileme gerektirir.

  • Yeni sütunlar: Desteklenir. Şema uyuşmazlıklarını çözmek için sorgu otomatik olarak yeniden başlatılır.
  • Sütun yeniden adlandırma: Desteklenir. Şema uyuşmazlıklarını çözmek için sorgu otomatik olarak yeniden başlatılır. Yeniden adlandırılan sütun, eklenen yeni bir sütun olarak değerlendirilir.
  • Silinmiş sütunlar: Desteklenir. Bırakılan sütunlar yumuşak silme olarak değerlendirilir, silinen sütun için yeni satırlar NULL olarak ayarlanır.
  • Tür genişletme: Desteklenmez. Şemanın güncelleştirilmesi tam yenileme gerektirir.
  • Diğer tür değişiklikleri: Desteklenmez. Şemanın güncelleştirilmesi tam yenileme gerektirir.

Kinesis, Kafka, Pub/Sub ve Pulsar bağlayıcıları

Yerel şema evrimi desteklenmez. Bağlayıcı işlevlerinin her biri ikili blob döndürür. Şema evrimi biçim ayrıştırıcısı tarafından işlenir.

  • Yeni sütunlar: Biçim ayrıştırıcısı tarafından işlenir.
  • Sütun yeniden adlandırma: Biçim ayrıştırıcısı tarafından işlenir.
  • Bırakılan sütunlar: Biçim ayrıştırıcısı tarafından işlenir.
  • Tür genişletme: Biçim ayrıştırıcısı tarafından işlenir.
  • Diğer tür değişiklikleri: Biçim ayrıştırıcısı tarafından işlenir.

Biçim ayrıştırıcıya göre şema evrimi desteği

from_json ayrıştırıcı

from_json Ayrıştırıcı şema evrimini desteklemez. Şemayı el ile güncelleştirmeniz gerekir. from_json, Lakeflow işlem hatlarında kullanıldığında, schemaLocationKey ve schemaEvolutionMode ile otomatik şema evrimi etkinleştirilebilir.

  • Yeni sütunlar: Otomatik şema evrimi etkinleştirildiğinde, Otomatik Yükleyici gibi davranır.
  • Sütunları yeniden adlandırma: Otomatik şema evrimi etkinleştirildiğinde, Otomatik Yükleyici gibi davranır.
  • Bırakılan sütunlar: Otomatik şema evrimi etkinleştirildiğinde, Otomatik Yükleyici gibi davranır.
  • Tür genişletme: Otomatik şema evrimi etkinleştirildiğinde, Otomatik Yükleyici gibi davranır.
  • Diğer tür değişiklikleri: Otomatik şema evrimi etkinleştirildiğinde, Otomatik Yükleyici gibi davranır.

from_avro ve from_protobuf ayrıştırıcılar

from_avro ve from_protobuf ayrıştırıcıları da aynı şekilde davranır. Şema Confluent Schema Registry'den getirilebilir veya kullanıcı bir şema sağlayabilir ve şemayı el ile güncelleştirmelidir. from_avro veya from_protobuf işlevinde şema evrimi kavramı yoktur; yürütme altyapısı ve Schema Registry tarafından işlenmelidir.

  • Yeni sütunlar: Confluent Schema Registry ile desteklenir. Aksi takdirde, kullanıcının şemayı el ile güncelleştirmesi gerekir.
  • Sütun yeniden adlandırma: Confluent Schema Registry ile desteklenir. Aksi takdirde, kullanıcının şemayı el ile güncelleştirmesi gerekir.
  • Silinen sütunlar: Confluent Schema Registry tarafından desteklenmektedir. Aksi takdirde, kullanıcının şemayı el ile güncelleştirmesi gerekir.
  • Tür genişletme: Confluent Schema Registry tarafından desteklenmektedir. Aksi takdirde, kullanıcının şemayı el ile güncelleştirmesi gerekir.
  • Diğer tür değişiklikleri: Confluent Schema Registry ile desteklenir. Aksi takdirde, kullanıcının şemayı el ile güncelleştirmesi gerekir.

from_csv ve from_xml ayrıştırıcılar

from_csv ve from_xml ayrıştırıcıları şema evrimini desteklemez.

  • Yeni sütunlar: Desteklenmez
  • Sütun yeniden adlandırma: Desteklenmez
  • Bırakılan sütunlar: Desteklenmez
  • Tür genişletme: Desteklenmez
  • Diğer tür değişiklikleri: Desteklenmez

Motora göre şema evrimi desteği

Yapılandırılmış Akış

Bir akış sorgusunun şeması planlama aşamasında kilitlenir ve tüm mikro toplu işlemler yeniden planlama yapmadan bu planı yeniden kullanır. Kaynak şema yürütmenin ortasında değişirse sorgu başarısız olur ve Spark'ın yeni şemada yeniden plan yapması için kullanıcının akış sorgusunu yeniden başlatması gerekir.

Akışın yazdığı veri kümesinin şema evrimini de desteklemesi gerekir.

  • Yeni sütunlar: Desteklenir. Sorgu başarısız olur ve şema uyuşmazlıklarını çözmek için akışı yeniden başlatmanız gerekir.
  • Sütun yeniden adlandırma: Desteklenir. Sorgu başarısız olur ve şema uyuşmazlıklarını çözmek için akışı yeniden başlatmanız gerekir.
  • Silinmiş sütunlar: Desteklenir. Sorgu başarısız olur ve şema uyuşmazlıklarını çözmek için akışı yeniden başlatmanız gerekir.
  • Tür genişletme: Desteklenir. Sorgu başarısız olur ve şema uyuşmazlıklarını çözmek için akışı yeniden başlatmanız gerekir.
  • Diğer tür değişiklikleri: Desteklenir. Sorgu başarısız olur ve şema uyuşmazlıklarını çözmek için akışı yeniden başlatmanız gerekir.

Veri kümesine göre şema evrimi

Akış tabloları

Akış tabloları varsayılan olarak birleştirme şeması evrim davranışını destekler. Şemanın güncelleştirilmesi el ile yeniden başlatma gerektirmez, ancak rastgele şema değişiklikleri tam yenileme gerektirir.

  • Yeni sütunlar: Desteklenir. Şema uyuşmazlığı çözmek için sorgu otomatik olarak yeniden başlatılır.
  • Sütun yeniden adlandırma: Desteklenir. Şema uyuşmazlığı çözmek için sorgu yeniden başlatılır. Yeniden adlandırılan sütun, eklenen yeni bir sütun olarak değerlendirilir.
  • Silinmiş sütunlar: Desteklenir. Bırakılan sütunlar geçici silme olarak kabul edilir ve silinen sütun için yeni satırlar NULL olarak ayarlanır.
  • Tür genişletme: Desteklenir. Tür genişletme, ya işlem hattı düzeyinde ya da doğrudan tabloda etkinleştirilmelidir. Bkz. Lakeflow işlem hatlarında tür genişletme.
  • Diğer tür değişiklikleri: Desteklenmez. Şemanın güncelleştirilmesi tam yenileme gerektirir.

Gerçekleştirilmiş görünümler

Şemada yapılan herhangi bir güncelleştirme veya tanımlama sorgusu, gerçekleştirilmiş görünümün tam olarak yeniden derlenmesine neden olabilir.

  • Yeni sütunlar: Tam yeniden derleme tetiklenmiş.
  • Sütun yeniden adlandırma: Tam yeniden hesaplama tetiklendi.
  • Bırakılan sütunlar: Tam yeniden hesaplama başlatıldı.
  • Tür genişletme: Tam yeniden derleme tetiklendi.
  • Diğer tür değişiklikleri: Tam yeniden derleme tetiklenmiş.

Delta tabloları

Delta tabloları tablo verilerini yeniden yazmadan sütun türlerini yeniden adlandırma, bırakma ve genişletme dahil olmak üzere tablo şemasını güncelleştirmek için çeşitli yapılandırmaları destekler. Desteklenen yapılandırmalar arasında birleşme şeması evrimi, sütun eşlemesi, tür genişletme ve şema üzerine yazma bulunmaktadır.

  • Yeni sütunlar: Desteklenir. Birleştirme şeması evrimi etkinleştirildiğinde, Delta tablosunun yeniden yazılmasını gerektirmeden otomatik olarak gelişir. Birleştirme şeması evrimi etkinleştirilmemişse güncelleştirmeler başarısız olur.
  • Sütun yeniden adlandırma: Desteklenir. Sütun eşlemesi etkinleştirildiyse, manuel ALTER TABLE DDL komutlar aracılığıyla yeniden adlandırılabilir. Delta tablosunun yeniden yazılmasını gerektirmez.
  • Silinmiş sütunlar: Desteklenir. Sütun eşleme etkinleştirilmiş komutları kullanarak sütunları el ile kaldırabilirsiniz. Delta tablosunun yeniden yazılmasını gerektirmez.
  • Tür genişletme: Desteklenir. Tür genişletme ve birleştirme şeması evrimi etkinleştirildiğinde tür değişikliğini otomatik olarak uygular. Tür genişletme etkinleştirildiğinde sütunları manuel ALTER TABLE DDL komutlarla genişletebilirsiniz. Her ikisinde de yapılandırılmadan işlemler başarısız olur. Bkz Otomatik şema evrimi ile türleri genişletme.
  • Diğer tür değişiklikleri: Desteklenir, ancak Delta Tablosunun tam yeniden yazılmasını gerektirir. Delta Tablosunun tam yeniden yazılmasını sağlayan öğesini etkinleştirmeniz overwriteSchemagerekir. Aksi takdirde işlemler başarısız olur.

İzlenimler

Görünümün yeni şemayla eşleşmeyen bir column_list şeması varsa veya ayrıştırılamayan bir sorgusu varsa, görünüm geçersiz olur. Tür değişiklikleri için SCHEMA TYPE EVOLUTION ile ve tür değişikliklerinin yanı sıra yeni, yeniden adlandırılmış ve bırakılan sütunlar için SCHEMA EVOLUTION ile şema evrimini etkinleştirebilirsiniz (bu, tür evriminin bir üst kümesidir).

  • Yeni sütunlar: Desteklenir. SCHEMA EVOLUTION moduyla, açık column_list bir işlem olmadığında, görünüm herhangi bir el ile müdahale olmadan kendiliğinden gelişir. Aksi takdirde, görünüm geçersiz hale gelebilir ve kullanıcı bunu sorgulayamaz.
  • Sütunları yeniden adlandırma: Desteklenir. SCHEMA EVOLUTION moduyla, açık column_list bir işlem olmadığında, görünüm herhangi bir el ile müdahale olmadan kendiliğinden gelişir. Aksi takdirde, görünüm geçersiz hale gelebilir.
  • Silinmiş sütunlar: Desteklenir. SCHEMA EVOLUTION moduyla, açık column_list bir işlem olmadığında, görünüm herhangi bir el ile müdahale olmadan kendiliğinden gelişir. Aksi takdirde, görünüm geçersiz hale gelebilir.
  • Tür genişletme: Desteklenir. SCHEMA TYPE EVOLUTION moduyla, görünüm her tür değişikliği için otomatik olarak uyarlanır. SCHEMA EVOLUTION moduyla, açık column_list bir işlem olmadığında, görünüm herhangi bir el ile müdahale olmadan kendiliğinden gelişir. Aksi takdirde, görünüm geçersiz hale gelebilir.
  • Diğer tür değişiklikleri: Desteklenir. SCHEMA TYPE EVOLUTION moduyla, görünüm her tür değişikliği için otomatik olarak uyarlanır. SCHEMA EVOLUTION moduyla, açık column_list bir işlem olmadığında, görünüm herhangi bir el ile müdahale olmadan kendiliğinden gelişir. Aksi takdirde, görünüm geçersiz hale gelebilir.

Example

Aşağıdaki örnekte, Confluent Schema Registry'de kayıtlı Avro kodlu yükleri olan bir Kafka konusunun nasıl alınıp şema evriminin etkinleştirildiği yönetilen bir Delta tablosuna nasıl yazıldığı gösterilmektedir.

Önemli noktalar gösterilmiştir:

  • Kafka Bağlayıcısı ile tümleştirme.
  • Kafka Schema Registry ile from_avro kullanarak Avro kayıtlarının kodunu çözme.
  • Şema evrimi avroSchemaEvolutionMode ayarlanarak yönetilebilir.
  • Eklemeli değişikliklere izin vermek amacıyla mergeSchema etkinleştirilmiş bir Delta tablosuna yazın.

Kod, Confluent şema kayıt defterini kullanarak Avro ile kodlanmış verilerin çıkışını veren bir Kafka konusuna sahip olduğunuzu varsayar.

# ----- CONFIG: fill these in -----
# Catalog and schema:
CATALOG = "<catalog_name>"
SCHEMA = "<schema_name>"
# Schema Registry:
# (This is where the producer evolves the schema)
SCHEMA_REG = "<schema registry endpoint>"
SR_USER    = "<api key>"
SR_PASS    = "<api secret>"
# Confluent Cloud: SASL_SSL broker:
BOOTSTRAP = "<server:ip>"
# Kafka topic:
TOPIC = "<topic>"
# ----- end: config -----

BRONZE_TABLE = f"{CATALOG}.{SCHEMA}.bronze_users"
CHECKPOINT = f"/Volumes/{CATALOG}/{SCHEMA}/checkpoints/bronze_users"

# Kafka auth (example for Confluent Cloud SASL/PLAIN over SSL)
KAFKA_OPTS = {
  "kafka.security.protocol": "SASL_SSL",
  "kafka.sasl.mechanism": "PLAIN",
  "kafka.sasl.jaas.config": f"kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username='{SR_USER}' password='{SR_PASS}';"
}

# ----- Evolution knobs -----
# spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", value = True)

from pyspark.sql.functions import col
from pyspark.sql.avro.functions import from_avro

# Build reader
reader = (spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", BOOTSTRAP)
  .option("subscribe", TOPIC)
  .option("startingOffsets", "earliest")
)

# Attach Kafka auth options
for k, v in KAFKA_OPTS.items():
    reader = reader.option(k, v)

# --- No native schema evolution supported. Returns a binary blob. ---
raw_df = reader.load()

# Decode Avro with Schema Registry
# --- The format parser handles updating the schema using the schema registry ---
decoded = from_avro(
    data=col("value"),
    jsonFormatSchema=None, # using SR
    subject=f"{TOPIC}-value",
    schemaRegistryAddress=SCHEMA_REG,
    options={
      "confluent.schema.registry.basic.auth.credentials.source": "USER_INFO",
      "confluent.schema.registry.basic.auth.user.info": f"{SR_USER}:{SR_PASS}",
      # Behavior on schema changes:
      "avroSchemaEvolutionMode": "restart", # fail-fast so you can restart and adopt new fields
      "mode": "FAILFAST"
    }
).alias("payload")

bronze_df = raw_df.select(decoded, "timestamp").select("payload.*", "timestamp")

# Write to a managed Delta table as a STREAM
# --- Need to enable schema evolution separately for streaming to a Delta separately with mergeSchema --
(bronze_df.writeStream
  .format("delta")
  .option("checkpointLocation", CHECKPOINT)
  .option("ignoreChanges", "true")
  .outputMode("append")
  .option("mergeSchema", "true") # only supports adding new columns. Renaming, dropping, and type changes need to be handled separately.
  .trigger(availableNow=True) # Use availableNow trigger for Databricks SQL/Unity Catalog
  .toTable(BRONZE_TABLE)
)
  • Last updated on