Yeni dosyalar geldiğinde işleri tetikler

file varış tetikleyicilerini kullanarak amazon S3, Azure depolama veya Google Cloud Storage gibi external konuma yeni dosyalar geldiğinde işinizin çalıştırılmasını tetikleyebilirsiniz. Bu özellik, zamanlanmış bir işin verimliliği düzensiz yeni veri gelenler tarafından tehlikeye atıldığında kullanışlıdır.

Dosya varış tetikleyicileri nasıl çalışır?

Dosya varış tetikleyicileri, her dakika yeni dosyaları denetlemek için en iyi çabayı gösterir, ancak bu, temel alınan bulut depolamanın performansından etkilenebilir. Dosya varış tetikleyicileri, depolama konumundaki dosyaları listelemeyle ilişkili bulut sağlayıcısı maliyetleri dışında ek maliyetler doğurmuyor.

Bir dosya varış tetikleyicisi, Unity Kataloğu dış konumunun veya biriminin kökünü ya da bir dış konumun veya birimin alt yolu izlemek için yapılandırılabilir. Örneğin, Unity Kataloğu birimi /Volumes/mycatalog/myschema/myvolume/için, dosya varış tetikleyicisi için geçerli yollar şunlardır:

/Volumes/mycatalog/myschema/myvolume/
/Volumes/mycatalog/myschema/myvolume/mydirectory/

Dosya varış tetikleyicisi, yapılandırılan konumun tüm alt dizinlerindeki yeni dosyaları yinelemeli olarak denetler. Örneğin, konum /Volumes/mycatalog/myschema/myvolume/mydirectory/ için bir dosya varış tetikleyicisi oluşturursunuz ve bu konumda aşağıdaki alt dizinler bulunur:

/Volumes/mycatalog/myschema/myvolume/mydirectory/subdirA
/Volumes/mycatalog/myschema/myvolume/mydirectory/subdirB
/Volumes/mycatalog/myschema/myvolume/mydirectory/subdirC/subdirD

Tetikleyici, mydirectory, subdirA, subdirB, subdirC ve subdirC/subdirD içindeki yeni dosyaları denetler.

Dosya olaylarıyla dosya varış tetikleyicileri

En iyi performans için dış konumun dosya olayları için etkinleştirilmesi gerekir. Bir dış konum için dosya olayları etkinleştirildiğinde, Azure Databricks, bulut sağlayıcılarından gelen değişiklik bildirimlerini işleyerek veri alım meta verilerini izlemek için dahili bir hizmet kullanır. Bu hizmet, hizmet tarafından belirlenen sıralı saklama süresi boyunca oluşturulan veya güncelleştirilen en son dosyalar için meta verileri korur ve dosya işleme verimliliğini artırır.

Bir dış konumda dosya olaylarını etkinleştirdikten sonra dakikalar içinde, bu dış konumun kapsadığı yolları izleyen mevcut dosya varış tetikleyicileri dosya olaylarının etkinleştirilmesinden yararlanmaya başlar ve yeni tetikleyiciler saniyeler içinde avantaj sağlar.

Dış konumlardaki dosya olaylarının performans ve kapasite avantajları hakkında daha fazla bilgi için bkz. Sınırlamalar. Dosya olayları hakkında sık sorulan sorular için bkz. Dosya olayları hakkında SSS.

Başlamadan önce

Dosya varış tetikleyicilerini kullanmak için aşağıdakiler gereklidir:

• Çalışma alanında Unity Kataloğu etkinleştirilmiş olmalıdır.

• Ya bir birim ya da Unity Kataloğu'nda yapılandırılmış harici bir konum olan bir depolama konumu kullanmanız gerekir. Bkz. Unity Kataloğu birimleri nedir? ve Dış konumlara genel bakış.

  Databricks , yönetilen dosya olayları için dış konumu etkinleştirmenizi önerir. Bu dış konumlardaki birimler varsayılan olarak dosya olay desteği alır. Dosya olaylarını etkinleştirmek için dış konum sahibi olmanız veya dış konumda ayrıcalığınız olması MANAGE gerekir. Dosya olaylarının avantajları hakkında bilgi için bkz. Dosya olaylarıyla dosya varış tetikleyicileri.

• Depolama konumu üzerinde READ iznine sahip olmanız ve iş üzerinde Yönetme izinlerine sahip olmanız gerekir. İş izinleri hakkında daha fazla bilgi için Job ACL'leri bölümüne bakın.

Dosya varış tetikleyicisi ekleme

Bir işe dosya varış tetikleyicisi eklemek için:

1. Azure Databricks çalışma alanınızın sol kenar çubuğunda İşler ve İşlem Hatları öğesine tıklayın.
2. İsteğe bağlı olarak İşler ve Bana ait filtrelerini seçin.
3. İşinizin Ad bağlantısına tıklayın.
4. Sağdaki İş ayrıntıları bölmesinde Tetikleyici ekle'ye tıklayın.
5. Tetikleyici türübölümünde Dosya varışseçin.
6. Depolama konumu alanına, Unity Catalog harici konumunun kök veya alt yolunun URL'sini veya izleme için Unity Catalog hacminin kök veya alt yolunu girin.
7. (İsteğe bağlı) Çalıştırmaların ne sıklıkta tetikleneceğini denetlemek için gelişmiş seçenekleri (saniye cinsinden tetikleyiciler arasındaki en düşük süre ve Son değişiklik sonrasında saniye cinsinden bekle) yapılandırın. Kurulum örnekleri için bkz. Çalıştırmaların ne sıklıkta tetiklendiğini denetleme.
8. Yapılandırmayı doğrulamak için Bağlantıyı test et'e tıklayın.
9. Kaydet'e tıklayın.

Bu tetikleyiciyi daha sonra düzenlemek, duraklatmak veya kaldırmak için İş ayrıntıları bölmesinin Zamanlamalar ve Tetikleyiciler bölümünü kullanın. Bkz. Mevcut tetikleyiciyi yönetme.

Çalıştırmaların ne sıklıkta tetikleneceğini denetleyin

Bir dosya varış tetikleyicisindeki iki gelişmiş seçenek, dosya varışlarının işlerin çalıştırılmasına nasıl dönüştüğünü denetler. Bu seçenekler, iki yaygın hız denetimi kalıbını uygular: bekleme süresi ve sekme giderme

• Saniye cinsinden tetikleyiciler arasındaki en kısa süre: İşi bu aralık başına en fazla bir çalıştırmayla sınırlar (çalıştırmalar arasındaki bekleme süresi). Bir çalıştırma tamamlandıktan sonra, bekleme süresi sırasında gelen dosyalar, aralık geçene kadar yeni bir çalıştırma başlatmaz. Sık gelen tetiklemelerin arka arkaya çalıştırmalar oluşturmaması için, çalıştırmaların ne sıklıkta oluşturulacağını sınırlamak üzere bu seçeneği kullanın.
• Son değişiklikten sonra beklenecek süre (saniye): Bir çalıştırma başlatılmadan önce, en son dosya ulaştıktan sonra bu kadar beklenir; her yeni dosya geldiğinde sayaç sıfırlanır (art arda tetiklemeyi önleme). Dosyalar toplu olarak geldiğinde ve tüm dosyalar geldikten sonra toplu işin tamamını tek bir çalıştırmada işlemek istediğinizde bu seçeneği kullanın.

Seçeneklerden birini kendi başına ayarlayabilir veya her ikisini birlikte ayarlayabilirsiniz. Aşağıdaki örneklere bakın.

En fazla 15 dakikada bir çalıştır

Dosyalar geldikçe, ancak 15 dakikada birden daha sık olmayacak şekilde çalıştırmalar oluşturmak için aşağıdaki gelişmiş seçeneği ayarlayın:

• Saniye cinsinden tetikleyiciler arasındaki en kısa süre: 900

Her çalıştırma tamamlandıktan sonra tetikleyici, dosyalar gelmeye devam etse bile başka bir çalıştırma başlatmadan önce 900 saniye (15 dakika) bekler. Bu, çalıştırma oluşturma işlemini 15 dakikada en fazla bir çalıştırmayla sınırlar.

Tam bir toplu işin gelmesini bekleyin

Dosyalar toplu olarak geldiğinde ve her toplu işlemi tek bir çalıştırmada işlemek istediğinizde, Son değişiklikten sonra saniye cinsinden bekle'yi toplu işler arasındaki boşluktan daha kısa ancak toplu iş içindeki dosyalar arasındaki boşluktan daha uzun bir değere ayarlayın. Örneğin, yaklaşık 5 dakikada bir yeni bir toplu iş başlatılırsa aşağıdaki gelişmiş seçeneği ayarlayın:

• Saniye cinsinden son değişikliğin ardından bekleyin: 60

Her yeni dosya zamanlayıcıyı sıfırlar, bu nedenle tetikleyici yalnızca yeni gelen olmadan 60 saniye geçtikten sonra bir çalıştırma başlatır. Bu kurulum, bir toplu iş içindeki dosyaların birbirini en fazla 60 saniye arayla takip ettiğini; böylece zamanlayıcının süresinin toplu işin ortasında dolmadığını ve toplu işler arasında 60 saniyeden fazla süre bulunduğunu; böylece ardışık toplu işlerin tek bir çalıştırmada birleşmediğini varsayar.

Sıklığı sınırlayın ve toplu işlemlerin tamamlanmasını bekleyin

Çalıştırmaların ne sıklıkla oluşturulduğunu sınırlandırmak ve ayrıca bir toplu işlemin ortasında çalıştırma başlatmaktan kaçınmak istediğinizde her iki seçeneği de birleştirebilirsiniz. Örneğin:

• Saniye cinsinden tetikleyiciler arasındaki en kısa süre: 900
• Saniye cinsinden son değişikliğin ardından bekleyin: 60

Bu yapılandırmayla tetikleyici, bir çalıştırma başlatmadan önce toplu işlemin tamamlanmasını bekler (60 saniye boyunca yeni dosya gelmezse) ve 15 dakikada birden fazla çalıştırma başlatmaz.

Varışınızda dosyaları bulma ve işleme

Dosya varış tetikleyicilerini tetikleyen dosyaları işlemek için Otomatik Yükleyici'yi kullanabilirsiniz. Otomatik Yükleyici, yeni dosyaları tam olarak bir kez işleme garantisiyle artımlı ve verimli bir şekilde işler. Örneğin, dosyaları Delta tablosuna yüklemek için aşağıdaki kod parçacığını kullanın.

Bu çözümü kullanmak için, dosya varış tetikleyicisiyle bir iş oluşturun ve aşağıdaki kodu içeren bir not defteri ekleyin. Her [REPLACE] yer tutucuyu uygun değerle değiştirin.

# Configuration
file_location = "[REPLACE]" # The same URL configured for the file arrival trigger.
checkpoint_location = "[REPLACE]" # a separate URL (outside `file_location`) used to store the Auto Loader checkpoint, which enables exactly-once processing.
sink_table = "[REPLACE]" # Delta table to write to

# Use Auto Loader to discover new files.
# Do not modify code below this line
streamingQuery = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "json") \
  .option("cloudFiles.schemaLocation", checkpoint_location) \
  .option("cloudFiles.useManagedFileEvents","true") \
  .load(file_location) \
  .writeStream \
  .option("checkpointLocation", checkpoint_location) \
  .trigger(availableNow = True) \
  .toTable(sink_table)

Yeni dosyaları özel mantıkla işlemeniz gerekiyorsa ve yalnızca yeni dosyaların URL'sini bulmak istiyorsanız, bunun yerine aşağıdaki kod parçacığında gösterildiği gibi kullanabilirsiniz foreachBatch . foreachBatch Yalnızca en az bir kez işleme garantisi sağladığını unutmayın. Daha fazla bilgi için foreachBatch kullanımına bakınız, Rastgele veri havuzlarına yazmak için foreachBatch kullanma

# Configuration
file_location = "[REPLACE]" # The same URL configured for the file arrival trigger.
checkpoint_location = "[REPLACE]" # a separate URL (outside `file_location`) used to store the Auto Loader checkpoint, which enables exactly-once processing.

def process_batch(batch_df, batch_id):
  file_url = batch_df.select("path").collect()[0].path
  # [REPLACE] Your custom function for processing newly arrived files


# Use Auto Loader to discover new files.
# Do not modify code below this line
streamingQuery = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "binaryFile") \
  .option("cloudFiles.useManagedFileEvents","true") \
  .load(file_location) \
  .drop("content") \
  .writeStream \
  .foreachBatch(process_batch) \
  .option("checkpointLocation", checkpoint_location) \
  .trigger(availableNow = True) \
  .start()

Başarısız dosya varış tetikleyicileriyle ilgili bildirimler alma

Bir dosya varış tetikleyicisinin değerlendirilmesi başarısız olursa bildirim almak için, iş hatası durumunda e-posta veya sistem hedefi bildirimlerini yapılandırın. Bkz. bir işe bildirim ekleme.

Sınırlama

• Yalnızca yeni dosyalar tetikleyicisi çalışır. Aynı ada sahip bir dosyayla var olan bir dosyanın üzerine yazmak çalıştırmayı tetiklemez.

• Depolama konumu dosya olayları için etkinleştirildiyse, var olan boş bir dosyaya içerik eklemek, yeni bir dosya gelişi olarak ele alındığından bir çalıştırma tetikler.
• Dosya olayları, bir dosyayı işlemek için FlushWithClose olayı dinler. Bazı Azure API kullanımı bu olayı yaymayabilir ve bu da dosya bulmayı geciktirebilir. Bu senaryoyla başa çıkabilmek için bkz. Klasik dosya bildirim olayları.

• Dosya varış tetikleyicisi için kullanılan yol, dış tablolar veya katalogların ve şemaların yönetilen konumlarını içermemelidir.

• Dosya varış tetikleyicisi için kullanılan yol, örneğin * veya ? gibi joker karakterler içeremez.

• Depolama konumu Unity Kataloğu'nda dış konum olarak yapılandırıldıysa ve bu dış konum dosya olayları için etkinleştirildiyse:

  • Depolama konumundaki dosya sayısıyla ilgili bir sınır yoktur.

  • Çok fazla gereksiz dosya güncelleştirmesi olduğunda tetikleyiciler zaman aşımında hata verebilir.

    Bir Unity Kataloğu dış konumunun veya biriminin alt yolu üzerinde bir dosya varış tetikleyicisi ayarlandığında, bu alt yolun dışındaki değişiklikler (dış konumun kökünde olduğu gibi) tetikleyicinin işlemesi gereken meta verileri artırabilir. Yüksek değişiklikli ortamlarda bu, tetikleyicinin işlem süresi sınırını aşmasına ve hata durumuna girmesine neden olabilir.

    Bunu önlemek için, özellikle izlemek istediğiniz alt dizine eşleyen bir Unity Kataloğu birimi oluşturun ve bu birimin kökünde dosya varış tetikleyicisini ayarlayın. Bu yaklaşım, tetikleyicinin etkili kökü olarak hedef yolunuzu yalıtarak ilgisiz kök düzeyindeki değişiklikleri azaltır ve tetikleyicinin hata durumuna geçmesini önler.

  • Mevcut bir dosya değiştirilirse ve meta verileri kayan saklama süresinin dışında kalırsa, bu değişiklik yeni bir dosyanın gelmesi olarak değerlendirilir ve bir işin çalışmasını tetikler. Yalnızca sabit dosyaları alarak bunu önleyebilir veya alma işleminin ilerleme durumunu izlemek için Otomatik Yükleyici ile dosya varış tetikleyicilerini kullanabilirsiniz.

• Depolama konumu dosya olayları için etkin değilse:

  • Azure Databricks çalışma alanında bu tür konumlarda bir dosya varış tetikleyicisi ile en fazla 50 iş yapılandırılabilir.
  • Depolama konumu en fazla 10.000 dosya içerebilir. Yapılandırılan depolama konumu bir Unity Kataloğu dış konumunun veya biriminin alt yoluysa, 10.000 dosya sınırı depolama konumunun köküne değil alt yol için geçerlidir. Örneğin, depolama konumunun kökü alt dizinleri genelinde 10.000'den fazla dosya içerebilir, ancak yapılandırılan alt dizinin 10.000 dosya sınırını aşmaması gerekir.

Ayrıca bkz . Dosya olayları sınırlamaları.

S3 ve GCS dış konumlarında var olmayan yollarda dosya varış tetikleyicileri

Yapılandırılan dizin mevcut olmadığında veya Amazon S3 veya Google Cloud Storage'dan silindiğinde, dosya varış tetikleyicileri hata olmadan değerlendirilmeye devam eder. Bu davranış, hem S3 hem de GCS'nin var olmayan, silinmiş ve boş dizinler arasında ayrım yapmaması nedeniyle oluşur.

Sonuç olarak, var olmayan veya silinmiş dizin yolunu izleyen bir dosya varış tetikleyicisi başarısız olmaz veya hata bildirimi oluşturmaz. Tetikleyici değerlendirmeye devam eder, dosya bulmaz ve dosyalar yeniden o yola eklenene kadar hiçbir görev çalıştırmasını tetiklemez. Bu bir hata koşulu değil beklenen davranıştır.