Blokowanie wiadomości i potwierdzenie

SDK Zerobus Ingest oferują kilka metod pobierania danych, które równoważą przepustowość z tym, ile potwierdzenia trwałości otrzymujesz. Ta strona wyjaśnia poszczególne metody oraz wskazuje, kiedy należy czekać na potwierdzenie trwałości. Aby reagować na potwierdzenia asynchronicznie zamiast blokować, zobacz sekcję Wywołania zwrotne potwierdzeń.

Przykłady na tej stronie korzystają z Python SDK. Dokładne limity czasu i opcje konfiguracyjne akceptowane przez każdą metodę (w tym ich domyślne i jednostki) można znaleźć w repozytorium Zerobus SDK. Inne SDK językowe udostępniają równoważne opcje.

Czym jest offset?

Każdemu rekordowi, który pozyskujesz, przypisywane jest przesunięcie: jego pozycja w strumieniu. Przesunięcie to sposób, w jaki odnosisz się do konkretnego zapisu, gdy chcesz potwierdzić, że został on trwało zapisany. Zerobus Ingest zapewnia gwarancję dostawy przynajmniej raz, a oczekiwanie na offset to sposób, w jaki klient potwierdza tę gwarancję dla danego rekordu.

Potwierdzenie przesunięcia oznacza, że rekord został trwale zapisany, a nie że można go już odpytywać w tabeli Delta. Zerobus Ingest zapisuje trwałe rekordy w tabeli niedługo później. Wartości opóźnień znajdują się w sekcji Opóźnienia.

Metody pozyskiwania

Zestawy SDK udostępniają dwa sposoby importowania rekordu. (Poniższe nazwy metod pochodzą z Python SDK. Inne SDK udostępniają równoważne metody.)

Metoda Returns Użyj go, gdy
Oparte na przesunięciach, ingest_record_offset() Przesunięcie rekordu po umieszczeniu rekordu w kolejce strumienia. Zalecana domyślna opcja. Chcesz wstawiać rekordy w kolejce i opcjonalnie potwierdzić trwałość później, czekając na przesunięcie.
Oparte na przyszłości, ingest_record() Element RecordAcknowledgment, na który możesz czekać. Deprecated. Wolę offset dla lepszej wydajności.

Na podstawie offsetu (zalecane)

ingest_record_offset() wysyła rekord i zwraca jego offset, gdy rekord zostanie umieszczony w kolejce w strumieniu. Wywołanie odbywa się w wątku wywołującym, więc rekordy są umieszczane w kolejce w kolejności, w jakiej wywołujesz tę metodę, a zwrócony offset pozwala później potwierdzić trwałość danych za pomocą wait_for_offset(). To zalecany domyślny wybór dla większości producentów i jest to metoda stosowana w przykładach Use Zerobus Ingest .

Oparty na przyszłości (wycofany)

ingest_record() zwraca obiekt RecordAcknowledgment, na który można oczekiwać, aby zapewnić trwałość. Jest uznawana za przestarzałą na rzecz metody opartej na przesunięciach, która zapewnia lepszą wydajność. Używaj go tylko do istniejącego kodu, który jeszcze nie został przeniesiony.

Rekord po rekordzie a przetwarzanie wsadowe

Każda metoda pozyskiwania danych ma wariant wsadowy (na przykład ingest_records_offset()), który przesyła listę rekordów w jednym wywołaniu. Porcjonowanie jest bardziej efektywne niż indywidualne wydania hurtowe.

W przypadku JSON i buforów protokołu (protobuf) partia zatwierdza się atomowo: albo każdy rekord w partii jest akceptowany i uczyniony trwałym, albo cała partia zostaje odrzucona. Zerobus Ingest nie wykonuje częściowych przesyłań ani częściowego potwierdzania odbioru w przypadku tych formatów, więc tabela nigdy nie zawiera niepełnej partii. Partia, która nie przechodzi walidacji (na przykład z powodu niedopasowania schematu), zostaje natychmiast odrzucona, zanim trafi do tabeli, zamiast zapisać część rekordów, a resztę odrzucić.

Ponieważ partia JSON lub protobuf jest wysyłana jako pojedyncza wiadomość, maksymalny rozmiar wiadomości 10 MB dotyczy zarówno pojedynczego rekordu, jak i całej partii: wszystkie rekordy w partii razem muszą mieścić się w granicach 10 MB. Dobierz wielkość partii tak, aby nie przekroczyć tego limitu. Zobacz Rozmiar płyty.

Partie Arrow Flight stanowią wyjątek

Pozyskiwanie danych za pomocą Apache Arrow Flight nie jest zgodne z opisanym powyżej modelem pojedynczego komunikatu typu „wszystko albo nic”. Partia Arrow może być znacznie większa niż partia JSON lub protobuf, a ścieżka lotu Arrow dzieli dużą partię na mniejsze wiadomości transportowe, które są wysyłane i potwierdzane indywidualnie, a nie jako jedna jednostka atomowa. W efekcie:

  • Limit 10 MB na każdą wiadomość, który obowiązuje w przypadku partii JSON i protobuf, nie dotyczy partii Arrow w taki sam sposób. Duża partia Arrow zostaje podzielona na wiadomości transportowe zamiast zostać odrzucona ze względu na rozmiar.
  • Trwałość jest potwierdzana na poziomie pojedynczego komunikatu transportowego, więc bardzo duża partia logiczna może zostać trwale zapisana tylko częściowo, jeśli awaria wystąpi w trakcie przetwarzania, zamiast być zatwierdzana w całości albo wcale.

ingest_batch() nadal zwraca pojedynczy logiczny offset dla przesłanej partii, a wait_for_offset() dla tego offsetu kończy się dopiero po potwierdzeniu każdego komunikatu transportowego składającego się na tę partię. Aby uzyskać informacje o pełnym modelu Arrow Flight, wskazówki dotyczące przetwarzania wsadowego oraz odzyskiwania niepotwierdzonych danych, zobacz Korzystanie z Arrow Flight z Zerobus Ingest.

Kiedy powinieneś zablokować wiadomość?

Blokowanie na offsetzie przenosi przepustowość na silniejszą gwarancję trwałości na rekord w kodzie klienta. Wybierz na podstawie swojego obciążenia:

  • Nie blokuj: to właściwe ustawienie domyślne w przypadku wysokowolumenowego przesyłania strumieniowego, gdy zależy Ci na utrzymaniu stałej przepustowości i możesz zbiorczo potwierdzać trwałość (na przykład przy zamknięciu strumienia lub za pomocą funkcji zwrotnej potwierdzenia). Większość producentów powinna zacząć właśnie tutaj.
  • Blokowanie na offsecie: należy to rozważyć, gdy aplikacja musi wiedzieć, że dany rekord został trwale zapisany, zanim podejmie kolejne działanie. Przykład:
    • Zaraz usuniesz lub potwierdzisz odbiór źródła danych (komunikatu kolejki, pliku, kursora upstream) i nie wolno go utracić, jeśli ingestia się nie powiedzie.
    • Przetwarzasz dane w punktach kontrolnych lub na granicach transakcji i musisz zapewnić trwałe zapisanie każdego punktu kontrolnego przed przejściem dalej.
    • Wykonujesz niskowolumenowe zapisy o wysokiej wartości, gdzie potwierdzenie każdego rekordu jest ważniejsze niż przepustowość.

Nie blokuj przetwarzania każdego rekordu w pętli o wysokiej przepustowości. To serializuje producenta na trasie w obie strony do serwera dla każdego rekordu i znacznie zmniejsza przepustowość. Zamiast tego usługa Azure Databricks zaleca pozyskanie dużej partii rekordów, a następnie jednorazowe potwierdzenie trwałości dla całej partii. Masz dwa sposoby: poczekaj na najnowszy offset albo przepłukaj strumień. Blokowanie pojedynczego rekordu powinno być zarezerwowane dla konkretnych przypadków powyżej, gdzie pojedynczy rekord musi zostać potwierdzony przed kolejnym działaniem.

Poczekaj na offset

wait_for_offset() blokuje działanie, dopóki Zerobus Ingest nie potwierdzi, że rekord pod tym przesunięciem został trwale zapisany, lub do upłynięcia limitu czasu. Użyj go, aby potwierdzić konkretny punkt w strumieniu, najczęściej ostatni rekord fragmentu. Pobieraj fragment, zachowaj ostatni offset, który zwraca pętla, i czekaj na ten jeden offset zamiast czekać po każdym rekordzie:

from zerobus.sdk.sync import ZerobusSdk
from zerobus.sdk.shared import TableProperties

sdk = ZerobusSdk(SERVER_ENDPOINT, DATABRICKS_WORKSPACE_URL)

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

try:
    last_offset = 0
    for row in records:
        last_offset = stream.ingest_record_offset(row)

    # Block until everything up to the last record of the chunk is durable
    stream.wait_for_offset(last_offset)
    print("Chunk durably written.")
finally:
    stream.close()

Opróżnij strumień

flush() blokuje działanie do momentu, aż wszystkie rekordy pozyskane do tej pory zostaną trwale zapisane, po czym zwraca sterowanie. W przeciwieństwie do , nie śledzisz offsetu: flush czeka na wszystko, co jest wait_for_offset()oczekujące na streamie. Nie zamyka strumienia, więc możesz dalej pobierać po nim.

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

    # Block until every pending record is durable
    stream.flush()
    print("All ingested records durably written.")
finally:
    stream.close()

wait_for_offset vs. spłukiwanie

Oba potwierdzają wytrzymałość dla kawałka gry. Wybierz w zależności od tego, co potwierdzasz:

  • Używaj wait_for_offset(offset), gdy chcesz potwierdzić przetwarzanie do określonego rekordu, na przykład do granicy punktu kontrolnego, podczas gdy inne rekordy mogą nadal być w przetwarzaniu za nim.
  • Używaj, flush() gdy chcesz potwierdzić, że wszystkie oczekujące rekordy są trwałe przed przejściem dalej, na przykład na końcu partii, przed przesunięciem kursora w górę lub przed wyłączeniem. flush() zależy od konfigurowalnego limitu czasu opróżniania.

close() opróżnia bufor i zamyka strumień, więc rekordy są zawsze trwale zapisywane podczas kontrolowanego zamykania. Zawsze nazywaj to blokiem finally .

Reaguj na potwierdzenia asynchronicznie

Jeśli zamiast blokować chcesz reagować na potwierdzenia wytrzymałości i błędy, gdy się pojawią, podczas gdy producent pracuje na pełnych obrotach, zarejestruj potwierdzenie na streamie. Wywołania zwrotne są odrębnym mechanizmem od wywołań blokujących opisanych na tej stronie. Zobacz potwierdzenia zwrotne.

  • Ostatnia aktualizacja o