Best practices voor gestructureerde streaming

Spark Structured Streaming biedt een enkele incrementele verwerkings-API voor continu draaiende streams en geplande batch-achtige taken. In Microsoft Fabric gebruik je deze best practices om beide patronen betrouwbaar uit te voeren, kosten te beheersen, de status te behouden en gezonde Delta-tabellen in een lakehouse te schrijven.

Dit artikel richt zich op ontwerp- en bedieningskeuzes voor streaming-workloads. Voor een end-to-end voorbeeld dat streaminggegevens naar een lakehouse wegschrijft, zie Data streaming into a lakehouse with Spark.

Voer productieworkloads uit

Gebruik notitieboeken om streaminglogica te ontwikkelen en te testen. Voer productiequeries uit als Spark-taakdefinities in plaats van interactieve notitieboeken.

Important

Streamingquery's worden asynchroon uitgevoerd. Een actieve query voorkomt niet dat een getriggerde notebook- of Spark-taakdefinitie een terminaltoestand bereikt. Roep query.awaitTermination() aan voor elke query waar de taak op moet wachten. Voor meerdere always-on queries, roep spark.streams.awaitAnyTermination() aan om te detecteren wanneer een query stopt, en behandel vervolgens die beëindiging en de resterende queries.

Kies de levensduur van de query op basis van de werklast:

  • Gebruik een trigger in de standaardmodus, met een vast interval of in de realtime-modus voor een continu draaiende taak. Configureer een onbeperkt herkansingsbeleid wanneer de taak actief moet blijven vanwege tijdelijke infrastructuurstoringen, onderhoudsgebeurtenissen of time-outs.
  • Gebruik de available-now trigger voor een geplande incrementele taak. Elke run verwerkt alle momenteel beschikbare invoer en stops. Plan latere runs met dezelfde query en checkpoint om de voortgang en status van de bron voort te zetten zonder Spark-berekening actief te houden tussen de runs.

De querylogica kan hetzelfde blijven wanneer je schakelt tussen geplande en altijd ingeschakelde werking. Het controlepunt behoudt bron-offsets en toestand bij beide patronen. Voordat je een checkpoint met een andere trigger hergebruikt, controleer je of de bron, het queryplan, het statusschema en de sink compatibel blijven.

Raadpleeg voor een stapsgewijze uitleg Get streaming data into lakehouse and access with SQL analytics endpoint. Zie voor richtlijnen voor productieomgevingen die ingaan op Spark-jobdefinities en nieuwpogingsgedrag Gegevens streamen naar een lakehouse met Spark.

Controlepunten beheren

Stel een duurzame checkpointLocation in voor elke streamingzoekopdracht. Het checkpoint slaat offsets, voortgangsmetadata en statusinformatie op die Spark gebruikt om een query na een herstart te hervatten.

Gebruik een aparte checkpointmap voor elke query. Deel geen checkpoint tussen query's, zelfs niet als ze uit dezelfde bron lezen of naar dezelfde tabel schrijven. Gedeelde checkpoints beschadigen herstelmetadata en kunnen leiden tot dubbele verwerking, overgeslagen records of mislukte herstartpogingen.

query = (
    events.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "Files/checkpoints/orders-to-bronze")
    .toTable("bronze_orders")
)

Kies een pad naar het meerhuis als locatie van het controlepunt. Gebruik geen lokale driveropslag voor checkpoints, want de data verdwijnt zodra de Spark-applicatie stopt.

Note

Checkpoints helpen Spark bij het mogelijk maken van fouttolerant herstel en exact-once-verwerking met ondersteunde sinks zoals Delta Lake. Als je aangepaste sinks gebruikt of foreachBatch, maak de schrijflogica idempotent, omdat Spark na een fout een batch opnieuw kan uitvoeren.

Behandel het checkpoint als onderdeel van de geïmplementeerde query. Wijzigingen in bronnen, stateful operators, groeperingssleutels, statusschema’s of typen sinks kunnen incompatibel zijn met een bestaand checkpoint. Test wijzigingen in query's met een kopie van op productie lijkende gegevens en begin met een nieuw checkpoint wanneer Spark herstel van de gewijzigde query niet ondersteunt.

