trigger (DataStreamWriter)

Ustawia wyzwalacz dla zapytania przesyłania strumieniowego. Jeśli nie zostanie ustawiona, zapytanie jest uruchamiane tak szybko, jak to możliwe, równoważne .processingTime='0 seconds' Jednocześnie można ustawić tylko jeden parametr wyzwalacza.

Zobacz Konfigurowanie interwałów wyzwalacza strukturalnego przesyłania strumieniowego.

Obliczenia i wykonywanie potoku

W rurociągach Lakeflow rurociąg zarządza realizacją strumienia. Użyj ustawień wykonywania potoków , aby kontrolować, kiedy przetwarzane są dane potoku.

Gdy używasz DataStreamWriter bezpośrednio na serwerze do notebooków i zadań, użyj availableNow=True. Domyślne wyzwalacze, processingTime, continuous, i realTime nie są obsługiwane w tym obliczeniu. Te ograniczenia nie dotyczą trybów uruchamiania potoków bez serwera. Zobacz Ograniczenia przesyłania strumieniowego.

Składnia

trigger(*, processingTime=None, once=None, continuous=None, availableNow=None, realTime=None)

Parametry

Parameter Typ Opis
processingTime str, opcjonalnie Ciąg interwału czasu przetwarzania (na przykład '5 seconds', '1 minute'). Okresowo uruchamia zapytanie mikrobajtowe na podstawie czasu przetwarzania.
once wartość logiczna, opcjonalnie Jeśli Trueprogram przetwarza tylko jedną partię danych, kończy zapytanie. Wycofane w Databricks Runtime 11.3 LTS i wyższych. Użyj availableNow=True zamiast tego.
continuous str, opcjonalnie Nie jest obsługiwane na Azure Databricks. Ten parametr umożliwia eksperymentalny tryb ciągłego przetwarzania w Apache Spark, który różni się od trybu ciągłego potokowego. Używaj trybu czasu rzeczywistego do zapytań o niskim opóźnieniu w Structured Streaming.
availableNow wartość logiczna, opcjonalnie Jeśli Trueprogram przetwarza wszystkie dostępne dane w wielu partiach, przerywa zapytanie.
realTime str, opcjonalnie Ciąg czasu trwania partii (na przykład '5 minutes'). Przetwarza rekordy w momencie, gdy dotrą w ramach długotrwałej partii. Wartość ta kontroluje czas trwania partii, a nie opóźnienia przetwarzania rekordów. Zobacz Tryb czasu rzeczywistego w Strukturalnym Przetwarzaniu Strumieniowym.

Zwroty

DataStreamWriter

Examples

df = spark.readStream.format("rate").load()

Wykonywanie wyzwalacza co 5 sekund:

df.writeStream.trigger(processingTime='5 seconds')
# <...streaming.readwriter.DataStreamWriter object ...>

Przetwarzanie wszystkich dostępnych danych w wielu partiach:

df.writeStream.trigger(availableNow=True)
# <...streaming.readwriter.DataStreamWriter object ...>

Konfiguruj tryb czasu rzeczywistego z pięciominutowymi partiami. Dokumenty są przetwarzane w miarę ich nadejścia w każdej partii. Użyj obsługiwanego źródła, sink i typu obliczeniowego:

kafka_df = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "<bootstrap-servers>")
    .option("subscribe", "<input-topic>")
    .load())
kafka_df.writeStream.trigger(realTime="5 minutes")
# <...streaming.readwriter.DataStreamWriter object ...>
  • Ostatnia aktualizacja o