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) 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 √ (z pewnymi ograniczeniami)
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.

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

  • Tryb czasu rzeczywistego wywołuje metodę handleInputRows(key: String, inputRows: Iterator[T], timerValues: TimerValues) dla każdego wiersza.
    • Iterator inputRows zwraca pojedynczą wartość. Tryb mikrosadowy wywołuje go raz dla każdego klucza, a iterator inputRows zwraca wszystkie wartości dla danego klucza w danej mikrosadowej partii.
    • Uwzględnij tę różnicę podczas pisania kodu
  • 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) ✓
Funkcja Pandas (Map) ✓

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:

  • Aby zminimalizować opóźnienie, ustaw rozmiar partii strzałki (spark.sql.execution.arrow.maxRecordsPerBatch) na 1.
    • Kompromis: ta konfiguracja optymalizuje opóźnienia kosztem przepływności. W przypadku większości obciążeń to ustawienie jest zalecane.
    • 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.
  • Funkcje Pandas UDF oraz inne funkcje nie działają dobrze z rozmiarem partii Arrow ustawionym na 1.
    • Jeśli używasz UDF-ów lub funkcji pandas, ustaw rozmiar partii Arrow na wyższą wartość (na przykład 100 lub więcej).
    • Oznacza to większe opóźnienie. Usługa Databricks zaleca używanie funkcji UDF strzałki, jeśli jest 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.
  • Ostatnia aktualizacja o