Korzystanie z trybu czasu rzeczywistego w potokach Lakeflow

Tryb czasu rzeczywistego umożliwia przetwarzanie danych o bardzo małych opóźnieniach z opóźnieniem końcowym nawet o pięciu milisekundach. Używaj trybu czasu rzeczywistego dla obciążeń operacyjnych, które wymagają natychmiastowej reakcji na dane przesyłane strumieniowo, takie jak wykrywanie oszustw i personalizacja w czasie rzeczywistym.

Tryb czasu rzeczywistego jest również dostępny bezpośrednio w Structured Streaming, poza potokami. Zobacz Tryb czasu rzeczywistego w Strukturalnym Przetwarzaniu Strumieniowym.

Jak tryb czasu rzeczywistego osiąga małe opóźnienie

Tryb czasu rzeczywistego różni się od standardowego ciągłego przetwarzania na trzy kluczowe sposoby:

• Długotrwałe partie: system przetwarza dane, gdy staną się dostępne w źródle w długotrwałych partiach (wartość domyślna to pięć minut).
• Jednoczesne planowanie etapów: wszystkie etapy zapytań są zaplanowane w tym samym czasie. Zasób obliczeniowy musi mieć wystarczającą ilość dostępnych miejsc zadań, aby uwzględnić wszystkie etapy jednocześnie. Zobacz Ustalanie rozmiaru zasobów obliczeniowych.
• Tasowanie strumieniowe: dane są przekazywane między etapami zaraz po ich wytworzeniu, zamiast czekać na zakończenie wcześniejszego etapu przed rozpoczęciem kolejnego.

Interwał punktu kontrolnego (skonfigurowany za pośrednictwem pipelines.trigger.interval) określa, jak często są utrwalane przesunięcia stanu i źródła w magazynie trwałym. Dłuższe interwały zmniejszają obciążenie punktów kontrolnych, ale zwiększają czas odzyskiwania po wystąpieniu awarii i opóźnieniu raportowania metryk. Krótsze interwały zwiększają trwałość, ale zwiększają obciążenie.

Tryb czasu rzeczywistego i ciągłe potoki przetwarzania

Tryb czasu rzeczywistego to wyspecjalizowany typ wyzwalacza ciągłego. Tryb ciągły jest nadal wymagany; tryb czasu rzeczywistego dodaje ponadto optymalizacje opóźnień na poziomie strumienia. Aby użyć trybu czasu rzeczywistego, potok musi najpierw działać w trybie ciągłym. Tryb czasu rzeczywistego stosuje następnie dodatkowe optymalizacje na poziomie przepływu, aby osiągnąć opóźnienie poniżej jednej sekundy, wykraczające poza to, co zapewnia standardowe przetwarzanie ciągłe.

Włączenie trybu w czasie rzeczywistym wymaga trzech kroków konfiguracji:

1. Ustaw potok na tryb ciągły.
2. Włącz tryb czasu rzeczywistego na poziomie pipeline’u.
3. Zdefiniuj przepływ aktualizacji w czasie rzeczywistym.

Requirements

Konfigurowanie trybu czasu rzeczywistego

Krok 1: Ustaw potok na tryb ciągły

W ustawieniach potoku ustaw parametr Tryb potoku na Ciągły, lub ustaw go w kodzie JSON potoku:

{
  "continuous": true
}

Krok 2: Włącz tryb czasu rzeczywistego na poziomie potoku

W ustawieniach potoku dodaj następujący klucz do konfiguracji Spark w sekcji Zaawansowana > konfiguracja Spark:

spark.databricks.streaming.realTimeMode.enabled = true

Możesz też ustawić to w kodzie JSON potoku:

{
  "continuous": true,
  "spark_conf": {
    "spark.databricks.streaming.realTimeMode.enabled": "true"
  }
}

Krok 3. Definiowanie przepływu aktualizacji w czasie rzeczywistym

Tryb czasu rzeczywistego wymaga przepływu aktualizacji. Użyj dp.create_sink() lub @dp.foreach_sink do zdefiniowania celu wyjściowego, a następnie użyj dekoratora @dp.update_flow z pipelines.trigger ustawionym na "RealTime" i target wskazującym na sink.

