Dokumentacja trybu w czasie rzeczywistym

Ta strona opisuje wsparcie dla zapytań w trybie czasu rzeczywistym, które bezpośrednio korzystają z API Structured Streaming. W przypadku zarządzanych potoków zobacz tryb czasu rzeczywistego w potokach Lakeflow, który ma oddzielne wymagania w czasie rzeczywistym, źródłowe, pochłaniacze i operatora.

Obsługiwane języki

Tryb czasu rzeczywistego obsługuje język Scala, Java i Python.

Typy obliczeniowe

Tryb czasu rzeczywistego obsługuje następujące typy obliczeń:

Typ środowiska obliczeniowego	Wsparte
Dedykowany (dawniej: dla pojedynczego użytkownika)	✓
Standardowa (dawniej: udostępnione)	√ (tylko Python)
Potoki lakeflow w wersji klasycznej	Obsługiwane przez konfigurację potoku
Potoki lakeflow bezserwerowe	Obsługiwane przez konfigurację potoku
Notesy i zadania bezserwerowe	Nie obsługiwane dla bezpośrednich zapytań w czasie rzeczywistym w Structured Streaming

W przypadku obciążeń wrażliwych na opóźnienia z funkcjami zdefiniowanymi przez użytkownika usługa Databricks zaleca korzystanie z trybu dedykowanego dostępu. Zobacz Funkcje tabeli.

Tryby wykonywania

Tryb czasu rzeczywistego obsługuje tylko tryb aktualizacji:

Tryb wykonywania	Wsparte
Tryb aktualizacji	✓
tryb dołączania	Niewspierane
Tryb pełny	Niewspierane

Źródła i ujścia

Tryb czasu rzeczywistego obsługuje następujące źródła i ujścia:

Źródło lub ujście	Jako źródło	Jako ujście
Apache Kafka	✓	✓
Event Hubs (przy użyciu łącznika Kafka)	✓	✓
Kinesis	✓ (zalecany tryb EFO; ankiety w Databricks Runtime 19 i wyższym)	Niewspierane
AWS MSK	✓	Niewspierane
Delta	Niewspierane	Niewspierane
Google Pub/Sub (usługa przesyłania wiadomości)	Niewspierane	Niewspierane
Apache Pulsar	Niewspierane	Niewspierane
Dowolne ujścia (przy użyciu forEachWriter)	Nie dotyczy	✓

Operatorów

Tryb czasu rzeczywistego obsługuje większość operatorów przesyłania strumieniowego ze strukturą:

Operacje bezstanowe

Obsługujący	Wsparte
Selekcja	✓
Projekcja	✓
mapPartitions	Nieobsługiwane (zobacz ograniczenie)
Union	√ (z pewnymi ograniczeniami)

UDFs

Obsługujący	Wsparte
Scala UDF	✓ (skalarne UDF). Funkcje grupowane, takie jak flatMapGroups i mapGroupsWithState, nie są obsługiwane.
Python funkcji zdefiniowanej przez użytkownika	√ (z pewnymi ograniczeniami)

Aggregation

Function	Wsparte
sum	✓
liczba	✓
max	✓
min	✓
avg	✓
Funkcje agregacji	✓

Windowing

Obsługujący	Wsparte
Tumbling	✓
Sliding	✓
Session	Niewspierane

Deduplication

Obsługujący	Wsparte
usuńDuplikaty	✓
UsuńDuplikatyWZnakuWodnym	✓

Przesyłanie strumieniowe do sprzężenia tabeli

Obsługujący	Wsparte
Sprzężenie wewnętrzne	✓
Sprzężenia zewnętrzne	✓
Rozgłaszaj sprzężenia tabeli (rozmiar tabeli o rozmiarze co najmniej 10 mb)	✓
Sprzężenia tabeli (bez emisji)	Niewspierane

Przesyłanie strumieniowe do sprzężenia strumieniowego

Obsługujący	Wsparte
Sprzężenie wewnętrzne	√ (Środowisko Databricks Runtime 18 LTS i nowsze, z niektórymi konfiguracjami)
Sprzężenia zewnętrzne	Niewspierane

Uwaga / Notatka

Aby używać strumienia do przesyłania strumieniowego sprzężeń w trybie czasu rzeczywistego, należy ustawić dodatkowe konfiguracje platformy Spark. Aby uzyskać więcej informacji na temat konfiguracji i wymagań dotyczących uruchamiania wielu strumieni, zobacz Stream to stream to stream joins (Przesyłanie strumieniowe do sprzężeń strumieniowych).

Dowolny operator stanowy

Obsługujący	Wsparte
(płaskie)MapGroupsWithState	Niewspierane
transformWithState	√ (z pewnymi różnicami)

Ujścia zdefiniowane przez użytkownika

Sink	Wsparte
forEach	✓
forEachBatch	Niewspierane

Uwagi specjalne

Niektóre operatory i funkcje mają konkretne zagadnienia lub różnice w przypadku użycia w trybie czasu rzeczywistego.

transformWithState w trybie czasu rzeczywistego

W przypadku tworzenia niestandardowych aplikacji stanowych usługa Databricks obsługuje API transformWithState w strukturze strumieniowej Apache Spark. Zobacz Zbudowaj niestandardową aplikację stanową, transformWithState aby uzyskać więcej informacji o API i fragmentach kodu.

transformWithState obsługuje zarówno procesory synchroniczne, jak i asynchroniczne w trybie czasu rzeczywistego. Wymagania dotyczące przetwarzania asynchronicznego zobacz Przetwarzanie asynchroniczne z .transformWithState

Jednak interfejs API działa inaczej w trybie czasu rzeczywistego niż w zapytaniach mikrosadowych.

• Dla procesorów synchronicznych tryb czasu rzeczywistego wywołuje metodę handleInputRows(key: String, inputRows: Iterator[T], timerValues: TimerValues) dla każdego wiersza.
  • Iterator inputRows zwraca pojedynczą wartość. Tryb mikro-batch wywołuje go raz dla każdego klucza, a inputRows iterator zwraca wszystkie wartości dla klucza w mikropartii.
  • Uwzględnij tę różnicę podczas pisania kodu.
• W przypadku procesorów asynchronicznych tryb czasu rzeczywistego grupuje wiersze, które przychodzą razem, i wywołuje handleInputRows raz dla każdego klucza w grupie, z wszystkimi wierszami tego klucza. Ponieważ tryb czasu rzeczywistego uruchamia długotrwałe partie, ten sam klucz może odebrać wiele wywołań w ramach jednej partii.
• Czasomierze czasu zdarzeń nie są obsługiwane w trybie czasu rzeczywistego.
• transformWithStateInPandas nie jest obsługiwany w trybie czasu rzeczywistego. Zamiast tego użyj interfejsu API opartego na transformWithState wierszach, który używa Row obiektów, a nie ramek danych biblioteki pandas.
• W trybie czasu rzeczywistego zegary są opóźnione w uruchamianiu w zależności od przybycia danych.
  • Jeśli czasomierz jest zaplanowany na 10:00:00, ale żadne dane nie docierają, czasomierz nie jest uruchamiany natychmiast.
  • Jeśli dane pojawią się o godzinie 10:00:10, czasomierz zostanie wyzwolony z 10-sekundowym opóźnieniem.
  • Jeśli żadne dane nie docierają, a długotrwała partia kończy działanie, czasomierz jest uruchamiany przed zakończeniem partii.

Uwaga / Notatka

W środowisku Databricks Runtime 18.1 lub nowszym, jeśli używasz transformWithState i trybu czasu rzeczywistego dla Python z niską przepływnością, mniej niż 5 rekordów na sekundę, może być widoczne zwiększone opóźnienie do kilkuset milisekund. Usługa Databricks zaleca uaktualnienie do środowiska Databricks Runtime 18.2 lub nowszego w celu rozwiązania problemu.

Python UDF w trybie czasu rzeczywistego

Usługa Databricks obsługuje większość funkcji zdefiniowanych przez użytkownika (UDF) Python w trybie czasu rzeczywistego:

Bezstanowa

Typ UDF	Wsparte
Python funkcji zdefiniowanej przez użytkownika (UDF) (Python funkcji zdefiniowanych przez użytkownika (UDF))	✓
Funkcja skalarna Arrow zdefiniowana przez użytkownika (UDF)	✓
Funkcja UDF skalarna biblioteki Pandas (funkcje zdefiniowane przez użytkownika biblioteki Pandas)	✓
Funkcja strzałkowa (mapInArrow)	Nieobsługiwane (zobacz ograniczenie)
Funkcja Pandas (Map)	Nieobsługiwane (zobacz ograniczenie)

Grupowanie stanowe (UDAF)

Typ UDF	Wsparte
transformWithState (tylko Row interfejs)	✓
transformWithStateInPandas	Niewspierane. Zamiast tego użyj interfejsu API opartego na transformWithState wierszach, który używa Row obiektów, a nie ramek danych biblioteki pandas. Aby uzyskać szczegółowe informacje, zobacz transformWithStateInPandas nieobsługiwane .
applyInPandasWithState	Niewspierane

Grupowanie niestanowe (UDAF)

Typ UDF	Wsparte
apply	Niewspierane
applyInArrow	Niewspierane
applyInPandas	Niewspierane

Funkcje tabeli

Typ UDF	Wsparte
UDTF (Python funkcje tabeli zdefiniowane przez użytkownika (UDTFs))	Niewspierane
UC UDF	Niewspierane

Istnieje kilka kwestii, które należy wziąć pod uwagę podczas korzystania z funkcji zdefiniowanych przez użytkownika Python w trybie czasu rzeczywistego:

• W Databricks Runtime 19 i wyższych tryb czasu rzeczywistego powala wiersze wejściowe według czasu dla wszystkich UDF-ów obsługiwanych przez tryb czasu rzeczywistego. Każda partia zbiera wiersze, które dotarły w krótkim czasie (10 milisekund), czyli do osiągnięcia maksymalnego rozmiaru partii, a następnie wysyła je do UDF. To równoważy opóźnienia i przepustowość, więc nie musisz ustawiać rozmiaru partii Arrow (spark.sql.execution.arrow.maxRecordsPerBatch) na 1.
• W Databricks Runtime 18 LTS i niższych rzędy są grupowane wyłącznie według liczby danych:
  • Aby zminimalizować opóźnienie, ustaw rozmiar partii strzałki (spark.sql.execution.arrow.maxRecordsPerBatch) na 1. Ta konfiguracja optymalizuje opóźnienia kosztem przepustowości. Zwiększ rozmiar partii tylko wtedy, gdy wymagana jest większa przepływność w celu uwzględnienia woluminu wejściowego, akceptując potencjalny wzrost opóźnienia.
  • UDF Pandas nie radzą sobie dobrze przy wielkości partii Arrow wynoszącej 1. Jeśli używasz UDF Pandas, ustaw rozmiar partii strzałek na wyższą wartość (np. 100 lub więcej). Oznacza to większe opóźnienie. Databricks zaleca użycie UDF Arrow, jeśli to możliwe.
• transformWithStateInPandas nie jest obsługiwany w trybie czasu rzeczywistego. Zamiast tego użyj interfejsu API opartego na transformWithState wierszach, który używa Row obiektów, a nie ramek danych biblioteki pandas. Zobacz transformWithStateInPandas nieobsługiwane i Rejezys przykład pracy Python przy użyciu interfejsu API opartego na wierszach.
• W przypadku obciążeń wrażliwych na opóźnienia z funkcjami zdefiniowanymi przez użytkownika usługa Databricks zaleca korzystanie z trybu dedykowanego dostępu. W standardowym trybie dostępu obciążenie związane z izolacją zabezpieczeń może spowolnić wydajność funkcji zdefiniowanej przez użytkownika.