Realtimemodus in gestructureerd streamen

Important

Real-time modus is alleen beschikbaar op Microsoft Fabric Spark Runtime 2.0 (Spark 4.1) of later. Deze nieuwe streamingmodus is niet beschikbaar op eerdere Fabric-runtimes.

Apache Spark 4.1 introduceert de Trigger.RealTime API voor real-time modus. Microsoft Fabric ondersteunt deze ultra-lage-latentie Structured Streaming uitvoeringsmodus in Fabric Runtime 2.0. Het gebruikt langlopende taken die continu uit streamingbronnen lezen, ondersteunde transformaties toepassen en schrijven naar ondersteunde sinks. Dit model vermindert planningsvertraging omdat Spark niet wacht op een discrete microbatch om elke eenheid werk te starten.

Gebruik de real-time-modus wanneer uw workload recente gebeurtenissen nodig heeft, zodat deze zo snel mogelijk door de query worden verwerkt. Gebruik standaard micro-batchmodus wanneer je prioriteit geeft aan doorvoer, brede connectorondersteuning of batchgerichte bewerkingen.

Real-time modus versus microbatch-modus

Gestructureerde Streaming gebruikt standaard microbatch-uitvoering. In micro-batchmodus plant Spark een reeks kleine batchtaken, plant deze in en voltooit ze. Een triggerinterval bepaalt hoe vaak Spark controleert op nieuwe data en de volgende batch start. Deze aanpak werkt goed voor de meeste lakehouse-data-inname, tabelonderhoud, aggregaties en op bestanden gebaseerde doelen.

Real-time modus verandert het uitvoeringspatroon. Spark start langlopende taken die actief blijven terwijl de query loopt. De taken verwerken records doorlopend in plaats van te wachten op de grens van de volgende microbatch. Dit patroon kan de end-to-end latentie voor ondersteunde queries verminderen, vooral wanneer gebeurtenissen met minimale vertraging tussen streamingsystemen moeten worden verplaatst.

Het nadeel is dat de realtime-modus een smallere ondersteuningsoppervlak heeft tijdens de preview. Je moet je bron, sink, outputmodus en transformaties valideren op Fabric Runtime 2.0 of later voordat je het gebruikt voor productieworkloads.

Hoe de realtime-modus data verwerkt

Real-time modus vervangt de korte microbatch-lus door een enkele langlopende batch die actief blijft gedurende het hele triggerinterval en records verwerkt zodra ze binnenkomen. Twee ontwerpkeuzes zorgen voor een lage latentie. Ten eerste start Spark elke fase van de query tegelijk en verbindt ze met een streaming shuffle, zodat een record van de bron via elke transformatie naar de sink kan gaan zonder te pauzeren bij een fasegrens. Ten tweede hoeft Spark, omdat de taken actief blijven, niet bij elke trigger de plannings- en inplanningskosten per batch te betalen.

Spark heeft nog steeds een punt nodig om vooruitgang te behouden. Het legt bronoffsets vast, werkt de statusopslag bij en publiceert streamingstatistieken op het grensvlak tussen de ene langlopende batch en de volgende. Die timing is waarom het triggerinterval zich anders gedraagt dan een microbatch-interval: het bepaalt hoe lang elke batch draait voordat Spark-checkpoints plaatsvinden, niet hoe lang Spark wacht voordat het begint met verwerken.

Stem het interval af met dat gedrag in gedachten. Bij een langer interval worden er minder vaak checkpoints gemaakt en meetwaarden gerapporteerd, waardoor bij een storing meer data opnieuw moet worden verwerkt en de monitoringweergave langzamer wordt ververst. Een korter interval zorgt voor frequentere checkpoints, waardoor herstel sneller verloopt en statistieken actueler zijn, maar het extra werk voor checkpoints kan ten koste gaan van de latentie. Er is geen universele beste waarde, dus vergelijk een paar intervallen met je eigen werkdruk. De voorbeelden in dit artikel beginnen met een interval van vijf minuten.

Prerequisites