from pyspark import pipelines as dp

# Define the output sink
dp.create_sink(
    "my_kafka_sink",
    "kafka",
    {
        "kafka.bootstrap.servers": "<bootstrap-servers>",
        "topic": "<output-topic>",
    }
)

# Define the real-time update flow targeting the sink
@dp.update_flow(
    name="my_rtm_flow",
    target="my_kafka_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",  # optional; defaults to 5 minutes
    }
)
def my_real_time_flow():
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", "<bootstrap-servers>")
            .option("subscribe", "<input-topic>")
            .load()
    )

Parametry konfiguracji na poziomie przepływu:

Przykłady kodu

Kafka do Kafka

Odczyt z tematu Kafka i zapis do wyjściowego miejsca docelowego Kafka:

from pyspark import pipelines as dp

dp.create_sink("kafka_output_sink", "kafka", {
    "kafka.bootstrap.servers": broker_address,
    "topic": output_topic,
})

@dp.update_flow(
    name="kafka_rtm_flow",
    target="kafka_output_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",
    }
)
def kafka_rtm_flow():
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", broker_address)
            .option("subscribe", input_topic)
            .option("startingOffsets", "latest")
            .load()
            .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "timestamp")
    )

Kafka do sinka ForEach

Czytaj z topiku Kafka i obsługuj każdy wiersz za pomocą funkcji Python w sinkuForEach:

from pyspark import pipelines as dp

@dp.foreach_sink(name="events_foreach_sink")
def handle_event(row):
    # Custom per-row logic here, such as a call to an external API.
    ...

@dp.update_flow(
    name="foreach_rtm_flow",
    target="events_foreach_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",
    }
)
def foreach_rtm_flow():
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", broker_address)
            .option("subscribe", input_topic)
            .option("startingOffsets", "latest")
            .load()
            .selectExpr("CAST(key AS STRING) AS key", "CAST(value AS STRING) AS value")
    )

Wzbogać za pomocą złączenia rozgłoszeniowego

Połącz strumień Kafka ze statyczną tabelą wyszukiwania. Obsługiwane są tylko sprzężenia emisji (strumieniowo-statyczne). Łączenia między strumieniami nie są obsługiwane w trybie czasu rzeczywistego.

from pyspark import pipelines as dp
from pyspark.sql.functions import broadcast, expr

dp.create_sink("enriched_output_sink", "kafka", {
    "kafka.bootstrap.servers": broker_address,
    "topic": enriched_output_topic,
})

@dp.update_flow(
    name="enriched_events_flow",
    target="enriched_output_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",
    }
)
def enriched_events():
    lookup = spark.read.table("catalog.schema.lookup_table")
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", broker_address)
            .option("subscribe", input_topic)
            .load()
            .withColumn("event_key", expr("CAST(value AS STRING)"))
            .join(broadcast(lookup), expr("event_key = lookup_key"))
            .select("event_key", "lookup_value", "timestamp")
    )

Agregacja

Zlicz zdarzenia przy użyciu klucza stanowego groupBy. Ustaw spark.sql.shuffle.partitions wartość zgodną z liczbą partycji wejściowych dla operacji stanowych:

from pyspark import pipelines as dp
from pyspark.sql.functions import col

dp.create_sink("event_counts_sink", "kafka", {
    "kafka.bootstrap.servers": broker_address,
    "topic": output_topic,
})

@dp.update_flow(
    name="event_counts_flow",
    target="event_counts_sink",
    spark_conf={
        "pipelines.trigger": "RealTime",
        "pipelines.trigger.interval": "5 minutes",
        "spark.sql.shuffle.partitions": "8",
    }
)
def event_counts():
    return (
        spark.readStream
            .format("kafka")
            .option("kafka.bootstrap.servers", broker_address)
            .option("subscribe", input_topic)
            .load()
            .selectExpr("CAST(key AS STRING) AS event_type", "timestamp")
            .groupBy(col("event_type"))
            .count()
    )

Obsługiwane źródła i odbiorniki

Obliczanie rozmiaru

