Lakebase'e bağlanma

Yerleşik toplu işleme, otomatik yeniden deneme ve çalışma alanı yönetimli kimlik doğrulamasıyla Lakebase'e yazmak için Structured Streaming'i kullanın.

Lakebase havuzu ne zaman kullanılır?

Lakebase'e düşük gecikme süreli akış yazma işlemleri için Lakebase havuzu kullanın. Bu alıcı, toplu işleme, bağlantı yönetimi ve hata işleme için özel foreachBatch işlevleri uygulamanızı gerektirmez.

Yaygın kullanım örnekleri şunlardır:

• İşletimsel panolar veya müşteriye yönelik özellikler için uygulama veritabanlarını gerçek zamanlı olarak güncelleştirin.
• Toplu veya filtrelenmiş akış sonuçları gibi sürekli değişen verileri işlemsel bir veritabanına eşitleyin.
• Gerçek zamanlı modu kullanarak bir Yapılandırılmış Akış sorgusunun çıkışını bir Lakebase tablosuna saniyenin altında gecikme süresiyle yazın.

Lakebase'den Lakehouse içindeki Delta Lake tablolarına veri eşitlemek için, yani ters yönde, bkz. Lakebase Değişiklik Veri Akışı.

Gereksinimler

• Databricks Runtime 18 ve üzeri
• Ayrılmış veya standart erişim modlarıyla klasik işlem.
• Lakebase veritabanı

Veritabanına bağlan

Lakebase havuzu aşağıdaki bağlantı yöntemlerini destekler:

Unity Kataloğu'yla kaydedilen Lakebase tabloları

Unity Kataloğu'na kayıtlı Lakebase tabloları için bağlayıcı kimlik bilgilerini otomatik olarak yönetir ve sorguyu çalıştıran kullanıcı veya hizmet sorumlusunun kimliğini kullanır. Tablo yoksa, bağlayıcı tabloyu oluşturur.

Bir Lakebase veritabanını Unity Kataloğu'na kaydetmek için bkz. Unity Kataloğu'nda Lakebase veritabanını kaydetme.

Lakebase tablosuna yazmak için, tam nitelikli tablo adı olarak .toTable() kullanıp catalog.schema.table yöntemini kullanın. Aşağıdaki örnek, gerekli seçeneklerin yanı sıra isteğe bağlı upsertkey seçeneği gösterir:

Python

(df.writeStream
  .outputMode("update")
  .option("upsertkey", "<primary-key-column>")  # Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .toTable("<catalog>.<schema>.<table>")
)

Scala

df.writeStream
  .outputMode("update")
  .option("upsertkey", "<primary-key-column>")  // Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .toTable("<catalog>.<schema>.<table>")

Aşağıdaki yer tutucuları değiştirin:

• <catalog>.<schema>.<table>: Hedef tablonun tam adı. catalog, Lakebase veritabanını kaydettiğinizde oluşturduğunuz Unity Kataloğu kataloğudur. Bkz. Unity Kataloğu'nda Bir Lakebase veritabanını kaydetme. Tablo yoksa, bağlayıcı bunu oluşturur.
• <primary-key-column>: İsteğe bağlı. Upsert anahtarını oluşturan sütunların virgülle ayrılmış listesi, örneğin id veya user_id,event_type. upsertkey öğesini atarsanız, sink anahtarı hedef tablonun birincil anahtarından çıkarır. Bkz. Upsert davranışı.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Sorgunun denetim noktası verilerini depoladığı bir Unity Catalog birim yolu. Bulut nesnesi depolama URI'sini de kullanabilirsiniz. Konum, yerel diskte değil, yazabileceğiniz depolama alanı olmalı ve her akış sorgusu için benzersiz olmalıdır. Bu, hedef tablodan bağımsızdır. Bkz . Yapılandırılmış Akış denetim noktaları.

