İlk Yapılandırılmış Akış iş yükünüzü çalıştırma

Bu makalede, Azure Databricks'te ilk Yapılandırılmış Akış sorgularınızı çalıştırmak için gereken temel kavramların kod örnekleri ve açıklaması sağlanır. Neredeyse gerçek zamanlı ve artımlı işleme iş yükleri için Yapılandırılmış Akış'ı kullanabilirsiniz.

Yapılandırılmış Akış, Lakeflow işlem hatlarında akış tablolarını destekleyen çeşitli teknolojilerden biridir. Databricks tüm yeni ETL, alım ve Yapılandırılmış Akış iş yükleri için Lakeflow işlem hatlarının kullanılmasını önerir. Bkz. Spark Bildirimsel İşlem Hatları.

Not

Lakeflow işlem hatları akış tablolarını bildirmek için biraz değiştirilmiş bir söz dizimi sağlarken, akış okumalarını ve dönüştürmelerini yapılandırmaya yönelik genel söz dizimi Azure Databricks tüm akış kullanım örnekleri için geçerlidir. Lakeflow işlem hatları ayrıca durum bilgilerini, meta verileri ve çok sayıda yapılandırmayı yöneterek akışı basitleştirir.

Nesne depolamadan akış verilerini okumak için Otomatik Yükleyici'yi kullanma

Aşağıdaki örnekte JSON verilerinin Otomatik Yükleyici ile yüklenmesi gösterilmektedir. Bu, biçim ve seçenekleri belirtmek için kullanılır cloudFiles . schemaLocation seçeneği şema çıkarımı ve evrimini etkinleştirir. Aşağıdaki kodu bir Databricks not defteri hücresine yapıştırın ve raw_df adlı bir DataFrame akışı oluşturmak için hücreyi çalıştırın.

file_path = "/databricks-datasets/structured-streaming/events"
checkpoint_path = "/tmp/ss-tutorial/_checkpoint"

raw_df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.schemaLocation", checkpoint_path)
    .load(file_path)
)

Azure Databricks'te yapılan diğer okuma işlemleri gibi, bir akış okuma yapılandırması da veri yüklemez. Akış başlamadan önce verilerde bir eylem tetiklemeniz gerekir.

Not

Akışkan bir DataFrame üzerinde display() çağrısı yapmak, bir akış işi başlatır. Çoğu Yapılandırılmış Akış kullanım örneğinde, akışı tetikleyen eylem havuza veri yazmak olmalıdır. Bakınız Yapılandırılmış Akış için Üretimle İlgili Dikkat Edilmesi Gerekenler.

Akış dönüştürmesi yapmak

Yapılandırılmış Akış, Azure Databricks ve Spark SQL'de kullanılabilen çoğu dönüştürmeyi destekler. Hatta MLflow modellerini UDF olarak yükleyebilir ve akış tahminlerini bir dönüşüm olarak gerçekleştirebilirsiniz.

Aşağıdaki kod örneği, Spark SQL işlevlerini kullanarak alınan JSON verilerini ek bilgilerle zenginleştirmek için basit bir dönüştürme işlemini tamamlar:

from pyspark.sql.functions import col, current_timestamp

transformed_df = (raw_df.select(
    "*",
    col("_metadata.file_path").alias("source_file"),
    current_timestamp().alias("processing_time")
    )
)

Sonuçta transformed_df , veri kaynağına ulaşan her kaydı yüklemek ve dönüştürmek için sorgu yönergeleri yer alır.

Not

Yapılandırılmış Akış, veri kaynaklarını ilişkisiz veya sonsuz veri kümeleri olarak kabul eder. Bu nedenle, bazı dönüştürmeler sınırsız sayıda öğenin sıralanması gerekeceğinden Yapılandırılmış Akış iş yüklerinde desteklenmez.

Çoğu toplama ve birçok birleştirme, filigranlar, pencereler ve çıkış moduyla durum bilgilerinin yönetilmesini gerektirir. Bkz Veri işleme eşiklerini denetlemek için filigranları uygulama.

Delta Lake'e kademeli toplu yazma yapın

Aşağıdaki örnek, belirtilen dosya yolunu ve denetim noktasını kullanarak Delta Lake'e yazar.

Önemli

Yapılandırdığınız her akış yazıcısı için her zaman benzersiz bir denetim noktası konumu belirttiğinizden emin olun. Denetim noktası, akış sorgunuzla ilişkili işlenen tüm kayıtları ve durum bilgilerini izleyerek akışınız için benzersiz kimlik sağlar.

