AUTO CDC API'leri: İşlem hatları ile değişiklik verilerini yakalamayı basitleştirme

Lakeflow işlem hatları, AUTO CDC ve AUTO CDC FROM SNAPSHOT API’leriyle değişiklik veri yakalamayı (CDC) basitleştirir. Bu API'ler, CDC akışından veya veritabanı anlık görüntülerinden yavaş değişen boyutların (SCD) Tür 1 ve Tür 2'nin işlem karmaşıklığını otomatikleştirir. AUTO CDC API, değişiklikleri iki zaman boyutu (Beta) arasında kaydeden bitemporal izlemeyi de destekler. SCD Tür 1 ve Tür 2 hakkında daha fazla bilgi edinmek için bkz. Veri yakalamayı ve anlık görüntüleri değiştirme. Bitemporal izleme hakkında daha fazla bilgi için bkz.: Bitemporal AUTO CDC.

Uyarı

AUTO CDC API'ler, APPLY CHANGES API'lerin yerini alır ve aynı söz dizimine sahip. APPLY CHANGES API'ler hala kullanılabilir durumdadır, ancak Databricks yerine AUTO CDC API'lerin kullanılmasını önerir.

Kullandığınız API, değişiklik verilerinizin kaynağına bağlıdır:

  • AUTO CDC: Kaynak veritabanında CDC akışı etkinleştirilmişse bunu kullanın. AUTO CDC değişiklik veri akışından (CDF) değişiklikleri işler. Hem istemci SQL hem de Python arabirimlerinde desteklenir.
  • AUTO CDC FROM SNAPSHOT: Kaynak veritabanında CDC etkinleştirilmediğinde ve yalnızca anlık görüntüler kullanılabiliyorsa bunu kullanın. Bu API, değişiklikleri belirlemek için anlık görüntüleri karşılaştırır ve sonra bunları işler. Hem istemci SQL hem de Python arabirimlerinde desteklenir.

SCD Tip 1 hedefleri için, bir AUTO CDC akışıyla bir veya daha fazla AUTO CDC FROM SNAPSHOT akışı birleştirebilirsiniz. Bir kaynak periyodik otoritatif anlık görüntüler ve anlık görüntüler arasında daha düşük gecikmeli bir CDC beslemesi sağlıyorsa bu deseni kullanın.

Her iki API de SCD Tür 1 ve Tür 2 kullanılarak tabloların güncelleştirilmesini destekler:

  • Kayıtları doğrudan güncelleştirmek için SCD Tür 1'i kullanın. Güncelleştirilmiş kayıtlar için geçmiş korunmaz.
  • Tüm güncelleştirmelerde veya belirtilen sütun kümesi güncelleştirmelerinde kayıtların geçmişini korumak için SCD Tür 2'yi kullanın.

Yalnızca AUTO CDC için, değişiklikleri iki zaman boyutunda izlemek üzere SCD Tür 2 geçmişini genişleten bitemporal depolamayı da kullanabilirsiniz: iş zamanı ve sistem zamanı. Bitemporal, Beta sürümündedir. Bkz. Bitemporal AUTO CDC.

AUTO CDC ayrıca, değişiklik kaydının sütunların yalnızca bir alt kümesini güncelleştirdiği kısmi güncelleştirmeleri de destekler. Bkz. Kısmi güncelleştirmeleri uygulama.

API'ler AUTO CDC Apache Spark Bildirimli İşlem Hatları tarafından desteklenmez.

Sentaks ve diğer başvurular için bkz. AUTO CDC INTO (pipelines), create_auto_cdc_flow ve create_auto_cdc_from_snapshot_flow.

Uyarı

Bu API'leri, kaynak verideki değişikliklere göre pipeline'larınızdaki tabloları güncellemek için kullanın. Delta tablolarında satır düzeyi değişiklik bilgilerini kaydetmeyi ve sorgulamayı öğrenmek için bkz. Azure Databricks değişiklik veri akışını kullanma.

Gereksinimler

CDC API'lerini kullanmak için işlem hattınızın sunucusuz Lakeflow işlem hatlarını veya Lakeflow işlem hatlarınıPro veya Advanced sürümlerini kullanacak şekilde yapılandırılması gerekir.

AUTO CDC nasıl çalışır?