batchsize ve batchinterval gibi isteğe bağlı yapılandırmalar için, Yapılandırma seçenekleri bölümüne bakın.

Unity Kataloğu'yla kaydedilmeyen Lakebase tabloları

Unity Kataloğu'na kayıtlı olmayan Lakebase tabloları için bağlayıcı kimlik bilgilerini otomatik olarak yönetir ve sorguyu çalıştıran kullanıcı veya hizmet sorumlusunun kimliğini kullanır. Tablo yoksa, bağlayıcı tabloyu oluşturur.

Lakebase tablosuna yazmak için endpoint ve dbtable seçeneklerini kullanın. Aşağıdaki örnek isteğe bağlı database ve upsertkey seçenekleri de içerir:

Python

(df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("endpoint", "<project-id>.<branch-id>.<endpoint-id>")
  .option("database", "<database>")  # Optional. Defaults to databricks_postgres.
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-column>")  # Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()
)

Scala

df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("endpoint", "<project-id>.<branch-id>.<endpoint-id>")
  .option("database", "<database>")  // Optional. Defaults to databricks_postgres.
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-column>")  // Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()

Aşağıdaki yer tutucuları değiştirin:

• <project-id>.<branch-id>.<endpoint-id>: Lakebase uç noktanız. Computes sekmesindeki Kimlik Al menüsünde, biçimi olan projects/<project-id>/branches/<branch-id>/endpoints/<endpoint-id> içinde üç değerin tümünü bulabilirsiniz. Bkz . İşlem tanımlayıcıları.
• <database>: İsteğe bağlı. Hedef Postgres veritabanının adı. Varsayılan değer databricks_postgres’dır. Bkz. Veritabanlarını yönetme.
• <schema>.<table>: schema.table biçimindeki hedef tablo. Şemayı atlarsanız havuz şemayı public kullanır. Harf veya alt çizgiyle başlayan ve yalnızca harf, sayı ve alt çizgi içeren basit tanımlayıcılar kullanın; tırnak içine alınmış tanımlayıcılar ve kısa çizgi gibi özel karakterler desteklenmez.
• <primary-key-column>: İsteğe bağlı. Upsert anahtarını oluşturan sütunların virgülle ayrılmış listesi, örneğin id veya user_id,event_type. upsertkey öğesini atarsanız, sink anahtarı hedef tablonun birincil anahtarından çıkarır. Bkz. Upsert davranışı.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Sorgunun denetim noktası verilerini depoladığı bir Unity Catalog birim yolu. Bulut nesnesi depolama URI'sini de kullanabilirsiniz. Konum, yerel diskte değil, yazabileceğiniz depolama alanı olmalı ve her akış sorgusu için benzersiz olmalıdır. Bu, hedef tablodan bağımsızdır. Bkz . Yapılandırılmış Akış denetim noktaları.

batchsize ve batchinterval gibi isteğe bağlı yapılandırmalar için, Yapılandırma seçenekleri bölümüne bakın.

Yapılandırma seçenekleri

Sink, tanınmayan seçeneklerde hata verir, JDBC_STREAMING_SINK_INVALID_OPTIONS.

Aşağıdaki seçenekler tüm bağlantı yöntemleri için geçerlidir:

Unity Kataloğu'yla kaydedilmeyen Lakebase tabloları

Unity Kataloğu'na kayıtlı olmayan bir Lakebase tablosuna bağlandığınızda aşağıdaki seçenekler geçerlidir:

Upsert davranışı

Upsert anahtarları varsa — bunlar ya upsertkey ile belirtilir ya da hedef tarafından tablonun birincil anahtarlarından çıkarılır — hedef, PostgreSQL'in INSERT INTO ... ON CONFLICT (<upsert_key>) DO UPDATE SET ... söz dizimini kullanarak tabloda upsert işlemi yapar.

Hiçbir upsert anahtarı olmadığında, havuz eklemeler gerçekleştirir. Sorgunun çıkış modunun upsert veya insert davranışı üzerinde hiçbir etkisi yoktur.

