Veri türü genişletme

Databricks Runtime 15.4 LTS ve üzeri sürümlerin Delta Lake tablolarında bulunan tür genişletme, veri dosyalarını yeniden yazmadan sütun veri türlerini daha geniş bir türe dönüştürmenizi sağlar.

Unity Kataloğu tarafından yönetilen tüm tablolarda varsayılan olarak Delta Lake kullanılır. Bkz. Delta Lake ve Apache Iceberg için Unity Catalog tarafından yönetilen tablolar.

Note

Tip genişletmeyi etkinleştirmek, okuyucu ve yazıcı protokollerini yükseltir. Bu, dış Delta Lake istemcileriyle uyumluluğu etkileyebilir. Bkz . Delta Lake özellik uyumluluğu ve protokolleri.

Tür genişletme etkinleştirilmiş tablolar yalnızca Databricks Runtime 15.4 LTS ve üzeri tarafından okunabilir.

Desteklenen tür değişiklikleri

Türleri aşağıdaki kurallara göre genişletebilirsiniz:

Kaynak türü Desteklenen daha geniş türler
BYTE SHORT, INT, BIGINT, DECIMAL, , DOUBLE
SHORT INT, BIGINT, DECIMAL, DOUBLE
INT BIGINT, DECIMAL, DOUBLE
BIGINT DECIMAL
FLOAT DOUBLE
DECIMAL DECIMAL daha yüksek hassasiyet ve ölçekle
DATE TIMESTAMP_NTZ
VOID Herhangi bir tür

Tür değişiklikleri yapıların, haritaların ve dizilerin içinde iç içe yerleştirilmiş en üst düzey sütunlar ve alanlar için desteklenir.

Note

VOID herhangi bir türe dönüştürme, tabloda tür genişletmenin etkinleştirilmesini gerektirmez. Bir sütunun türünü VOID güncelleştiren tüm işlemler ek yapılandırma olmadan başarılı olur. VOID tür genişletme, Databricks Runtime 18.2 ve üstü sürümlerde kullanılabilir.

Ondalık davranışı

Bir işlem bir tamsayı türünü decimal veya double olarak yükselttiğinde ve aşağı akış alımı değeri bir tamsayı sütununa geri yazarken, Spark varsayılan olarak değerin kesirli bölümünü keser. Atama ilkesi davranışı hakkında ayrıntılı bilgi için bkz Mağaza ataması.

Herhangi bir sayısal türü decimalolarak değiştirirken, toplam duyarlık başlangıç duyarlığına eşit veya bundan büyük olmalıdır. Ölçeği de artırırsanız, toplam kesinlik aynı oranda artmalıdır.

byte, shortve int türleri için en düşük hedef decimal(10,0). long için asgari hedef decimal(20,0).

decimal(10,1)olan bir alana iki ondalık basamak eklemek istiyorsanız, en düşük hedef decimal(12,3).

Tür genişletmeyi etkinleştirme

Note

Tip genişletmeyi etkinleştirmek, okuyucu ve yazıcı protokollerini yükseltir. Bu, dış Delta Lake istemcileriyle uyumluluğu etkileyebilir. Bkz . Delta Lake özellik uyumluluğu ve protokolleri.

delta.enableTypeWidening tablo özelliğini trueolarak ayarlayarak var olan bir tabloda tür genişletmeyi etkinleştirebilirsiniz:

  ALTER TABLE <table_name> SET TBLPROPERTIES ('delta.enableTypeWidening' = 'true')

Tablo oluşturma sırasında tür genişletmeyi de etkinleştirebilirsiniz:

  CREATE TABLE T(c1 INT) TBLPROPERTIES('delta.enableTypeWidening' = 'true')

Tür değişikliğini el ile uygulama

ALTER COLUMN Türleri el ile değiştirmek için komutunu kullanın:

ALTER TABLE <table_name> ALTER COLUMN <col_name> TYPE <new_type>

Bu işlem, temel alınan veri dosyalarını yeniden yazmadan tablo şemasını güncelleştirir. Daha fazla bilgi için bkz. ALTER TABLE.

Otomatik şema evrimi ile türleri genişletme

Hedef tablolardaki veri türlerini gelen veri türüyle eşleşecek şekilde güncelleştirmek için tür genişletme ile şema evrimini kullanın.

Note