AUTO CDC ile CDC işlemesi gerçekleştirmek için bir akış tablosu oluşturun ve ardından SQL'de AUTO CDC ... INTO deyimini veya Python’da create_auto_cdc_flow() işlevini kullanarak kaynak, anahtarlar ve değişiklik akışı için sıralama belirtin. Sıralama ve SCD mantığının nasıl çalıştığına ilişkin bir açıklama için bkz. Veri yakalamayı ve anlık görüntüleri değiştirme. AUTO CDC örneklerine bakın.

Değişiklik akışı olan bir kaynaktan ilk hidrasyon için bir akışla AUTO CDC kullanın once ve ardından değişiklik akışını işlemeye devam edin. Bkz . AUTO CDC kullanarak dış RDBMS tablosunu çoğaltma.

Söz dizimi ayrıntıları için bkz. AUTO CDC INTO (pipelines) veya create_auto_cdc_flow.

AUTO CDC ANLIK GÖRÜNTÜDEN nasıl çalışır?

AUTO CDC FROM SNAPSHOT sıralı anlık görüntüleri karşılaştırarak kaynak verilerdeki değişiklikleri belirler. Delta tablosundan, bulut depolama dosyalarından veya JDBC'den anlık görüntüleri doğrudan okuyabilirsiniz. Hem istemci SQL hem de Python arabirimlerinde desteklenir.

CDC işlemini yapmak için AUTO CDC FROM SNAPSHOT, bir akış tablosu oluşturun ve ardından akışı tanımlayın:

  • SQL'de, CREATE FLOW ifadesinin AUTO CDC ... FROM SNAPSHOT biçimini veya CREATE STREAMING TABLE öğesinin gömülü FLOW AUTO CDC yan tümcesini kullanın. Kaynak, zorunlu FROM SNAPSHOT (snapshot_query) bir madde ve işlenecek bir sonraki anlık görüntü sürümünü seçen WITH VERSION (version_query) isteğe bağlı bir maddedir. Bkz CREATE FLOW (pipelines).
  • Python'da, anlık görüntüyü, anahtarları ve diğer argümanları belirtmek için create_auto_cdc_from_snapshot_flow() fonksiyonunu kullanın. Bkz. create_auto_cdc_from_snapshot_flow.

İki veri alımı modeli ve her birinin ne zaman kullanılacağı hakkında ayrıntılı bilgi için bkz Anlık görüntü işleme desenleri. AUTO CDC FROM SNAPSHOT örneklerine bakın.

AUTO CDC FROM SNAPSHOT ve AUTO CDC'yi birleştirin

Anlık versiyon ve CDC sıralama sütunu tek bir sıralama etki alanı oluşturur. En son anlık görüntüden daha yeni bir CDC etkinliği, anlık görüntüden önceliklidir. Daha yeni bir anlık görüntü, eski CDC olaylarına göre önceliklidir ve snapshot'ta olmayan anahtarları snapshot versiyonunda silinmiş olarak kabul eder. Daha yeni CDC etkinlikleri bu anahtarları koruyabilir veya geri getirebilir.

Birleşik bir hedef, aşağıdaki gereksinimleri karşılamalıdır:

  • Hedef sistem SCD Tip 1 kullanır.
  • Hedefin tam olarak bir AUTO CDC FROM SNAPSHOT akışı ve bir veya daha fazla AUTO CDC akışı vardır.
  • Her AUTO CDC akışın kendine özgü bir adı vardır.
  • Tüm akışlar aynı sayıda anahtar kullanır ve aynı sırayı izler. Anlık akış anahtar isimleri, AUTO CDC anahtar isimleriyle büyük/küçük harf duyarsız olarak karşılaştırılır. Birden fazla AUTO CDC akış aynı anahtar adlarını ve büyük/küçük harf kullanımını kullanmalıdır.
  • Snapshot versiyonu ve her CDC dizileme sütunu tam olarak aynı veri türüne sahiptir.
  • AUTO CDC FROM SNAPSHOT Akış beklentileri tanımlamaz.
  • AUTO CDC IGNORE NULL UPDATES veya onun sütun listesi varyantlarını akışlar kullanmaz.
  • Boru hattı tetiklemeli mod kullanır.
  • Bir AUTO CDC FROM SNAPSHOT akışını mevcut bir AUTO CDC hedefine eklerseniz, hedef rezerve edilmiş AUTO CDC sistem sütunlarıyla çakışan adlara sahip kullanıcı sütunları içermemelidir.