Bezserwerowe potoki na kanale podglądowym

Azure Databricks rezerwuje sloty zadań dla każdego przepływu w czasie rzeczywistym, dzięki czemu możesz uruchamiać wiele przepływów w jednym pipeline bez rywalizacji przepływów o sloty zadań. Nie musisz dobierać zasobów obliczeniowych.

Adaptacyjne podziały na etapie źródłowym automatycznie dostosowują liczbę zadań odczytujących z każdego źródła Kafki lub Kinesis do natężenia ich ruchu. Aby z niego korzystać, pozostaw maxPartitions nieustawione w źródłach Kafka, a maxPartitions i shardsPerTask pozostaw nieustawione w źródłach Kinesis. Jeśli ustawisz te opcje, źródło zamiast tego używa skonfigurowanej stałej liczby zadań. Liczba podziałów tasowania nie jest regulowana, więc ustaw spark.sql.shuffle.partitions dla przepływów stanowych, jak pokazano w Agregacji.

Bezserwerowe potoki na bieżącym kanale

Usługa Azure Databricks dobiera odpowiednie zasoby obliczeniowe za Ciebie. Liczba zadań odczytujących z każdego źródła jest stała, więc ustaw maxPartitions w źródłach Kafka lub maxPartitions lub shardsPerTask w źródłach Kinesis, aby obsłużyć szczytowy ruch.

Obliczenia klasyczne

Zasób obliczeniowy musi mieć wystarczającą liczbę slotów zadań, aby uruchomić wszystkie zadania na wszystkich etapach zapytania jednocześnie. Poniższa tabela pokazuje wymagane sloty zadań dla typowych typów potoków:

Jeśli nie ustawisz maxPartitions, użyj liczby partycji w temacie w Kafka.

Obsługa operatora

transformWithState w trybie czasu rzeczywistego

transformWithState jest obsługiwany w trybie czasu rzeczywistego z następującymi różnicami względem przetwarzania w mikropartiach:

• handleInputRows jest wywoływany raz na wiersz, a nie raz dla każdego klucza w każdej partii. Iterator inputRows zwraca jedną wartość przy każdym wywołaniu.
• Czasomierze czasu zdarzeń nie są obsługiwane. Czasomierze przetwarzania są uruchamiane, gdy długotrwała partia kończy się, jeśli żadne dane nie dotarły.
• transformWithStateInPandas nie jest obsługiwana.

UDF-y Pandas w czasie rzeczywistym

Aby zminimalizować opóźnienia przy użyciu funkcji UDF biblioteki pandas, ustaw spark.sql.execution.arrow.maxRecordsPerBatch na 1. Pozwala to zoptymalizować pod kątem opóźnień kosztem przepływności. Jeśli przepływność jest również ważna, ustaw tę wartość na 100 lub większą.

Monitorowanie wydajności trybu w czasie rzeczywistym

Tryb czasu rzeczywistego wyświetla metryki opóźnień w StreamingQueryProgress pod polem latencies. Uzyskaj dostęp do tych metryk przez element StreamingQueryListener lub sprawdzając właściwość lastProgress w zapytaniu strumieniowym.

Każda metryka jest podawana w percentylach p50, p90, p95 i p99.

Ograniczenia

W przypadku klasycznych zasobów obliczeniowych oraz potoków bezserwerowych, które korzystają z bieżącego kanału, zaleca się jeden przepływ w czasie rzeczywistym na potok. Dozwolone są różne przepływy, ale konkurują one o sloty zadań, co zwiększa opóźnienia. Aby uruchomić wiele przepływów czasu rzeczywistego w jednym potoku bez wzrostu opóźnień, zobacz Potoki Serverless na kanale podglądowym.

Aby uzyskać pełną listę ograniczeń operatora i źródła, zobacz Ograniczenia trybu w czasie rzeczywistym.

Dodatkowe zasoby

• Koncepcje trybu czasu rzeczywistego
• Odniesienie do trybu w czasie rzeczywistym
• Wyzwalany a tryb potoku ciągłego
• Konfigurowanie klasycznego środowiska obliczeniowego dla potoków