Voordat je Real-time Mode inschakelt, vul je deze vereisten uit:

  • Gebruik Microsoft Fabric Spark Runtime 2.0 (Spark 4.1) of nieuwer. Real-time Mode is niet beschikbaar in eerdere runtimes.
  • Voer de werklast uit in een Fabric Spark-ervaring die Runtime 2.0 ondersteunt, zoals een notebook of Spark-taakdefinitie.
  • Gebruik een streamingbron en -sink die Real-time Mode ondersteunt in de geselecteerde runtime.
  • Configureer een duurzame checkpointlocatie voor de query.
  • Bied voldoende taakslots voor de query. Omdat Real-time Mode alle fasen tegelijk plant, heeft de pool minstens evenveel beschikbare taakslots nodig als het totale aantal taken in elke fase van de query. Plan één Real-time Mode-query per pool, tenzij je bevestigt dat de pool vrije slots heeft voor meer.

Voor de setupstappen van Runtime 2.0, zie Runtime 2.0 in Fabric.

Schakel Real-time modus in

Schakel Real-time Mode in door de streamingtrigger aan Trigger.RealTime("<interval>") te writeStreamzetten. Het interval is de langlopende batchduur die de frequentie van checkpoint en metriek bepaalt, zoals beschreven in Hoe Real-time Mode data verwerkt.

Important

Real-time modus ondersteunt alleen de update outputmodus. De uitvoermodi append en complete worden niet ondersteund. Stel .outputMode("update") in voor de zoekopdracht.

Note

De realtime trigger is beschikbaar in de JVM streaming API. Tijdens de preview stelt PySpark dit niet standaard beschikbaar via DataStreamWriter.trigger().

De voorbeelden gaan ervan uit dat input_options, output_options en c_path uw bronopties, doelopties en het checkpointpad bevatten. Voer deze voorbeelden alleen uit op Fabric Spark Runtime 2.0 (Spark 4.1) of later.

Maak eerst de streaming DataFrame aan in Python.

passthrough = (
    spark.readStream
    .format("kafka")
    .options(**input_options)
    .load()
    .selectExpr("CAST(key AS STRING) AS key", "CAST(value AS STRING) AS value")
)

Pas dan de tijdelijke JVM-brugoplossing toe. PySpark verschijnt Trigger.RealTime nog niet in DataStreamWriter.trigger(), dus deze workaround creëert de JVM-trigger via py4j en past deze toe op de onderliggende JavaDataStreamWriter. Vervang deze code door de ingebouwde PySpark-trigger-API zodra ondersteuning beschikbaar is.

# Create RTM trigger object via JVM bridge
rtm_trigger = (
    spark._jvm.org.apache.spark.sql.streaming
        .Trigger.RealTime("5 minutes")
)

# Define streaming query
query = (
    passthrough.writeStream
    .format("kafka")
    .options(**output_options)
    .option("checkpointLocation", c_path)
    .outputMode("update")
)

# Apply RTM trigger via Java DataStreamWriter
query._jwrite = query._jwrite.trigger(rtm_trigger)

# Start streaming query
query = query.start()

query.awaitTermination()

Ontwerpqueries voor Real-time Mode

Realtime-modus valideert de bron, het doel, de uitvoermodus en het queryplan wanneer de query wordt gestart. Tijdens de preview ondersteunt het een gerichte set streamingpatronen. Bevestig de exacte ondersteuning voor jouw runtime-versie voordat je een query naar productie verplaatst, want de preview-dekking verandert in de loop van de tijd.