availableNow Tetikleyicinin ayarı Yapılandırılmış Akış'a kaynak veri kümesinden daha önce işlenmemiş tüm kayıtları işlemesini ve ardından kapatmasını bildirir; böylece bir akışı çalışır durumda bırakma konusunda endişelenmeden aşağıdaki kodu güvenle yürütebilirsiniz:

target_path = "/tmp/ss-tutorial/"
checkpoint_path = "/tmp/ss-tutorial/_checkpoint"

transformed_df.writeStream
    .trigger(availableNow=True)
    .option("checkpointLocation", checkpoint_path)
    .option("path", target_path)
    .start()

Bu örnekte, veri kaynağımıza yeni kayıt ulaşmadığından, bu kodun yinelenmiş yürütülmesi yeni kayıtları almaz.

Uyarı

Yapılandırılmış Akış yürütmesi, otomatik sonlandırmanın işlem kaynaklarını kapatmasını engelleyebilir. Beklenmeyen maliyetleri önlemek için akış sorgularını sonlandırdığından emin olun.

Delta Lake'ten veri okuma, dönüştürme ve Delta Lake'e yazma

Delta Lake hem kaynak hem de havuz olarak Yapılandırılmış Akış ile çalışmak için kapsamlı desteğe sahiptir. Bkz. Delta Lake tablosu için akış okuma ve yazma işlemleri.

Aşağıdaki örnekte, Delta Lake tablosundaki tüm yeni kayıtları artımlı olarak yüklemek, bunları başka bir Delta Lake tablosunun anlık görüntüsüyle birleştirmek ve bir Delta Lake tablosuna yazmak için örnek söz dizimi gösterilmektedir:

(spark.readStream
    .table("<table-name1>")
    .join(spark.read.table("<table-name2>"), on="<id>", how="left")
    .writeStream
    .trigger(availableNow=True)
    .option("checkpointLocation", "<checkpoint-path>")
    .toTable("<table-name3>")
)

Kaynak tabloları okumak ve hedef tablolara ve belirtilen denetim noktası konumuna yazmak için yapılandırılmış uygun izinlere sahip olmanız gerekir. Veri kaynaklarınız ve havuzlarınız için ilgili değerleri kullanarak açılı ayraçlarla (<>) belirtilen tüm parametreleri doldurun.

Not

Lakeflow işlem hatları, Delta Lake işlem hatları oluşturmak ve tetikleyiciler ve denetim noktaları gibi özellikleri otomatik olarak yönetmek için tam bildirim temelli bir söz dizimi sağlar. Bkz. Spark Bildirimsel İşlem Hatları.

Kafka'dan veri okuma, dönüştürme ve Kafka'ya yazma

Apache Kafka ve diğer mesajlaşma veri yolları, büyük veri kümeleri için en düşük gecikme sürelerinden bazılarını sağlar. Azure Databricks'i kullanarak Kafka'dan alınan verilere dönüştürmeler uygulayabilir ve sonra verileri Kafka'ya geri yazabilirsiniz.

Not

Bulut nesne depolama alanına veri yazmak ek gecikme yükü ekler. Delta Lake'te bir mesajlaşma veri yolu verilerini depolamak istiyorsanız ancak akış iş yükleri için mümkün olan en düşük gecikme süresini istiyorsanız Databricks, verileri lakehouse'a almak ve aşağı akış mesajlaşma veri yolu havuzları için neredeyse gerçek zamanlı dönüştürmeler uygulamak için ayrı akış işleri yapılandırmanızı önerir.

Aşağıdaki kod örneğinde, Verileri Delta Lake tablosundaki verilerle birleştirip Kafka'ya geri yazarak Kafka'dan verileri zenginleştirmeye yönelik basit bir desen gösterilmektedir:

(spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "<server:ip>")
    .option("subscribe", "<topic>")
    .option("startingOffsets", "latest")
    .load()
    .join(spark.read.table("<table-name>"), on="<id>", how="left")
    .writeStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "<server:ip>")
    .option("topic", "<topic>")
    .option("checkpointLocation", "<checkpoint-path>")
    .start()
)

Kafka hizmetinize erişim için yapılandırılmış uygun izinlere sahip olmanız gerekir. Veri kaynaklarınız ve havuzlarınız için ilgili değerleri kullanarak açılı ayraçlarla (<>) belirtilen tüm parametreleri doldurun. Bkz. Apache Kafka'ya bağlanma.

  • Last updated on