Hem AUTO CDC hem de AUTO CDC FROM SNAPSHOT akışları Python veya SQL pipeline arayüzünü kullanabilir. SQL ve Python akışlarını aynı hedefte karıştırabilirsiniz.

once=True Python'da veya ONCE SQL'de, AUTO CDC akışları sonraki güncellemelerde yeni olayları işlemeye devam ederken, bir anlık görüntüyü bir kez geri doldurmak için AUTO CDC FROM SNAPSHOT akışında kullanın. Hedefin tam yenilenmesi, tek seferlik snapshot akışını yeniden çalıştırır. Tam bir örnek için bkz. Bir AUTO CDC SCD Tip 1 tablosuna geri doldurma ekleme.

Bu desen SCD Tip 2 veya iki zamanlı hedefleri desteklemez.

Sıralama için birden çok sütun kullanma

Birden çok sütuna göre sıralama yapmak için (örneğin, bağlamları çözmek için bir zaman damgası ve kimlik numarasını kullanarak), bunları birleştirmek için bir STRUCT kullanın. API ilk alana göre sipariş eder ve bir eşitlik durumunda ikinci alanı dikkate alır ve bu şekilde devam eder.

SQL

SEQUENCE BY STRUCT(timestamp_col, id_col)

Piton

sequence_by = struct("timestamp_col", "id_col")

AUTO CDC örnekleri

Aşağıdaki örneklerde, değişiklik veri akışı kaynağı kullanılarak SCD Tür 1 ve Tür 2 işleme gösterilmektedir. Örnek veriler yeni kullanıcı kayıtları oluşturur, kullanıcı kaydını siler ve kullanıcı kayıtlarını güncelleştirir. SCD Tür 1 örneğinde, son UPDATE işlemler geç gelirler ve olayların sırasız işlenmesini göstererek hedef tablodan çıkarılır.

Bu örneklerde kullanılan giriş kayıtları aşağıda verilmiştir. Bu veriler, Örnek veri oluşturma bölümünde sorgu çalıştırılarak oluşturulur.

userId name city Operasyon sıraNumarası
124 Raul Oaxaca INSERT 1
123 Isabel Monterrey INSERT 1
125 Mercedes Tijuana INSERT 2
126 Zambak Cancun INSERT 2
123 sıfır sıfır DELETE 6
125 Mercedes Guadalajara UPDATE 6
125 Mercedes Mexicali UPDATE 5
123 Isabel Chihuahua UPDATE 5

Örnek veri oluşturma sorgusundaki son satırın açıklamasını kaldırdığınızda, tablonun kesilmesini (tabloyu temizleme) belirten aşağıdaki kaydı sequenceNum=3 ekler:

userId name city Operasyon sıraNumarası
sıfır sıfır sıfır TRUNCATE 3

Uyarı

Aşağıdaki örneklerin tümü hem hem DELETE de TRUNCATE işlemleri belirtme seçeneklerini içerir, ancak her biri isteğe bağlıdır.

Örnek veri oluşturma

Örnek bir veri kümesi oluşturmak için aşağıdaki deyimleri çalıştırın. Bu kodun bir işlem hattı tanımının parçası olarak çalıştırılması amaçlanmamıştır. Bunu dönüştürmeler klasörü yerine işlem hattınızın keşif klasöründen çalıştırın.

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.users_cdf
AS SELECT
  col1 AS userId,
  col2 AS name,
  col3 AS city,
  col4 AS operation,
  col5 AS sequenceNum
FROM (
  VALUES
  -- Initial load.
  (124, "Raul",     "Oaxaca",      "INSERT", 1),
  (123, "Isabel",   "Monterrey",   "INSERT", 1),
  -- New users.
  (125, "Mercedes", "Tijuana",     "INSERT", 2),
  (126, "Lily",     "Cancun",      "INSERT", 2),
  -- Isabel is removed from the system and Mercedes moved to Guadalajara.
  (123, null,       null,          "DELETE", 6),
  (125, "Mercedes", "Guadalajara", "UPDATE", 6),
  -- This batch of updates arrived out of order. The batch at sequenceNum 6 is the final state.
  (125, "Mercedes", "Mexicali",    "UPDATE", 5),
  (123, "Isabel",   "Chihuahua",   "UPDATE", 5)
  -- Uncomment to test TRUNCATE.
  -- ,(null, null,      null,          "TRUNCATE", 3)
);

SCD Tür 1 güncelleştirmelerini işleme