Het beste mentale model is een pad met lage latentie dat gebeurtenissen tussen streamingberichtsystemen verplaatst en onderweg lichte verwerking toepast. Ontwerp je query rond dat model:

  • Lees en schrijf streaming-messagingsystemen. Real-time Mode is gebouwd voor Kafka-compatibele eindpunten, waaronder Apache Kafka en Azure Event Hubs via de Kafka-connector. Om resultaten direct naar een extern systeem te sturen, roep je .foreach(...) aan met een ForeachWriter.
  • Bewaar Lakehouse en dien I/O in op een standaard trigger. Delta-tabellen, lakehouse-tabellen en bestandsgebaseerde bronnen en sinks zijn geen Real-time Mode-doelen. Schrijf data weg naar een lakehouse, beheer tabellen of schrijf in plaats daarvan bestanden met een microbatch-trigger, en reserveer Real-time Mode voor de latentiegevoelige stap tussen berichtsystemen.
  • Geef de voorkeur aan lichte transformaties. Projecties, casts, filters, kolomexpressies en eenvoudige verrijking geven de laagste latentie. Hoe meer status en schudden een query toevoegt, hoe meer je deze eerst in Real-time Mode moet testen.

Wanneer je het Event Hubs Kafka-compatibele eindpunt gebruikt, lijn dan de idle- en metadata-intervallen van de Kafka-client af met de idle-timeout van de Event Hubs. Een mismatch in time-outs kan rond de grens van een langlopende batch vertraging veroorzaken bij het opnieuw verbinden. Raadpleeg Kafka-compatibele verbindingen gezond houden voor de aanbevolen bron- en doelopties.

Real-time Mode ondersteunt veel operators met status, waaronder vensteraggregaties, deduplicatie en join-bewerkingen, maar de ondersteunde patronen zijn beperkter dan in de micro-batch-modus. Gebruik de volgende matrices als snelle referentie bij het ontwerpen van een query en bevestig het huidige gedrag op je runtime-versie.

De connectormatrix vat samen waar Real-time Mode in de pijplijn past:

Connector Als bron Als een gootsteen
Kafka-compatibele eindpunten (Apache Kafka, Azure Event Hubs via de Kafka-connector) Ondersteund Ondersteund
Aangepaste sink via .foreach(...) en ForeachWriter Niet van toepassing Ondersteund
Delta- en lakehouse-tabellen Niet ondersteund Niet ondersteund
Bestandsgebaseerde formaten Niet ondersteund Niet ondersteund

De operatormatrix groepeert transformaties op basis van hoe goed ze passen bij het laag-latentiemodel:

Transformation Realtime-modus geschikt Wat moet u doen?
Stateless bewerkingen: selecteren, filteren, typeconversie, projecteren, scalaire UDF's Ondersteund Voorkeurspad; geeft de laagste latentie
Aggregaties en tuimelende of schuivende tijdsvensters Ondersteund Voeg een watermerk toe aan de gebonden toestand
Deduplicatie met of binnen een watermerk Ondersteund Kies sleutels die een logische gebeurtenis identificeren
Sessievensters (op tussenpozen gebaseerde) Niet ondersteund Voer de query uit op een microbatch-trigger
stream-naar-referentiejoin Ondersteund wanneer de referentiezijde een broadcast verzendt Houd de referentiedataset klein
Samenvoeging van stroom met stroom Alleen inner join, met extra configuratie Vermijd buitenste samenvoegingen tussen twee stromen
Aangepaste status met transformWithState Ondersteund met verschillende semantiek Zie de volgende noot
Operatoren voor één partitie tegelijk: mapPartitions, mapInPandas, mapInArrow Niet ondersteund Herschrijf met UDF's op rijniveau, filters of expressies voor complexe typen
Unie met zichzelf, unie met een batchbron, of unie na een operator met status Niet ondersteund Gebruik onafhankelijke streaming-inputs en pas union toe vóór stateful work

Ondersteund betekent niet dat een operatie de laagst mogelijke latentie behoudt. Aggregaties, join-bewerkingen, deduplicatie, aangepaste toestand en Python-UDF’s voegen toestandsbeheer, shufflebewerkingen, serialisatie of buffering toe. Voer een benchmark uit op de volledige query met realistische invoersnelheden en een realistische statusomvang, in plaats van een statusloze doorvoerquery te meten.

Important

Als Spark een bron-, sink-, operationele of outputmodus voor Real-time Mode afwijst, voer die query dan uit met een standaard microbatch-trigger. Ga er niet van uit dat een query die werkt in micro-batchmodus ook in Real-time modus werkt.