Tür genişletme etkinleştirilmeden, şema evrimi her zaman hedef tablodaki sütun türlerini eşleştirmek için verileri alta aktarmayı dener. Hedef tablolarınızdaki veri türlerini otomatik olarak genişletmeyi istemiyorsanız, şema evrimi etkin iş yüklerini çalıştırmadan önce tür genişletmeyi kapatmanız gerekir.

Alma sırasında bir sütunun veri türünü genişletmek için şema evrimini kullanmak için aşağıdaki koşulları karşılamanız gerekir:

  • Yazma komutu, otomatik şema evrimi etkin olarak çalışır.
  • Hedef tabloda tür genişletme etkinleştirildi.
  • Kaynak sütun türü hedef sütun türünden daha geniştir.
  • Tür genişletme, tür değişikliğini destekler.

Bu koşulların tümünü karşılamayan tür uyuşmazlıkları normal şema uygulama kurallarına tabi olur. Şema uygulaması için bkz..

Example

Aşağıdaki örneklerde, tür genişletmenin şema evrimi ile nasıl çalıştığı gösterilmektedir.

Python

INT sütununa sahip bir hedef tablo ve BIGINT sütununa sahip bir kaynak tablo oluşturun:

spark.sql("CREATE TABLE target_table (id INT, data STRING) TBLPROPERTIES ('delta.enableTypeWidening' = 'true')")
spark.sql("CREATE TABLE source_table (id BIGINT, data STRING)")

Ekleme işlemi sırasında saveAsTable() sütununu otomatik olarak INT olarak genişletmek için BIGINT öğesini şema evrimiyle kullanın:

spark.table("source_table").write.mode("append").option("mergeSchema", "true").saveAsTable("target_table")

Şema evrimiyle MERGE INTO kullanın:

from delta.tables import DeltaTable

source_df = spark.table("source_table")
target_table = DeltaTable.forName(spark, "target_table")

(target_table.alias("target")
  .merge(source_df.alias("source"), "target.id = source.id")
  .withSchemaEvolution()
  .whenMatchedUpdateAll()
  .whenNotMatchedInsertAll()
  .execute()
)

Scala

INT sütununa sahip bir hedef tablo ve BIGINT sütununa sahip bir kaynak tablo oluşturun:

spark.sql("CREATE TABLE target_table (id INT, data STRING) TBLPROPERTIES ('delta.enableTypeWidening' = 'true')")
spark.sql("CREATE TABLE source_table (id BIGINT, data STRING)")

Ekleme işlemi sırasında saveAsTable() sütununu otomatik olarak INT olarak genişletmek için BIGINT öğesini şema evrimiyle kullanın:

spark.table("source_table").write.mode("append").option("mergeSchema", "true").saveAsTable("target_table")

Şema evrimiyle MERGE INTO kullanın:

import io.delta.tables.DeltaTable

val sourceDf = spark.table("source_table")
val targetTable = DeltaTable.forName(spark, "target_table")

targetTable.alias("target")
  .merge(sourceDf.alias("source"), "target.id = source.id")
  .withSchemaEvolution()
  .whenMatched().updateAll()
  .whenNotMatched().insertAll()
  .execute()

SQL

INT sütununa sahip bir hedef tablo ve BIGINT sütununa sahip bir kaynak tablo oluşturun:

CREATE TABLE target_table (id INT, data STRING) TBLPROPERTIES ('delta.enableTypeWidening' = 'true');
CREATE TABLE source_table (id BIGINT, data STRING);

Ekleme işlemi sırasında INSERT INTO sütununu otomatik olarak INT olarak genişletmek için BIGINT öğesini şema evrimiyle kullanın:

INSERT WITH SCHEMA EVOLUTION INTO target_table SELECT * FROM source_table;

Şema evrimiyle MERGE INTO kullanın:

MERGE WITH SCHEMA EVOLUTION INTO target_table
USING source_table
ON target_table.id = source_table.id
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *;

Otomatik Yükleyici

Important

Otomatik Yükleyici'de tür genişletme desteği Genel Önizleme aşamasındadır.

Otomatik Yükleyici, otomatik şema evrimi ile tür genişletmeyi destekler. Tür genişletmesi ve şema evrimi etkin bir Delta Lake tablosuna veri almak için Otomatik Yükleyici'yi kullandığınızda, sütun türleri gelen verilerle eşleşecek şekilde otomatik olarak genişletilir.

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", "<path-to-schema-location>")
  .load("<path-to-source-data>")
  .writeStream
  .option("mergeSchema", "true")
  .option("checkpointLocation", "<path-to-checkpoint>")
  .trigger(availableNow=True)
  .toTable("table_name")
)

Daha fazla bilgi için bkz. Otomatik Yükleyici ile Otomatik Tür Genişletme. Ayrıca, hedef tabloda tür genişletme etkinleştirilmelidir. Bkz . Tür genişletmeyi etkinleştirme.

Tür genişletme tablosu özelliğini devre dışı bırakma

özelliğini falseolarak ayarlayarak etkinleştirilen tablolarda yanlışlıkla tür genişletmeyi önleyebilirsiniz:

  ALTER TABLE <table_name> SET TBLPROPERTIES ('delta.enableTypeWidening' = 'false')

Bu ayar, tabloda gelecekteki tür değişikliklerini engeller, ancak tür genişletme tablosu özelliğini kaldırmaz veya önceki tür değişikliklerini geri almaz.

Tür genişletme tablosu özelliklerini tamamen kaldırmanız gerekiyorsa, aşağıdaki örnekte gösterildiği gibi DROP FEATURE komutunu kullanabilirsiniz:

 ALTER TABLE <table-name> DROP FEATURE 'typeWidening' [TRUNCATE HISTORY]

Note

Databricks Runtime 15.4 LTS kullanarak tür genişletmeyi etkinleştiren tablolar bunun yerine özelliği typeWidening-preview bırakmanızı gerektirir.

Tür genişletmeyi bırakırken Databricks, geçerli tablo şemasına uymayan tüm veri dosyalarını yeniden yazar. Bkz . Delta Lake tablosu özelliğini bırakma ve tablo protokolünü düşürme.

Delta Lake tablosundan veri akışı

Yapılandırılmış Akış'ta tür genişletme desteği Databricks Runtime 16.4 LTS ve üzerinde kullanılabilir.

Tür genişletmesi etkinleştirilmiş bir Delta Lake tablosundan akış yaparken, hedef tablodaki seçenekle mergeSchema şema evrimini etkinleştirerek akış sorguları için otomatik tür genişletmeyi yapılandırabilirsiniz. Hedef tabloda tür genişletme etkinleştirilmelidir. Bkz . Tür genişletmeyi etkinleştirme.

Python

(spark.readStream
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", "/path/to/checkpointLocation")
  .option("mergeSchema", "true")
  .toTable("output_table")
)

Scala

spark.readStream
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", "/path/to/checkpointLocation")
  .option("mergeSchema", "true")
  .toTable("output_table")

mergeSchema etkinleştirildiğinde ve hedef tabloda tür genişletme açık olduğunda:

  • Tür değişiklikleri, el ile müdahale gerekmeden aşağı akış tablosuna otomatik olarak uygulanır.
  • Yeni sütunlar aşağı akış tablosu şemasına otomatik olarak eklenir.

mergeSchema etkin değildir; bu durumda, değerler spark.sql.storeAssignmentPolicy yapılandırmasına göre işlenir, bu da varsayılan olarak değerleri hedef sütun türü ile uyumlu olacak şekilde aşağı dönüştürür. Atama politikası davranışı hakkında daha fazla bilgi için bkz Mağaza ataması.

Akıştaki tür değişikliklerini işleme

Delta Lake tablosundan akış yaparken, tür değişiklikleri dahil olmak üzere eksiz şema değişikliklerini izlemek için bir şema izleme konumu sağlayabilirsiniz. Databricks Runtime 18.0 ve altında şema izleme konumu sağlamak gereklidir ve Databricks Runtime 18.1 ve sonraki sürümleri için isteğe bağlıdır.

schemaTrackingLocation SQL kullanarak ayarlayamazsınız. Bkz . Desteklenmeyen özellikler.

schemaTrackingLocation akış denetim noktanızla aynı yol içinde bir konuma ayarlanmalıdır. Örneğin:

Python

checkpoint_path = "/path/to/checkpointLocation"

(spark.readStream
  .option("schemaTrackingLocation", checkpoint_path)
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("output_table")
)

Scala

val checkpointPath = "/path/to/checkpointLocation"

spark.readStream
  .option("schemaTrackingLocation", checkpointPath)
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpointPath)
  .toTable("output_table")

Bir şema izleme konumu ayarlandıktan sonra akış, bir veri türü değişikliği algıladığında izlenen şemasını günceller ve ardından durur. Bu sırada, aşağı akış tablosunda tür genişletmeyi etkinleştirme veya akış sorgusunu güncelleştirme gibi tür değişikliğini işlemeniz gerekir.

İşlemeye devam etmek için Spark yapılandırmasını spark.databricks.delta.streaming.allowSourceColumnTypeChange veya DataFrame okuyucu seçeneğini allowSourceColumnTypeChangeaşağıdaki örnekte olduğu gibi ayarlayın:

Python

checkpoint_path = "/path/to/checkpointLocation"

(spark.readStream
  .option("schemaTrackingLocation", checkpoint_path)
  .option("allowSourceColumnTypeChange", "<delta_source_table_version>")
  # alternatively to allow all future type changes for this stream:
  # .option("allowSourceColumnTypeChange", "always")
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("output_table")
)

Scala

val checkpointPath = "/path/to/checkpointLocation"

spark.readStream
  .option("schemaTrackingLocation", checkpointPath)
  .option("allowSourceColumnTypeChange", "<delta_source_table_version>")
  // alternatively to allow all future type changes for this stream:
  // .option("allowSourceColumnTypeChange", "always")
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpointPath)
  .toTable("output_table")

SQL

  -- To unblock for this particular stream just for this series of schema change(s):
  SET spark.databricks.delta.streaming.allowSourceColumnTypeChange.ckpt_<checkpoint_id> = "<delta_source_table_version>"
  -- To unblock for this particular stream:
  SET spark.databricks.delta.streaming.allowSourceColumnTypeChange = "<delta_source_table_version>"
  -- To unblock for all streams:
  SET spark.databricks.delta.streaming.allowSourceColumnTypeChange = "always"

Akış durduğunda, denetim noktası kimliği <checkpoint_id> ve Delta Lake kaynak tablosu sürümü <delta_source_table_version>bir hata iletisi görüntülenir.

Delta Lake akış seçeneklerinin tam listesi için bkz. Delta Lake.

Lakeflow işlem hatları

Lakeflow işlem hatları için tür genişletmeyi işlem hattı düzeyinde veya tek tek tablolar için etkinleştirebilirsiniz. Tür genişletme, akış tablolarının tam olarak yenilenmesine gerek kalmadan sütun türlerinin işlem hattı yürütmesi sırasında otomatik olarak genişletilmesine olanak tanır. Gerçekleştirilmiş görünümlerdeki tür değişiklikleri her zaman tam bir yeniden derlemeyi tetikler ve bir kaynak tabloya tür değişikliği uygulandığında, bu tabloya bağlı gerçekleştirilmiş görünümler yeni türleri yansıtmak için tam bir yeniden derleme gerektirir.

İşlem hattının tamamı için tür genişletmeyi etkinleştirme

bir işlem hattındaki tüm tablolar için tür genişletmeyi etkinleştirmek için işlem hattı yapılandırmasını pipelines.enableTypeWideningayarlayın:

JSON

{
  "configuration": {
    "pipelines.enableTypeWidening": "true"
  }
}

YAML

configuration:
  pipelines.enableTypeWidening: 'true'

Belirli tablolar için tür genişletmeyi etkinleştirme

Tablo özelliğini delta.enableTypeWideningayarlayarak tek tek tablolar için tür genişletmeyi de etkinleştirebilirsiniz:

Python

import dlt

@dlt.table(
  table_properties={"delta.enableTypeWidening": "true"}
)
def my_table():
  return spark.readStream.table("source_table")

SQL

CREATE OR REFRESH STREAMING TABLE my_table
TBLPROPERTIES ('delta.enableTypeWidening' = 'true')
AS SELECT * FROM source_table

Aşağı akış okuyucularla uyumluluk

Tür genişletmesi etkinleştirilmiş tablolar yalnızca Databricks Runtime 15.4 LTS ve üzerinde okunabilir. İşlem hattınızda tür genişletmesi etkinleştirilmiş bir tablonun Databricks Runtime 14.3 ve altındaki okuyucular tarafından okunabilir olmasını istiyorsanız, aşağıdakilerden birini kullanmanız gerekir:

  • Özelliği delta.enableTypeWidening/pipelines.enableTypeWidening kaldırarak veya false olarak ayarlayarak tür genişletmeyi kapatın ve tablonun tam yenilemesini tetikleyin.
  • Tablonuzda Uyumluluk Modu'nu etkinleştirin.

OpenSharing

Note

OpenSharing'de Tür Genişletme desteği Databricks Runtime 16.1 ve üzerinde kullanılabilir.

Tür genişletmesi etkinleştirilmiş bir Delta Lake tablosunun paylaşılması Databricks-to-Databricks OpenSharing'de desteklenir. Sağlayıcı ve alıcı Databricks Runtime 16.1 veya üzerinde olmalıdır.

OpenSharing kullanılarak tür genişletme etkinleştirilmiş bir Delta Lake tablosundan değişiklik veri akışını okumak için yanıt biçimini olarak deltaayarlamanız gerekir:

spark.read
  .format("deltaSharing")
  .option("responseFormat", "delta")
  .option("readChangeFeed", "true")
  .option("startingVersion", "<start version>")
  .option("endingVersion", "<end version>")
  .load("<table>")

Tür değişiklikleri boyunca değişiklik veri akışını okumak desteklenmez. Bunun yerine işlemi, biri tür değişikliğini içeren tablo sürümünde biten, diğeri de tür değişikliğini içeren sürümden başlayan iki ayrı okuma işlemine bölmeniz gerekir.

Limitations

Apache Iceberg Uyumluluğu

Apache Iceberg, tür genişletmenin kapsadığı tüm tür değişikliklerini desteklemez. Bkz. Iceberg Schema Evolution.

Desteklenmeyen tür değişiklikleri şunları içerir:

  • byte, short, int, long veya decimaldouble
  • ondalık ölçek artırımı
  • date'dan timestampNTZ'e

Iceberg uyumluluğu etkinleştirilmiş bir Delta Lake tablosunda UniForm'u etkinleştirdiğinizde, yukarıda belirtilen tür değişikliklerinden birini uygulamak bir hataya neden olur. Bkz. UniForm kullanarak Iceberg istemcileri ile Delta Lake tablolarını okuma.

Bu desteklenmeyen tür değişikliklerinden birini Delta Lake tablosuna uygularsanız iki seçeneğiniz vardır:

  • Iceberg meta verilerini yeniden oluşturma: Tür genişletme tablosu özelliği olmadan Iceberg meta verilerini yeniden oluşturmak için aşağıdaki komutu kullanın:

    ALTER TABLE <table-name> SET TBLPROPERTIES ('delta.universalFormat.config.icebergCompatVersion' = '<version>')

    Bu, uyumsuz tür değişiklikleri uyguladıktan sonra Tekdüzen uyumluluğu korumanıza olanak tanır.

  • Tür genişletme tablosu özelliğini bırakın: Bkz. Tür genişletme tablosunu devre dışı bırakma.

Türe bağımlı işlevler

Bazı SQL işlevleri, giriş veri türüne bağlı sonuçlar döndürür. Örneğin, bağımsız değişken türü farklıysa işlevhash aynı mantıksal değer için farklı karma değerleri döndürür: hash(1::INT) değerinden hash(1::BIGINT)farklı bir sonuç döndürür.

Diğer türe bağımlı işlevler şunlardır: xxhash64, bit_get, bit_reverse, typeof.

Bu işlevleri kullanan sorgularda kararlı sonuçlar elde etmek için, değerleri açıkça istenen türe atamanız gerekir:

Python

spark.read.table("table_name") \
  .selectExpr("hash(CAST(column_name AS BIGINT))")

Scala

spark.read.table("main.johan_lasperas.dlt_type_widening_bronze2")
  .selectExpr("hash(CAST(a AS BIGINT))")

SQL

-- Use explicit casting for stable hash values
SELECT hash(CAST(column_name AS BIGINT)) FROM table_name

Desteklenmeyen özellikler

  • Delta Lake tablosundan tür değişikliğiyle akış yaparken SQL kullanarak şema izleme konumu ayarlayamazsınız.
  • OpenSharing kullanarak, tür genişletme özelliği etkin olan bir tabloyu Databricks olmayan tüketicilerle paylaşamazsınız.
  • Last updated on