upsertkey sütunları şunları yapmalıdır:

• DataFrame sütunlarının boş olmayan bir alt kümesi olun.
• Hedef tablodaki PRIMARY KEY ile tam olarak eşleştir. Belirttiğiniz sütunlar birincil anahtarla eşleşmiyorsa sorgu başarısız olur.
• Sayısal veya dize türleri gibi benzer türler olabilir. Eşzamanlı yazma işlemleri sırasında veritabanı kilitlenmelerini önlemek için havuz, satırları her toplu iş içindeki upsert anahtarına göre sıralar. Upsert anahtarları karmaşık veya yapı türlerini desteklemez.

Sütun adları, ayrılmış anahtar kelimeleri ve büyük/küçük harf karışık adları ele alan PostgreSQL varsayılanı olan "çift tırnak ile otomatik olarak alıntılanır.

Tablo ve şema adları, harf veya alt çizgiyle başlayan ve yalnızca harf, sayı ve alt çizgi içeren basit tanımlayıcılar kullanmalıdır. Havuz, tablo veya şema adlarında tırnak içine alınmış tanımlayıcıları veya kısa çizgi gibi özel karakterleri desteklemez.

Performans Ayarlama

Toplu işleme ve geri basınç

Koşullardan herhangi biri karşılandığında temizleme tetiklenir:

• Arabellek, varsayılan değeri batchsize olan 1000 satıra ulaşır.
• Arabellek yaşı, varsayılan değeri batchinterval olan 100 milliseconds değerini aşıyor.

Veritabanı gelen veri hızına ayak uyduramadığında, alıcı geri basıncı kaynağa doğru üst akışa iletir.

Gecikme süresi ve aktarım hızı kılavuzu:

• Gerçek zamanlı moddaki düşük gecikmeli iş yüklerinde, boşaltma öncesindeki azami sürenin daha kısa olmasını sağlamak için batchinterval değerini azaltın. Kavramlar için Yapılandırılmış Akış'ta Gerçek Zamanlı Mod bölümüne, kod örneği için Gerçek Zamanlı Mod Örnekleri bölümüne bakın.
• Yüksek işlem hacimli iş yükleri için, işlem başına ek yükü azaltmak amacıyla batchsize değerini artırın.

Bağlantı davranışı

Sink, executorlarda bağlantı havuzu kullanır. Varsayılan olarak, her görev bir veritabanı bağlantısı kullanır.

Databricks, her bağlantı için varsayılan görev değerini 1 kullanmanızı önerir. Her bağlantı için görev sayısını artırırsanız, bağlantı çakışmalarına neden olabilir ve yüksek aktarım hızı bağlantıları için gecikme sürelerini artırabilirsiniz.

Görevlerin bağlantılara oranını yapılandırmak için Spark yapılandırmasını ayarlayın spark.databricks.sql.streaming.jdbc.tasksPerConnection . Hedef veritabanında düşük bir bağlantı sınırı varsa, shuffle bölüm sayısını azaltın veya spark.databricks.sql.streaming.jdbc.tasksPerConnection değerini artırın.

Sink, bağlantı hataları, kilitlenmeler ve oran sınırlaması dahil geçici JDBC hatalarını otomatik olarak yeniden dener. Havuz tüm yeniden denemeleri tüketirse sorgu başarısız olur.

Desteklenen tetikleyiciler ve çıkış modları

Triggers

Bu tabloda Yapılandırılmış Akış tetikleyici türleri için destek gösterilmektedir:

Çıkış modları

Bu tabloda Yapılandırılmış Akış çıkış modları için destek gösterilmektedir:

Sınırlama

• Sunucusuz işlem ve Lakeflow işlem hatları desteklenmez.
• Yazma hedefi olarak yalnızca Lakebase desteklenir. Dış PostgreSQL uyumlu veritabanları desteklenmez.