SCD Tür 1, her kaydın yalnızca en son sürümünü tutar. Aşağıdaki örnek, yukarıda oluşturulan değişiklik veri akışından okur ve değişiklikleri akış tablosu hedefine uygular. Bu kodu çalıştırmak için bir boru hattı gerekir. Bkz. Lakeflow Pipelines Düzenleyicisi ile ETL işlem hatlarını geliştirme ve hatalarını ayıklama.

Piton

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_current")

dp.create_auto_cdc_flow(
  target = "users_current",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  apply_as_truncates = expr("operation = 'TRUNCATE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = 1
)

SQL

CREATE OR REFRESH STREAMING TABLE users_current;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_current
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
APPLY AS TRUNCATE WHEN
  operation = "TRUNCATE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 1;

SCD Tür 1 örneğini çalıştırdıktan sonra hedef tablo aşağıdaki kayıtları içerir:

userId name city
124 Raul Oaxaca
125 Mercedes Guadalajara
126 Zambak Cancun

Kullanıcı 123 (Isabel) silindi ve görünmüyor. Kullanıcı 125 (Mercedes) yalnızca en son şehri (Guadalajara) gösterir çünkü SCD Type 1 önceki değerlerin üzerine yazar. Daha sonraki bir güncelleştirme geldiğinden, önceki UPDATEsequenceNum=5 sürümü sequenceNum=6 bırakıldı.

Örneği TRUNCATE kaydı yorum satırı olmadan çalıştırdıktan sonra, tablo sequenceNum=3'de temizlenir. Bu, 124 ve 126 kayıtlarının tabloda olmadığı ve nihai hedef tablonun yalnızca aşağıdaki kaydı içerdiği anlamına gelir:

userId name city
125 Mercedes Guadalajara

SCD Tür 2 güncelleştirmelerini işleme

SCD Tür 2, __START_AT__END_AT bir kaydın her sürümü için yeni satırlar oluşturarak ve her sürümün ne zaman etkin olduğunu gösteren sütunlar oluşturarak değişikliklerin tam geçmişini korur.

Piton

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2"
)

SQL

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2;

SCD Tür 2 örneğini çalıştırdıktan sonra hedef tablo aşağıdaki kayıtları içerir:

userId name city __START_AT __END_AT
123 Isabel Monterrey 1 5
123 Isabel Chihuahua 5 6
124 Raul Oaxaca 1 sıfır
125 Mercedes Tijuana 2 5
125 Mercedes Mexicali 5 6
125 Mercedes Guadalajara 6 sıfır
126 Zambak Cancun 2 sıfır

Tablo, eksiksiz geçmişi korur. Kullanıcı 123'ün iki sürümü vardır (silindiğinde 6. sırada sona erdi). Kullanıcı 125'in şehir değişikliklerini gösteren üç sürümü vardır. ile __END_AT = null kayıtlar şu anda etkindir.

SCD Tür 2 ile sütun alt kümesini izleme

Varsayılan olarak, HERHANGI bir sütun değeri değiştiğinde SCD Tür 2 yeni bir sürüm oluşturur. İzlenmesi gereken sütunların bir alt kümesini belirtebilirsiniz, böylece diğer sütunlarda yapılan değişiklikler yeni bir geçmiş kaydı oluşturmak yerine geçerli sürümü yerinde güncelleştirir.

Aşağıdaki örnek city sütununu geçmiş izlenmesinden hariç tutar:

Piton

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.view
def users():
  return spark.readStream.table("main.cdc_tutorial.users_cdf")

dp.create_streaming_table("users_history")

dp.create_auto_cdc_flow(
  target = "users_history",
  source = "users",
  keys = ["userId"],
  sequence_by = col("sequenceNum"),
  apply_as_deletes = expr("operation = 'DELETE'"),
  except_column_list = ["operation", "sequenceNum"],
  stored_as_scd_type = "2",
  track_history_except_column_list = ["city"]
)

SQL

CREATE OR REFRESH STREAMING TABLE users_history;

CREATE FLOW apply_cdc AS AUTO CDC INTO
  users_history
FROM
  stream(main.cdc_tutorial.users_cdf)
KEYS
  (userId)
APPLY AS DELETE WHEN
  operation = "DELETE"
SEQUENCE BY
  sequenceNum
COLUMNS * EXCEPT
  (operation, sequenceNum)
STORED AS
  SCD TYPE 2
TRACK HISTORY ON * EXCEPT
  (city)