Schakel RocksDB in als streamingstatusopslag

Schakel de RocksDB state store-provider in voor je streamingworkloads.

Standaard gebruikt Spark een door HDFS ondersteunde state store waarvan de actieve status wordt bijgehouden in door de JVM beheerd geheugen, wat de druk op de heap kan verhogen en lange garbagecollectionpauzes kan veroorzaken naarmate de status groeit. RocksDB beheert in plaats daarvan de actieve toestand in het native geheugen en op de lokale schijf van elke executor, terwijl duurzame snapshots of changelogs worden opgeslagen op de checkpoint-locatie van de query in OneLake. Deze aanpak vermindert de druk op de JVM-heap en zorgt voor een voorspelbaarder geheugengebruik. Het is gunstig voor stateful query’s zoals vensteraggregaties, stream-stream-joins, deduplicatie en transformWithState, en is een verstandige standaardkeuze, zelfs wanneer query’s weinig of geen status bijhouden.

spark.conf.set(
    "spark.sql.streaming.stateStore.providerClass",
    "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider"
)

Stel de state store-provider in vóór de eerste run van de query en houd deze consistent bij herstarts.

Checkpointing van RocksDB-wijzigingslogboeken inschakelen

Apache Spark ondersteunt changelog-checkpointing als optimalisatie voor de RocksDB state store. In plaats van bij elk checkpoint een RocksDB-snapshot te uploaden, uploadt Spark de wijzigingen sinds het vorige checkpoint en maakt periodiek snapshots op de achtergrond. Schakel deze functie in om de checkpoint-latentie te verminderen voor query’s met veel statusgegevens.

spark.conf.set(
    "spark.sql.streaming.stateStore.rocksdb.changelogCheckpointing.enabled",
    True
)

Changelog-checkpointing is achterwaarts compatibel met RocksDB-snapshot-controlepunten, dus je kunt dit in- of uitschakelen voor een bestaande RocksDB-query zonder de state te hoeven weggooien. Herstart de query na het wijzigen van de instelling en test de checkpoint- en herstelprestaties met een statusomvang die vergelijkbaar is met productie.

Blijf een begrensde status houden met watermerken, joinvoorwaarden op basis van tijdsbereik en time-to-live (TTL), zelfs wanneer je RocksDB gebruikt. RocksDB verandert waar de status zich bevindt, maar het verwijdert niet de noodzaak om te bepalen hoeveel staat de query behoudt. Voor meer informatie over operators met status en statusinspectie, raadpleeg Stateful stream processing.

Optimaliseer Delta-sinks

Streamingtaken maken vaak kleine bestanden, omdat elke microbatch gegevens onafhankelijk wegschrijft. Kleine bestanden verhogen de overhead van metadata en vertragen downstream reads. Gebruik schrijfstrategieën en onderhoudsstrategieën samen:

  • Schakel Optimize Write in zodat Delta Lake de bestandsgroottes kan verbeteren voordat het data schrijft.
  • Gebruik triggerintervallen om input te batchen in minder, grotere commits.
  • Kies partitiekolommen die overeenkomen met veelgebruikte queryfilters en maak niet te veel kleine directories.
  • Schakel automatische verdichting in wanneer de streamingwerklast periodieke extra latentie van verdichting tolereert om de opbouw van kleine bestanden te beperken.

Voor gedetailleerde voorbeelden van Optimize Write, trigger batching en partitionering, zie Data streaming naar een lakehouse met Spark. Voor compactie-opties, zie Tabel compactie.

spark.conf.set("spark.databricks.delta.optimizeWrite.enabled", True)

query = (
    events.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "Files/checkpoints/orders-to-silver")
    .partitionBy("event_date")
    .trigger(processingTime="1 minute")
    .toTable("silver_orders")
)

Begin met een triggerinterval dat overeenkomt met je freshness-doel, en inspecteer vervolgens de bestandsgroottes en batchduur. Kortere intervallen verminderen de latentie, maar creëren meestal meer commits en kleinere bestanden.

Schema's beheren

Bied een schema voor streamingopdrachten:

  • Bestandsgebaseerde streamingbronnen, waaronder Parquet, JSON, CSV en ORC, vereisen standaard een expliciet schema. Spark kan het schema van een bestandsbron alleen afleiden wanneer je spark.sql.streaming.schemaInference inschakelt, maar schema-afleiding wordt niet aanbevolen voor productiestreams.
  • Kafka-compatibele bronnen bieden een vast bronschema voor velden zoals key, value, , topicen timestamp. Geef een schema wanneer je de berichtwaarde ontleedt in gestructureerde kolommen.
  • Delta-streamingbronnen lezen het schema uit de metadata van de Delta-tabel. Behandel incompatibele wijzigingen in het brontabelschema als geplande wijzigingen in de streamingquery.

Expliciete schema's maken parsing voorspelbaar, voorkomen herhaalde bestandsinspectie en helpen je ongeldige records te identificeren voordat ze downstream tabellen bereiken.

from pyspark.sql import functions as F
from pyspark.sql.types import DoubleType, StringType, StructField, StructType, TimestampType

order_schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("customer_id", StringType(), False),
    StructField("amount", DoubleType(), True),
    StructField("event_time", TimestampType(), False),
])

parsed_events = (
    raw_events
    .withColumn("body_text", F.col("body").cast("string"))
    .withColumn("order", F.from_json("body_text", order_schema))
    .select("order.*")
)

Ga doelbewust om met schema drift. Wanneer producenten velden toevoegen, bepaal dan of je de velden negeert, ze vastlegt in een ruwe bronzen tabel, of het doelschema bijwerkt. Wanneer producenten datatypes wijzigen of vereiste velden verwijderen, stuur de getroffen records naar een quarantainetabel voordat ze de streamingquery breken.

Plan ook op wijzigingen aan een Delta-bron. Een standaard Delta-streaming-leesbewerking verwacht commits waarbij alleen gegevens worden toegevoegd en kan mislukken wanneer de brontabel wordt bijgewerkt, er gegevens uit worden verwijderd, ermee wordt samengevoegd of deze wordt overschreven. Gebruik change data feed wanneer downstream-verwerking wijzigingen op rijniveau moet kunnen waarnemen. Gebruik opties die change commits alleen overslaan als het bewust negeren van die wijzigingen correct is voor de workload.

Voor een stateful query die tijdens het lezen van een bestaande Delta-tabel een event-time-watermark toepast, kunt u overwegen withEventTimeOrder in te schakelen voor de initiële snapshot. Het verwerken van de initiële bestanden op volgorde van wijzigingstijd kan er anders toe leiden dat oudere gebeurtenissen worden weergegeven nadat het watermerk is opgeschoven, waardoor Spark deze als te laat binnengekomen gegevens behandelt.

Houd Kafka-compatibele connecties gezond

Kafka-client idle-instellingen kunnen conflicteren met een beheerde Kafka-compatibele service. De Kafka Java-client staat metadata.max.age.ms standaard op 300.000 milliseconden en connections.max.idle.ms op 540.000 milliseconden. Azure Event Hubs sluit een inactieve Kafka-verbinding na 240.000 milliseconden. De client kan daarom proberen een verbinding te hergebruiken die Event Hubs al heeft gesloten, wat kan verschijnen als een verlopen batch, een verzendtime-out of een latentiegap terwijl de client opnieuw verbinding maakt.

Voor Event Hubs Kafka-compatibele bronnen en sinks stel beide waarden in op minder dan 240.000 milliseconden. Het kortere metadata-interval stuurt een verzoek vóór de idle-limiet van Event Hubs, terwijl de kortere idle-drempel van de client ervoor zorgt dat de client een idle-verbinding beëindigt voordat de service dat doet.

event_hubs_idle_options = {
    "kafka.connections.max.idle.ms": "180000", # set to less than 240,000 ms
    "kafka.metadata.max.age.ms": "180000", # set to less than 240,000 ms
}

source = (
    spark.readStream
    .format("kafka")
    .options(**input_options)
    .options(**event_hubs_idle_options)
    .load()
)

sink = (
    source.writeStream
    .format("kafka")
    .options(**output_options)
    .options(**event_hubs_idle_options)
)

Pas deze opties toe op beide kanten van een Kafka-naar-Kafka-zoekopdracht. Deze afstelling is vooral belangrijk voor query’s met weinig verkeer en query’s in de realtime-modus, waarbij de verbinding rond de grens van een langlopende batch inactief kan zijn. Als je Apache Kafka of een andere Kafka-compatibele dienst gebruikt, verander dan de standaardinstellingen alleen wanneer de broker, load balancer of firewall een kortere idle-timeout heeft dan de client.

Voor de volledige aanbevelingen en beperkingen van Event Hubs, zie Apache Kafka client configurations for Azure Event Hubs.

Archiveer of verwijder verwerkte bronbestanden

Een bestandsbron-checkpoint registreert welke bestanden Spark heeft verwerkt, maar verwijdert die bestanden niet. Laat een landingsdirectory niet zonder limiet groeien. Grote mappen verhogen de opslag- en lijstoverhead, wat de ontdekking van nieuwe bestanden vertraagt.

Gebruik cleanSource="archive" om verwerkte bestanden buiten het bronpad te verplaatsen. Houd de archiefmap gescheiden zodat dezelfde stream de gearchiveerde bestanden niet opnieuw kan ontdekken:

file_orders = (
    spark.readStream
    .schema(order_schema)
    .format("json")
    .option("cleanSource", "archive")
    .option("sourceArchiveDir", "Files/archive/orders")
    .load("Files/landing/orders")
)

query = (
    file_orders.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "Files/checkpoints/file-orders")
    .toTable("bronze_orders")
)

Als je de bronbestanden niet hoeft te bewaren, verwijder ze dan permanent na verwerking door de archiefopties te vervangen door:

file_orders = (
    spark.readStream
    .schema(order_schema)
    .format("json")
    .option("cleanSource", "delete")
    .load("Files/landing/orders")
)

Gebruik ingebouwde cleanup alleen wanneer geen enkel ander proces dezelfde bronbestanden nodig heeft. Cleanup verloopt asynchroon en kan achterlopen op een stroom met veel volume. Houd de landingsmap in de gaten en gebruik een apart retentie- of levenscyclusproces wanneer ingebouwde opruiming het tempo niet kan bijhouden.

Fouten en veerkracht behandelen

Behandel een continu actieve streamingtaak als een herstartbare dienst, niet als een proces dat voor onbepaalde tijd zonder onderbreking draait. De Spark-applicatie kan uiteindelijk stoppen door een run timeout, onderhoud aan de infrastructuur, een storing in driver- of executor, een tijdelijk netwerkprobleem, bronbeperking of een niet-beschikbare sink. Ontwerp elke laag zodat een nieuwe applicatie veilig kan worden hervat.

Gebruik het volgende antwoord voor elk fouttype:

Storingssoort Antwoord
Tijdelijke bron-, sink-, netwerk- of infrastructuurstoring Laat connector-retries korte onderbrekingen afhandelen, en laat vervolgens het Spark-taakdefinitie-herkansingsbeleid een nieuwe applicatie starten.
Geplande time-out, onderhoudsgebeurtenis of inzet Herstart dezelfde querydefinitie vanaf het bestaande checkpoint.
Ongeldig record Behoud de ruwe gegevenslading en stuur het record naar een quarantaine-tabel.
Ongeldige inloggegevens, een incompatibele schemawijziging of een beschadigd controlepunt Waarschuw en vereis ingrijpen in plaats van eindeloos opnieuw op te starten zonder onderzoek.
Aanhoudende inputachterstand Schaal de rekencapaciteit, verminder het werk per record, stem de bron af of pas het triggerinterval aan.

Configureer bewaakte herstarts

Voer productiestromen uit als Spark-taakdefinities en configureer een herprobeerbeleid. Gebruik een onbeperkt aantal nieuwe pogingen voor een workload die altijd actief is en moet herstellen na verwachte beëindiging van de toepassing, en stel een interval tussen nieuwe pogingen in dat een te snelle herstartlus voorkomt.

Bel query.awaitTermination() nadat je de query hebt gestart. Deze aanroep houdt de getriggerde Spark-taakdefinitie of notebook draaiende en brengt een terminalquery-uitzondering naar de hoofdapplicatie. Voor meerdere query’s die altijd actief zijn, roep spark.streams.awaitAnyTermination() aan, controleer welke query is gestopt en laat de resterende query’s mislukken of stop ze, zodat het beleid voor opnieuw proberen op taakniveau de volledige applicatie opnieuw kan starten.

Vang geen terminal-streaminguitzondering en geef geen succes terug. Een succes-gekleurde exit voorkomt dat het job retry-beleid het falen herkent. Koppel onbeperkte herhalingen met waarschuwingen bij herhaalde mislukkingen, herstartfrequentie en tijd sinds het laatste succesvolle checkpoint. Deterministische fouten, zoals ongeldige inloggegevens of incompatibele toestandswijzigingen, moeten worden gecorrigeerd in plaats van een eindeloze herkansingslus.

Zie Run streaming jobs in production voor de retryconfiguratie van Fabric.

Herstelstatus behouden tussen applicaties

Houd controlepunten in duurzame opslag in OneLake en hergebruik hetzelfde controlepunt na een time-out, onderhoudsgebeurtenis of tijdelijke storing. Een nieuwe Spark-applicatie reconstrueert de bronvoortgang en -status van het checkpoint, en gaat vervolgens verder vanaf de laatst toegewezen batch.

Houd de bron herspeelbaar en zorg dat elke sink idempotent is. Voor Delta-sinks met writeStream coördineren Spark en Delta commits via checkpointmetadata. Voor foreachBatch en aangepaste sinks gebruik je stabiele event keys, de batch-ID of transactie-id’s van de bestemming, zodat opnieuw afspelen geen dubbele bedrijfsrecords aanmaakt.

Verwijder geen checkpoint om te herstellen van een tijdelijke fout. Gebruik een nieuw checkpoint alleen voor een opzettelijk incompatibele querywijziging, en behandel die wijziging als een nieuwe implementatie met een gedefinieerd replay- of start-offsetplan. Zorg ervoor dat inloggegevens, bibliotheken, omgevingsinstellingen en bron- of sinkconfiguraties opnieuw kunnen worden aangemaakt telkens wanneer Fabric een nieuwe Spark-applicatie start.

Isoleer ongeldige records

Ontwerp streamingopdrachten zodat één slecht record de pipeline niet stopt. Bewaar de ruwe payload, parseer deze naar getypeerde kolommen en scheid de ongeldige records voor latere controle.

parsed = (
    raw_events
    .withColumn("body_text", F.col("body").cast("string"))
    .withColumn("order", F.from_json("body_text", order_schema))
)

valid_orders = parsed.where(F.col("order").isNotNull()).select("order.*")

invalid_orders = (
    parsed
    .where(F.col("order").isNull())
    .select("body_text", F.current_timestamp().alias("quarantine_time"))
)

Plangestuurde uitschakeling en inzet

Wanneer je code het afsluiten regelt, roep query.stop() aan en wacht totdat de query is beëindigd voordat de applicatie wordt afgesloten. Als een geplande implementatie eerst alle momenteel beschikbare input moet verwerken, gebruik dan een available-now uitvoering met een compatibele query en checkpoint. Voor een implementatie kun je het bestaande checkpoint alleen behouden wanneer de bron, het queryplan, het statusschema en de sink compatibel blijven.

Vertrouw niet op een gracieuze shutdown voor correctheid. Infrastructuurstoringen kunnen een applicatie zonder waarschuwing beëindigen, dus checkpoint-herstel en het idempotente gedrag van sinks moeten ook met abrupte beëindiging kunnen omgaan.

Achterstand beheersen na herstel

Een opnieuw gestarte query kan resulteren in een grote achterstand. Als de invoersnelheid consequent de processnelheid overschrijdt, verhoog dan de capaciteit, verminder het werk per record, pas de bronsnelheid aan, of gebruik een langer triggerinterval zodat elke batch genoeg tijd heeft om af te ronden voordat de volgende batch begint. Houd de hersteltijd en de retentie van de bron in de gaten, zodat de bron geen ongelezen gegevens verwijdert voordat de query heeft ingehaald.

Streamingqueries monitoren

Gebruik het tabblad Gestructureerde Streaming in de Fabric monitoring hub om streaming-queries te observeren. Bekijk deze statistieken samen in plaats van te vertrouwen op één enkel getal:

  • Input Rate geeft aan hoe snel records binnenkomen.
  • Process Rate laat zien hoe snel Spark records verwerkt.
  • De batchduur geeft aan hoe lang elke microbatch duurt.
  • Input Rows helpt je bij het detecteren van onderbrekingen in de bron, pieken en het opnieuw afspelen na een herstart.
  • Operation Duration helpt je dure lees-, verwerkings- en schrijffasen te identificeren.

Als de batchduur toeneemt terwijl de verwerkingssnelheid achterblijft bij de invoersnelheid, bouwt de query een achterstand op. Onderzoek het bronvolume, transformaties, schuddegrootte, sink latency en Delta-bestandsindeling.

Breng latentie en kosten in balans

Kies een zoektijdsduur en trigger die past bij de zakelijke behoeften. Een direct beschikbare job geeft rekenkracht vrij nadat de huidige backlog is verwerkt, wat goed werkt wanneer periodieke actualiteit voldoende is. Een continu lopende query voorkomt het opstarten van de planner en verwerkt nieuwe invoer zonder te wachten op de volgende geplande uitvoering.

Voor continu draaiende microbatch-taken verlagen korte intervallen de end-to-endlatentie, maar ze kunnen meer kleine bestanden genereren. Langere intervallen verbeteren de batchefficiëntie, maar data komt later in de tabel binnen.

Real-time Mode richt zich op lagere latentie voor ondersteunde Spark Structured Streaming workloads. Gebruik het wanneer je situatie lage latentie en record-voor-record verwerking nodig heeft in plaats van standaard microbatchverwerking. Voor details en beperkingen, zie Real-time Mode. Voor triggergedrag en uitvoermodi, zie Triggers en outputmodi.

Samenvatting van aanbevolen procedures

  • Voer productiequeries uit als Spark-taakdefinities. Gebruik een standaardtrigger, een trigger met een vast interval of een trigger in realtime-modus met herkansingsbeleid voor altijd ingeschakelde verwerking, of gebruik de modus 'Nu beschikbaar' voor geplande incrementele verwerking.
  • Stel een duurzame checkpointLocation code in voor elke query en deel geen checkpoints tussen queries.
  • Schakel de RocksDB-statusopslagprovider en changelog-checkpointing in om heapdruk, garbagecollectionpauzes en checkpointlatentie te verminderen.
  • Gebruik Delta Lake als gootsteen wanneer je betrouwbare transacties en herstel nodig hebt.
  • Activeer Optimize Write, kies triggerintervallen weloverwogen en compacteer Delta-tabellen wanneer kleine bestanden zich ophopen.
  • Definieer expliciete schema's voor parsen en behandel schema-drift als een gepland compatibiliteitsevent.
  • Stem de idle-instellingen van Kafka-clients af op de time-outwaarden van Event Hubs of andere Kafka-compatibele services.
  • Archiveer of verwijder verwerkte bronbestanden zodat bestandsbronmappen niet onbeperkt groeien.
  • Houd gifberichten uit de hoofdtabel door ongeldige records naar een quarantainetabel te schrijven.
  • Behandel elke continu actieve taak als herstartbaar, configureer bewaakte nieuwe pogingen en herstel vanaf een persistent checkpoint.
  • Maak aangepaste sinks en foreachBatch-logica idempotent.
  • Monitor invoersnelheid, processnelheid, batchduur en operationele duur in het tabblad Gestructureerde Streaming.
  • Balanceer latentie, bestandsgrootte en rekenkosten wanneer je kiest tussen microbatch-triggers en Real-time Mode.

Verwante onderwerpen

  • Last updated on