create_auto_cdc_from_snapshot_flow

İşlev, create_auto_cdc_from_snapshot_flow veritabanı anlık görüntülerinden kaynak verileri işlemek için Lakeflow işlem hatlarını veri yakalama (CDC) işlevini değiştiren bir akış oluşturur. Bkz. ANLıK GÖRÜNTÜDEN OTOMATIK CDC nasıl çalışır?

Uyarı

Bu işlev önceki işlevinin apply_changes_from_snapshot()yerini alır. İki işlev aynı imzaya sahiptir. Databricks, yeni adı kullanmak için güncelleştirme önerir.

Önemli

Bu işlem için bir hedef akış tablonuz olmalıdır. Gerekli hedef tabloyu oluşturmak için create_streaming_table() işlevini kullanabilirsiniz.

Sözdizimi

from pyspark import pipelines as dp

dp.create_auto_cdc_from_snapshot_flow(
  target = "<target-table>",
  source = Any,
  keys = ["key1", "key2", "keyN"],
  stored_as_scd_type = "1",
  track_history_column_list = None,
  track_history_except_column_list = None,
  once = False
)

Uyarı

İşleme için AUTO CDC FROM SNAPSHOT varsayılan davranış, hedefte aynı anahtarlara sahip eşleşen bir kayıt olmadığında yeni bir satır eklemektir. Eşleşen bir kayıt varsa, yalnızca satırdaki değerlerden herhangi biri değişmişse güncelleştirilir. Hedefte anahtarları bulunan ancak kaynakta artık bulunmayan satırlar silinir.

Anlık görüntülerle CDC işleme hakkında daha fazla bilgi edinmek için bkz. AUTO CDC API'leri: İşlem hatları ile değişiklik verilerini yakalamayı basitleştirme. Fonksiyonun create_auto_cdc_from_snapshot_flow() kullanımına dair örnekler için periyodik anlık alım, tarihsel anlık alım ve SCD Tip 1 geri doldurma örneklerine bakınız.

Parametreler

Parametre Türü Description
target str Gerekli. Güncelleştirilecek tablonun adı. İşlevi yürütmeden create_auto_cdc_from_snapshot_flow() önce hedef tabloyu oluşturmak için create_streaming_table() işlevini kullanabilirsiniz.
source str veya lambda function Gerekli. Bir tablonun adı veya düzenli aralıklarla anlık görüntülenecek görünüm ya da işlenecek anlık görüntü DataFrame'i ve anlık görüntü sürümünü döndüren bir Python lambda işlevi. Bkz. Bağımsız değişkeni uygulamasource.
keys list Gerekli. Kaynak verilerdeki bir satırı benzersiz olarak tanımlayan sütun veya sütun bileşimi. Bu, hedef tablodaki belirli kayıtlara hangi CDC olaylarının uygulanacağını belirlemek için kullanılır. Şunları belirtebilirsiniz:
  • Dizelerin listesi: ["userId", "orderId"]
  • Spark SQL col() işlevlerinin listesi: [col("userId"), col("orderId"]. İşlevlere yönelik col() bağımsız değişkenler niteleyici içeremez. Örneğin, kullanabilirsiniz col(userId), ancak kullanamazsınız col(source.userId).
stored_as_scd_type str veya int Kayıtların SCD türü 1 veya SCD tür 2 olarak depolanması. 1 SCD türü 1 veya 2 SCD türü 2 için olarak ayarlayın. Varsayılan değer SCD tür 1'dir.
track_history_column_list veya track_history_except_column_list list Hedef tablodaki geçmiş için izlenecek çıktı sütunlarının bir alt kümesi. İzlenecek sütunların tam listesini belirtmek için kullanın track_history_column_list . İzlemenin dışında tutulacak sütunları belirtmek için kullanın track_history_except_column_list . Değeri dize listesi olarak veya Spark SQL col() işlevleri olarak bildirebilirsiniz:
  • track_history_column_list = ["userId", "name", "city"]
  • track_history_column_list = [col("userId"), col("name"), col("city")]
  • track_history_except_column_list = ["operation", "sequenceNum"]
  • track_history_except_column_list = [col("operation"), col("sequenceNum")

İşlevlere yönelik col() bağımsız değişkenler niteleyici içeremez. Örneğin, kullanabilirsiniz col(userId), ancak kullanamazsınız col(source.userId). Varsayılan değer, işleve hiçbir track_history_column_list veya track_history_except_column_list bağımsız değişken geçirilmediğinde hedef tabloya tüm sütunları eklemektir.
once bool Snapshot akışını sadece bir kez çalıştırıp çalıştırmamak meselesi. Eğer True, akış başarılı bir şekilde bağlandıktan sonraki artımlı güncellemelerde atlanır. Hedefin tam yenilenmesi akışı tekrar başlatır. Bu seçeneği tek bir kez snapshot yüklemesi veya doldurma için kullanın. Varsayılan değer: False.

Notes

Tetiklenen boru hatlarında SCD Tip 1 hedefleri için, bir create_auto_cdc_from_snapshot_flow() akış Python veya SQL'de tanımlanmış bir veya daha fazla AUTO CDC akışla hedefi paylaşabilir. Aşağıdaki gereksinimler geçerlidir:

  • Her akışa AUTO CDC benzersiz bir isim verin.
  • Tüm akışlar için aynı sayıda tuş sırasıyla kullanın. Anlık akış anahtar isimleri, anahtar isimleriyle büyük harf duyarsızlığı AUTO CDC ile karşılaştırılır. Birden fazla AUTO CDC akış aynı anahtar isimlerini ve kılıfı kullanmalıdır.
  • Her akışın anlık versiyonu ve sıralama sütunu AUTO CDC için tam olarak aynı veri türünü kullanın.
  • Akışta beklentileri AUTO CDC FROM SNAPSHOT tanımlamayın.
  • IGNORE NULL UPDATES değerini kullanmayın. Python'da, , ignore_null_updates_column_list, veya ignore_null_updates_except_column_list akışlar üzerinde create_auto_cdc_flow() ayarlamayınignore_null_updates.
  • Mevcut AUTO CDC FROM SNAPSHOT bir hedefe akış AUTO CDC eklerseniz, hedefin isimleri rezerve AUTO CDC edilmiş sistem sütunlarıyla çelişen kullanıcı sütunları içermemelidir.
  • Bu deseni SCD Tip 2 veya iki zamanlık hedefler için kullanmayın.

Örneğin, bkz. Bir AUTO CDC SCD Tip 1 tablosuna bir geri doldurma ekleyin.

source bağımsız değişkenini uygula

create_auto_cdc_from_snapshot_flow() işlevi, source bağımsız değişkenini içerir. Geçmiş anlık görüntüleri işlemek için, source bağımsız değişkeninin, işlenecek anlık görüntü verilerini içeren bir Python DataFrame ve bir anlık görüntü sürümü olmak üzere iki değer döndüren bir Python lambda işlevi olması beklenir.

Fonksiyonun ilk çağrısı anlık bir DataFrame ve versiyonunu döndürmelidir. İlk çağrı dönerse None, pipeline güncellemesi başarısız olur. En az bir anlık görüntü işlendikten sonra, ek bir anlık fotoğraf bulunmadığını belirtmek için geri None dön.

Lambda işlevinin imzası aşağıdadır:

lambda Any => Optional[(DataFrame, Any)]
  • Lambda işlevine verilen bağımsız değişken, en son işlenmiş anlık görüntü sürümüdür.
  • Lambda işlevinin dönüş değeri veya iki değerden oluşan bir tanımlama grubudur None : Tanımlama grubunun ilk değeri, işlenecek anlık görüntüyü içeren bir DataFrame'dir. Öbeğin ikinci değeri, anlık görüntünün mantıksal düzenini temsil eden anlık görüntü sürümüdür.

Lambda işlevini uygulayan ve çağıran bir örnek:

def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Tuple[DataFrame, Optional[int]]:
 if latest_snapshot_version is None:
   return (spark.read.load("filename.csv"), 1)
 else:
   return None

create_auto_cdc_from_snapshot_flow(
  # ...
  source = next_snapshot_and_version,
  # ...
)

Lakeflow işlem hatları çalışma zamanı, işlevi içeren create_auto_cdc_from_snapshot_flow() işlem hattı her tetiklendiğinde aşağıdaki adımları gerçekleştirir:

  1. Sonraki anlık görüntüdeki DataFrame'i ve buna karşılık gelen anlık görüntü sürümünü yüklemek için next_snapshot_and_version işlevini çalıştırır.
  2. İlk çağrı DataFrame döndürmezse, güncelleme başarısız olur. Daha sonraki bir çağrı DataFrame döndürmezse, çalıştırma sonlanır ve pipeline güncellemesi tamamlanmış olarak işaretlenir.
  3. Yeni anlık görüntüdeki değişiklikleri algılar ve bunları hedef tabloya artımlı olarak uygular.
  4. Sonraki anlık görüntüyü ve sürümünü yüklemek için 1. adıma döner.
  • Son güncelleştirme: