create_sink

create_sink() işlevi Apache Kafka veya Azure Event Hubs gibi bir olay akış hizmetine ya da bildirim temelli bir işlem hattından Delta tablosuna yazar. işleviyle create_sink() bir havuz oluşturduktan sonra, havuza veri yazmak için ekleme akışında veya güncelleştirme akışında havuzu kullanırsınız. Ekleme ve güncelleştirme akışları, işleviyle create_sink() desteklenen tek akış türleridir. create_auto_cdc_flowgibi diğer akış türleri desteklenmez. Lakeflow işlem hatlarındaki diğer havuz türleri hakkında ayrıntılı bilgi için bkz. Lakeflow işlem hatlarındaki havuzlar.

Delta sink, Unity Catalog'un dış ve yönetilen tablolarını ve Hive metastore'un yönetilen tablolarını destekler. Tablo adları tam olarak tanımlanmış olmalıdır. Örneğin, Unity Kataloğu tablolarında üç katmanlı tanımlayıcı kullanılmalıdır: <catalog>.<schema>.<table>. Hive meta veri deposu tablolarında <schema>.<table>kullanılmalıdır.

Uyarı

• tam yenileme güncellemesi çalıştırıldığında havuzlardaki veri temizlenmez. Yeniden işlenmiş veriler havuza eklenir ve mevcut veriler değiştirilmez.
• API'de sink ile ilgili beklentiler desteklenmez.

Sözdizimi

from pyspark import pipelines as dp

dp.create_sink(name=<sink_name>, format=<format>, options=<options>)

Parametreler

Parametre	Türü	Description
name	str	Gerekli. Lavaboyu tanımlayan ve lavaboya başvurmak ve yönetmek için kullanılan bir dize. Havuz adları, işlem hattının parçası olan tüm kaynak kodu dosyaları da dahil olmak üzere işlem hattı için benzersiz olmalıdır.
format	str	Gerekli. çıkış biçimini tanımlayan bir dize( kafka veya delta).
options	dict	{"key": "value"} olarak biçimlendirilmiş lavabo seçenekleri listesi; burada anahtar ve değer her ikisi de stringdir. Kafka ve Delta havuzları tarafından desteklenen tüm Databricks Runtime seçenekleri desteklenir. Kafka seçenekleri için bkz. Kafka Yapılandırılmış Akış işleyicisini yapılandırma. Delta seçenekleri için bkz. Delta Lake tablolarını havuz olarak kullanma.

Örnekler

from pyspark import pipelines as dp

# Create a Kafka sink
dp.create_sink(
  "my_kafka_sink",
  "kafka",
  {
    "kafka.bootstrap.servers": "host:port",
    "topic": "my_topic"
  }
)

# Create an external Delta table sink with a file path
dp.create_sink(
  "my_delta_sink",
    "delta",
    { "path": "/path/to/my/delta/table" }
)

# Create a Delta table sink using a table name
dp.create_sink(
  "my_delta_sink",
    "delta",
    { "tableName": "my_catalog.my_schema.my_table" }
)