Potwierdzenia zwrotne

Funkcja zwrotna potwierdzenia umożliwia klientowi asynchroniczne reagowanie na potwierdzenia rekordów i błędy, bez blokowania pętli producenta. Gdy rekordy stają się trwałe lub zawodzą, Zerobus Ingest uruchamia twoje callback w tle, dzięki czemu możesz śledzić postępy i aktualizować metryki bez spowolniania producenta oraz dowiedzieć się o awariach od razu, gdy się pojawią.

To różni się od oczekiwania na offset lub flushing: to są blokujące wywołania, gdzie kod czeka na trwałość w linii. Wywołanie zwrotne nie jest wywołaniem blokującym. To handler, którego SDK wywołuje za Ciebie, gdy pojawiają się potwierdzenia.

Obsługiwane są wywołania zwrotne potwierdzeń dla strumieni SDK JSON i Protocol Buffers (protobuf). Strumienie Arrow Flight nie obsługują callbacków; aby potwierdzić trwałość danych w strumieniu Arrow, użyj wait_for_offset() lub flush(). Zobacz Korzystanie z Arrow Flight z Zerobus Ingest.

Poniższe nazwy metod i typów pochodzą z Python SDK. Pozostałe zestawy SDK Zerobus udostępniają mechanizmy wywołań zwrotnych potwierdzeń tam, gdzie są one obsługiwane, z użyciem równoważnych konstrukcji w każdym języku.

Jak działają callbacki

Funkcję wywołania zwrotnego definiujesz, tworząc podklasę klasy AckCallback i implementując dwie metody:

  • on_ack(offset: int): wywoływane, gdy zgłoszenie (rekord lub partia) zostaje pomyślnie uznane przez serwera za trwałe. offset identyfikuje potwierdzone zgłoszenie.
  • on_error(offset: int, error_message: str): wywoływane, gdy zgłoszenie napotyka błąd. Element on_error jest opcjonalny. Zaimplementuj to, aby radzić sobie z awariami lub rejestrować niepowodzenia.

Funkcja zwrotna jest wywoływana raz dla każdego przesłanego rekordu lub wsadu, gdy jego logiczny offset zostanie potwierdzony lub gdy potwierdzenie się nie powiedzie, dzięki czemu stanowi bieżący wskaźnik postępu przyjmowania danych w całym strumieniu.

Twoje metody callback działają na wątkach tła SDK, więc ich wywołanie nie blokuje producenta. Niech będą szybkie i nieblokujące. Co zrobić w przypadku awarii to odpowiedzialność klienta: zarejestrować, powiadomić, spróbować ponownie lub zatrzymać. Niektóre błędy są krytyczne, a jeśli on_error zgłosi, że strumień uległ trwałej awarii, musisz wznowić działanie przy użyciu nowego strumienia. Zobacz wzorce odzyskiwania i ponawiania prób.

Skonfiguruj wywołanie zwrotne

Dołączasz funkcję wywołania zwrotnego do strumienia, przekazując instancję swojej podklasy AckCallback jako opcję ack_callback w StreamConfigurationOptions podczas tworzenia strumienia. Funkcja wywołania zwrotnego ma wtedy zastosowanie do każdego rekordu przetworzonego w ramach tego strumienia.

from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import AckCallback, StreamConfigurationOptions, TableProperties

class MyAckCallback(AckCallback):
    def on_ack(self, offset: int) -> None:
        print(f"Record acknowledged at offset: {offset}")

    def on_error(self, offset: int, error_message: str) -> None:
        print(f"Error at offset {offset}: {error_message}")

sdk = ZerobusSdk(SERVER_ENDPOINT, DATABRICKS_WORKSPACE_URL)

table_properties = TableProperties("main.default.air_quality")
options = StreamConfigurationOptions(
    ack_callback=MyAckCallback(),
)
stream = sdk.create_stream(CLIENT_ID, CLIENT_SECRET, table_properties, options)

try:
    for row in records:
        stream.ingest_record_offset(row)
finally:
    stream.close()

Gdy oddzwonienie jest zarejestrowane, nie czekasz w kolejce. on_ack jest wywoływane po potwierdzeniu trwałego zapisu każdego rekordu, a on_error jest wywoływane, jeśli zapis rekordu się nie powiedzie.

Wywołania zwrotne a blokowanie

Funkcje wywołania zwrotnego i wywołania blokujące rozwiązują różne problemy, i można ich używać razem:

  • Użyj funkcji wywołania zwrotnego potwierdzenia, aby asynchronicznie reagować na potwierdzenia trwałości i błędy na bieżąco, przy jednoczesnym zachowaniu wysokiej przepustowości. Dobre do śledzenia postępów, metryk i rejestrowania błędów.
  • Użyj wait_for_offset() lub flush(), gdy kod musi czekać, aż konkretny rekord lub wszystkie rekordy oczekujące na zapis zostaną trwale zapisane, zanim przejdziesz dalej.
  • Ostatnia aktualizacja o