city Değişiklikler izlenmediği için, şehir güncelleştirmeleri yeni bir sürüm oluşturmak yerine geçerli satırın üzerine yazılır. Hedef tablo aşağıdaki kayıtları içerir:

userId name city __START_AT __END_AT
123 Isabel Chihuahua 1 6
124 Raul Oaxaca 1 sıfır
125 Mercedes Guadalajara 2 sıfır
126 Zambak Cancun 2 sıfır

AUTO CDC FROM SNAPSHOT örnekleri

Aşağıdaki bölümlerde, anlık görüntüleri SCD Tür 1 veya Tür 2 hedef tablolarında işlemek için kullanma AUTO CDC FROM SNAPSHOT örnekleri verilmiştir. Bu API'nin ne zaman kullanılacağına ilişkin arka plan için bkz. Veri yakalamayı ve anlık görüntüleri değiştirme.

Bu bölümdeki örnekler arasında Python ve SQL uygulamaları yer alır. Tam SQL sözdizimi için bkz. CREATE FLOW (pipelines)

Örnek: Periyodik anlık görüntüleri işle

Anlık görüntüler düzenli ve sıralı geldiğinde bu yaklaşımı kullanın. Python arayüzü, sürümlendirme için pipeline güncelleme sırasını kullanır. SQL için, örnek yeni bir anlık görüntünün ne zaman mevcut olduğunu belirlemek için ayrı bir sürüm tablosu kullanır.

Delta tabloları, bulut depolama dosyaları ve JDBC bağlantıları gibi birden çok kaynak türünden anlık görüntüleri okuyabilirsiniz.

1. Adım: Örnek veriler oluşturma

Anlık görüntü verilerini içeren bir tablo oluşturun. İşlem hattınızın klasöründe bir not defterinden veya Databricks SQL'den explorations aşağıdaki kodu çalıştırın:

CREATE SCHEMA IF NOT EXISTS main.cdc_tutorial;

CREATE TABLE main.cdc_tutorial.snapshot (
  userId INT,
  city STRING
);

CREATE TABLE main.cdc_tutorial.snapshot_version (
  version BIGINT
);

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (1, 'Oaxaca'),
  (2, 'Monterrey'),
  (3, 'Tijuana');

INSERT INTO main.cdc_tutorial.snapshot_version VALUES (0);

2. Adım: Anlık Görüntüden Auto CDC'yi çalıştırma

Bu adımda kodu çalıştırmak için bir pipeline ihtiyacınız var. Bkz. Lakeflow Pipelines Düzenleyicisi ile ETL işlem hatlarını geliştirme ve hatalarını ayıklama.

Python için, anlık görüntü görünümü için bir kaynak türü seçin (örnek oluşturma kodu bir Delta tablosu oluşturur):

Seçenek A: Delta tablosundan okuma 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.table("main.cdc_tutorial.snapshot")
B Seçeneği: Bulut depolamadan okuma 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return spark.read.format("csv").option("header", True).load("<snapshot-path>")
C Seçeneği: JDBC'den okuma (yalnızca klasik işlem) 
from pyspark import pipelines as dp

@dp.view(name="source")
def source():
  return (spark.read
    .format("jdbc")
    .option("url", "<jdbc-url>")
    .option("dbtable", "<table-name>")
    .option("user", "<username>")
    .option("password", "<password>")
    .load()
  )

Sonra hedef tabloyu ve akışı ekleyin. SQL uygulaması, yeni bir anlık görüntünün ne zaman mevcut olduğunu belirlemek için snapshot_version tablosunu kullanır:

Piton

dp.create_streaming_table("target")

dp.create_auto_cdc_from_snapshot_flow(
  target = "target",
  source = "source",
  keys = ["userId"],
  stored_as_scd_type = 2
)

SQL

CREATE OR REFRESH STREAMING TABLE target;

CREATE FLOW target_snapshot_flow AS
AUTO CDC INTO target
FROM SNAPSHOT (
  SELECT userId, city
  FROM main.cdc_tutorial.snapshot
)
WITH VERSION (
  SELECT version
  FROM main.cdc_tutorial.snapshot_version
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
)
KEYS (userId)
STORED AS SCD TYPE 2;

İlk işlem hattı çalıştırıldıktan sonra tüm kayıtlar etkin satırlar olarak eklenir:

userId city __START_AT __END_AT
1 Oaxaca 0 sıfır
2 Monterrey 0 sıfır
3 Tijuana 0 sıfır

