Opțiunea trigger() definește momentul în care Spark analizează datele sosite și generează un micro-batch:
processingTime='10 seconds': Verifică și procesează datele sosite la fiecare 10 secunde (latență scăzută).availableNow=True: (Disponibil din Spark 3.3+) Procesează toate datele disponibile în coadă în cel mai eficient număr de micro-batch-uri și apoi oprește automat fluxul. Este cel mai ieftin și eficient mod de a rula joburi de streaming pe clustere cloud pornite periodic prin cron.
Exemplu: Procesare incrementală cu oprire automată (availableNow)
query = df_curat.writeStream \
.format("delta") \
.outputMode("append") \
.trigger(availableNow=True) \
.option("checkpointLocation", "/tmp/checkpoints/batch_inc") \
.start("/mnt/lake/destinatie")
query.awaitTermination()
print("Toate datele disponibile au fost procesate cu succes!")
Întrebări & Exemple de Cod:
Configurarea Declanșării Micro-batch-urilor cu triggerAi o nelămurire despre utilizarea comenzii
Configurarea Declanșării Micro-batch-urilor cu triggersau ai descoperit un truc / fragment de cod util? Distribuie-l cu ceilalți programatori!Fii primul care pune o întrebare sau distribuie un fragment de cod / prompt!
Nu au fost adăugate încă întrebări sau exemple de cod pentru această pagină. Dacă ai întâlnit o problemă sau ai un snippet util, adaugă-l mai jos pentru a-i ajuta pe ceilalți.