Als je een aangepaste stateful processor bouwt met transformWithState, verwacht dat het rij-voor-rij uitvoeringsmodel zijn gedrag verandert. Real-time Mode levert gebeurtenissen aan je processor zodra ze binnenkomen in plaats van als een batch per sleutel, dus schrijf de processor zonder aan te nemen dat hij elke rij voor een sleutel in één enkele aanroep ziet. Event-time timers worden niet ondersteund, en verwerkingstijd-timers kunnen worden uitgesteld totdat de data arriveert of de langlopende batch eindigt. De pandas-gebaseerde variant van de API is niet beschikbaar, dus gebruik de rijgebaseerde API. Voor het algemene programmeermodel, zie Stateful stream processing.

Gebruikssituaties

Real-time modus werkt het beste wanneer latentie belangrijker is dan maximale doorvoer of brede dekking van functies. Overweeg het voor workloads zoals eventroutering, operationele waarschuwingen, lichte streamverrijking en overdracht met lage latentie tussen streaming-systemen.

Standaard microbatchmodus is meestal een betere keuze wanneer je:

  • Gegevensinname met hoge doorvoer in een lakehouse-Delta-tabel.
  • Grotere batches voor geoptimaliseerde bestandsgroottes en lagere schrijfoverhead.
  • Complexe aggregaties, samenvoegingen of statusafhankelijke verwerking.
  • Uitgebreide compatibiliteit met connectoren en sinks.
  • Lagere kosten door minder voortdurend actieve rekenkracht.

Kies de modus per werklast. Een werkruimte kan de realtime-modus gebruiken voor latentiegevoelige eventstromen en de microbatchmodus voor lakehouse-ingestie of analytische pijplijnen.

Monitor realtime queries

Monitor Real-time Mode-queries vanaf de Fabric monitoring hub. Open de Spark-applicatie en gebruik het tabblad Gestructureerde Streaming om streamingstatistieken te bekijken zoals invoersnelheid, processnelheid, invoerrijen, querystatus en operationele duur.

Real-time-modus publiceert de voortgang van query’s en checkpoints tussen langlopende batches. Het tabblad Structured Streaming en lastProgress worden bijgewerkt nadat een batch is voltooid, dus de actualiteit ervan hangt af van het triggerinterval van de realtime-modus. Beschouw batchduur of de leeftijd van lastProgress niet als verwerkingslatentie per record.

Gebruik de queryhandle om te controleren of de query actief is, inspecteer de huidige status en haal de meest recente voltooide batch-metrics op:

import json

monitoring_snapshot = {
    "isActive": query.isActive,
    "status": query.status,
    "lastProgress": query.lastProgress,
}

print(json.dumps(monitoring_snapshot, indent=2, default=str))

if not query.isActive and query.exception() is not None:
    raise query.exception()

Bekijk in de details van de Spark-toepassing de actieve stages en executors terwijl een batch met een lange looptijd wordt uitgevoerd. Zoek naar mislukte taken, onbeschikbare executors, aanhoudende druk op CPU of geheugen, en onvoldoende taakslots.

Spark-voortgangsstatistieken alleen zijn niet voldoende voor live operationele monitoring omdat ze bij de batchgrens worden bijgewerkt. Controleer de achterstand van de Kafka-consumentengroep of een vergelijkbare achterstand aan de bronzijde, invoer- en uitvoersnelheden, snelheidsbeperking en fouten in de bron- en doelsystemen. Voor Azure Event Hubs gebruik je Azure Monitor naast de Fabric monitoring hub.

Meet de end-to-end latentie buiten het Spark-voortgangsevenement. Neem de producer event time op in elk record en vergelijk dit vervolgens met de tijd dat een downstream consument het resultaat ontvangt. Deze meting registreert bronvertraging, Spark-verwerking, netwerkoverdracht en sinkvertraging. Waarschuwing bij querybeëindiging, verouderde checkpoints, toenemende bronvertraging, sinkfouten en end-to-end latentie die het workload-doel overschrijdt.

Verwante onderwerpen

  • Last updated on