Uyarı

Bunun yerine SCD Tip 1'i kullanmak ve yalnızca geçerli durumu korumak için, Python'da STORED AS SCD TYPE 1 veya SQL'de stored_as_scd_type=1 ayarlayın. Bu durumda, hedef tablo __START_AT ve __END_AT sütunlarını içermez.

3. Adım: Yeni bir anlık görüntü benzet ve yeniden çalıştır

Kaynak tablosunu yeni bir anlık görüntünün gelmesini simüle etmek için güncelleyin (bu kodu pipeline'ınızın klasöründeki explorations bir notebook veya SQL dosyasından çalıştırın):

TRUNCATE TABLE main.cdc_tutorial.snapshot;

INSERT INTO main.cdc_tutorial.snapshot VALUES
  (2, 'Carmel'),
  (3, 'Los Angeles'),
  (4, 'Death Valley'),
  (6, 'Kings Canyon');

UPDATE main.cdc_tutorial.snapshot_version SET version = 1;

İşlem hattını yeniden çalıştırın. AUTO CDC FROM SNAPSHOT yeni anlık görüntüyü öncekiyle karşılaştırır ve 1 kullanıcısının silindiğini, 2. ve 3. kullanıcıların güncelleştirildiğini ve 4. ve 6. kullanıcıların eklendiğini algılar. Bu, bir değişiklik akışı oluşturur ve çıkış tablosunu oluşturmak için AUTO CDC kullanır.

SCD Tür 2 ile ikinci çalıştırmadan sonra hedef tablo aşağıdaki kayıtları içerir:

userId city __START_AT __END_AT
1 Oaxaca 0 1
2 Monterrey 0 1
2 Carmel 1 sıfır
3 Tijuana 0 1
3 Los Angeles 1 sıfır
4 Ölüm Vadisi 1 sıfır
6 Kings Kanyonu 1 sıfır

Kullanıcı 1 sonlandırıldı (silindi). 2 ve 3 kullanıcılarının her birinin şehir değişikliklerini gösteren iki sürümü vardır. 4 ve 6 kullanıcıları yeni eklendi.

SCD Tür 1 ile ikinci çalıştırmadan sonra hedef tablo yalnızca geçerli durumu gösterir:

userId city
2 Carmel
3 Los Angeles
4 Ölüm Vadisi
6 Kings Kanyonu

Örnek: Sürüm işlevlerini kullanarak anlık görüntüleri işleme

Anlık görüntü sıralama üzerinde açık denetime ihtiyacınız olduğunda bu yaklaşımı kullanın. Örneğin, aynı anda birden çok anlık görüntü geldiğinde veya anlık görüntüler sıra dışında geldiğinde bu yaklaşımı kullanın. Bir sonraki anlık görüntüyü ve sürüm numarasını nasıl seçeceğinizi tanımlarsınız. API, anlık görüntüleri artan sürüm sırasına göre işler:

  • Depolamada birden fazla anlık görüntü varsa, örnek seçici mevcut anlık görüntüleri sürüm sırasına göre işler.
  • Bir anlık görüntü sırasız gelirse (örneğin, snapshot_3 sonradan snapshot_4gelirse), örnek seçici onu hariç tutar çünkü versiyonu son belirlenmiş versiyondan büyük değildir.
  • Yeni anlık görüntüler yoksa, sürüm fonksiyonu veya sorgu sonuç vermez ve işlem gerçekleşmez.

1. Adım: Anlık görüntü dosyalarını hazırlama

Anlık görüntü verilerini içeren CSV dosyaları oluşturun ve bunları bir birime veya bulut depolama konumuna ekleyin. Dosyaları kronolojik olarak adlandırın (örneğin, snapshot_1.csv, snapshot_2.csv).

Her dosya userId ve city sütunlarını içermelidir. Örneğin:

snapshot_1.csv:

userId city
1 Oaxaca
2 Monterrey
3 Tijuana

snapshot_2.csv:

userId city
2 Carmel
3 Los Angeles
4 Ölüm Vadisi

2. Adım: AUTO CDC FROM SNAPSHOT komutunu bir sürüm fonksiyonu ile çalıştırın

Bir pipeline'da, transformations klasöründe yeni bir Python veya SQL dosyası oluşturun ve aşağıdaki kodu ekleyin. Ardından işlem hattını çalıştırın. Bkz. Lakeflow Pipelines Düzenleyicisi ile ETL işlem hatlarını geliştirme ve hatalarını ayıklama.

Piton

from pyspark import pipelines as dp
from typing import Optional, Tuple
from pyspark.sql import DataFrame

def next_snapshot_and_version(latest_snapshot_version: Optional[int]) -> Optional[Tuple[DataFrame, int]]:
  snapshot_dir = "/Volumes/main/cdc_tutorial/snapshots/" # or the location you created the sample data

  files = dbutils.fs.ls(snapshot_dir)
  snapshot_files = [f.name for f in files if f.name.startswith("snapshot_") and f.name.endswith(".csv")]

  snapshot_versions = []
  for filename in snapshot_files:
    try:
      version = int(filename.replace("snapshot_", "").replace(".csv", ""))
      snapshot_versions.append(version)
    except ValueError:
      continue

  snapshot_versions.sort()

  if latest_snapshot_version is None:
    if snapshot_versions:
      next_version = snapshot_versions[0]
    else:
      return None
  else:
    next_versions = [v for v in snapshot_versions if v > latest_snapshot_version]
    if next_versions:
      next_version = next_versions[0]
    else:
      return None

  snapshot_path = f"{snapshot_dir}snapshot_{next_version}.csv"
  df = spark.read.format("csv").option("header", True).load(snapshot_path)
  return (df, next_version)


dp.create_streaming_table("main.cdc_tutorial.target_versioned")

dp.create_auto_cdc_from_snapshot_flow(
  target = "main.cdc_tutorial.target_versioned",
  source = next_snapshot_and_version,
  keys = ["userId"],
  stored_as_scd_type = 2
)

SQL

CREATE OR REFRESH STREAMING TABLE main.cdc_tutorial.target_versioned;

CREATE FLOW target_versioned_snapshot_flow AS
AUTO CDC INTO main.cdc_tutorial.target_versioned
FROM SNAPSHOT (
  SELECT CAST(userId AS INT) AS userId, city
  FROM read_files(
    '/Volumes/main/cdc_tutorial/snapshots/',
    format => 'csv',
    header => true
  )
  WHERE _metadata.file_name = CONCAT(
    'snapshot_',
    CAST((SELECT version FROM current_snapshot_version()) AS STRING),
    '.csv'
  )
)
WITH VERSION (
  WITH snapshot_files AS (
    SELECT
      CAST(REGEXP_EXTRACT(path, 'snapshot_([0-9]+)[.]csv$', 1) AS BIGINT) AS version
    FROM list_files('/Volumes/main/cdc_tutorial/snapshots/')
    WHERE path RLIKE 'snapshot_[0-9]+[.]csv$'
  )
  SELECT version
  FROM snapshot_files
  WHERE (
    NOT EXISTS (SELECT 1 FROM last_snapshot_version())
    OR version > (SELECT version FROM last_snapshot_version())
  )
  ORDER BY version
  LIMIT 1
)
KEYS (userId)
STORED AS SCD TYPE 2;

Uyarı

Bunun yerine SCD Tip 1 kullanmak için STORED AS SCD TYPE 1 Python'da veya stored_as_scd_type=1 SQL'de ayarlayın.

işlendikten snapshot_1.csvsonra hedef tablo aşağıdaki kayıtları içerir:

userId city __START_AT __END_AT
1 Oaxaca 1 sıfır
2 Monterrey 1 sıfır
3 Tijuana 1 sıfır

işlendikten snapshot_2.csvsonra hedef tablo aşağıdaki kayıtları içerir:

userId city __START_AT __END_AT
1 Oaxaca 1 2
2 Monterrey 1 2
2 Carmel 2 sıfır
3 Tijuana 1 2
3 Los Angeles 2 sıfır
4 Ölüm Vadisi 2 sıfır

Uyarı

SCD Tür 1 için tablonun en son anlık görüntüye tam olarak benzediğini unutmayın. Aradaki fark, aşağı akış sorgularının değişiklik akışını yalnızca değiştirilen kayıtları işlemek için kullanabilmesidir.

3. Adım: Yeni anlık görüntüler ekleme

Değiştirilen verilerle (örneğin, değiştirilen şehir değerleri, yeni satırlar veya kaldırılan satırlar) depolama konumuna yeni bir CSV dosyası ekleyin. Ardından yeni anlık görüntüyü işlemek için işlem hattını yeniden çalıştırın.

Sınırlamalar

Ek kaynaklar

  • Son güncelleştirme: