Auto Loader bewaken en controleren

Automatische Loader-pijplijnen vereisen actieve bewaking om problemen te detecteren, zoals groeiende achterstanden, schemadrift, beschadigde gegevens en vastgelopen stromen voordat ze van invloed zijn op downstreamgebruikers. Op deze pagina wordt beschreven hoe u belangrijke metrische gegevens bewaakt, de status op bestandsniveau opvraagt, dashboards voor waarneembaarheid bouwt en veelvoorkomende problemen oplost.

Zie Auto Loader configureren voor productieworkloads voor details over de productieconfiguratie. Zie Auto Loader best practices voor aanbevolen procedures voor configuratie.

Prerequisites

Verschillende monitoringworkflows op deze pagina maken gebruik van cloud_files_state() om de innamestatus per bestand te monitoren, zoals backlogquery's, latentieberekeningen en detectie van schemadrift. cloud_files_state() is een tabelwaardefunctie die de opnamestatus op bestandsniveau retourneert voor een controlepunt voor het automatisch laden. Niet alle velden zijn standaard beschikbaar. Beschikbaarheid is afhankelijk van uw Databricks Runtime-versie en -configuratie:

• Databricks Runtime 18.2 en hoger: discovery_time, processed_timeen commit_time zijn automatisch beschikbaar. In Databricks Runtime 16.4-18.1 zijn deze velden alleen beschikbaar wanneer cloudFiles.cleanSource deze zijn ingeschakeld.
• Databricks Runtime 16.4 en hoger met cloudFiles.cleanSource ingeschakeld: archive_time, archive_modeen move_location zijn beschikbaar.

Het inschakelen cloudFiles.cleanSource heeft enige overhead voor de prestaties. Voer benchmarks uit op basis van uw workloads in een preproductieomgeving voordat u dit inschakelt in uw productieomgeving.

Additionally:

• Aantekeningen toevoegen aan opgenomen gegevens met de _metadata kolom. Leg minimaal file_path vast en file_modification_time. Zie kolom Met metagegevens van bestand.
• Schakel de _rescued_data- en _corrupt_record-kolommen in.

Belangrijke metrische gegevens voor automatisch laden

De volgende tabel bevat een overzicht van de belangrijkste metrische gegevens die moeten worden bewaakt voor AutoLoader-pijplijnen. Deze statistieken zijn beschikbaar via StreamingQueryListener voortgangsevents, waarbij Auto Loader-specifieke waarden worden weergegeven onder de metrics-map van elke bron.

Metriek	Wat het u vertelt
numFilesOutstanding	Aantal bestanden in de achterstand die moet worden verwerkt
numBytesOutstanding	Grootte van de bestandsachterstand in bytes
approximateQueueSize	Diepte van de cloudwachtrij (alleen in de modus voor bestandsmeldingen)
numInputRows	Aantal verwerkte rijen per verwerkingsbatch
inputRowsPerSecond	Snelheid van aankomst van gegevens
processedRowsPerSecond	Verwerkingssnelheid
durationMs Verdeling	Hoe de tijd in elke batch wordt besteed

Waar moet ik naar kijken?

De volgende patronen geven aan dat uw pijplijn mogelijk aandacht nodig heeft.

• Groeiende numFilesOutstanding: De achterstand loopt op. Je pijplijn kan de binnenkomende data niet bijhouden.
• processedRowsPerSecond < inputRowsPerSecond: De pijplijn verwerkt gegevens langzamer dan deze binnenkomt.
• Groot durationMs.latestOffset: Het vinden van bestanden is traag. Overweeg om over te schakelen naar bestandsgebeurtenissen.
• Groot durationMs.addBatch: gegevensverwerking is traag. Overweeg rekenkracht te schalen of transformaties te optimaliseren.

Zie bronmetriek van Auto Loader voor de volledige naslaginformatie.

Status op bestandsniveau opvragen met cloud_files_state

De cloud_files_state() tabelwaardefunctie biedt gedetailleerde informatie over elk bestand dat is gedetecteerd door automatisch laden. De volgende velden zijn beschikbaar. Velden die zijn gemarkeerd als Databricks Runtime 16.4 en hoger of 18.2 en hoger, worden alleen ingevuld onder de voorwaarden die worden beschreven in Vereisten.

Veld	Type	Description
path	STRING	Het pad van het bestand
size	BIGINT	De grootte van het bestand in bytes
create_time	TIMESTAMP	Toen het bestand is gemaakt
discovery_time	TIMESTAMP	Wanneer automatisch laden het bestand heeft gedetecteerd (Databricks Runtime 16.4 en hoger)
processed_time	TIMESTAMP	Wanneer Auto Loader het bestand heeft verwerkt (Databricks Runtime 16.4 en hoger)
commit_time	TIMESTAMP	Wanneer het bestand is doorgevoerd in het controlepunt (Databricks Runtime 16.4 en hoger)
archive_time	TIMESTAMP	Wanneer het bestand is gearchiveerd (vereist cloudFiles.cleanSource)
archive_mode	STRING	MOVE, DELETE of NULL (vereist cloudFiles.cleanSource)
move_location	STRING	Doelpad wanneer cloudFiles.cleanSource is MOVE
ingestion_state	STRING	Huidige status van bestandsopname

Onderzoek de status van de bestandsinname

De volgende query's hebben betrekking op veelvoorkomende diagnostische scenario's.

Alle niet-verwerkte bestanden zoeken (de huidige achterstand):

SELECT * FROM cloud_files_state('path/to/checkpoint')
WHERE ingestion_state != 'COMMITTED';

Gemiddelde opnamelatentie berekenen (tijd van het maken van bestanden tot doorvoeren):

SELECT avg(unix_timestamp(commit_time) - unix_timestamp(create_time)) AS avg_latency_seconds
FROM cloud_files_state('path/to/checkpoint')
WHERE commit_time IS NOT NULL AND create_time IS NOT NULL;

Beschadigde of overgeslagen bestanden zoeken:

SELECT path, ingestion_state, size, create_time
FROM cloud_files_state('path/to/checkpoint')
WHERE ingestion_state LIKE 'SKIPPED%';

Voortgang van archivering bijhouden (vereist cloudFiles.cleanSource):

SELECT archive_mode, count(*) AS file_count
FROM cloud_files_state('path/to/checkpoint')
GROUP BY archive_mode;

Zoek bestanden met een hoge latentie voor detectie-naar-doorvoer om knelpunten te identificeren:

SELECT
  path,
  size,
  unix_timestamp(commit_time) - unix_timestamp(discovery_time) AS processing_latency_seconds,
  unix_timestamp(commit_time) - unix_timestamp(create_time) AS end_to_end_latency_seconds
FROM cloud_files_state('path/to/checkpoint')
WHERE commit_time IS NOT NULL
ORDER BY end_to_end_latency_seconds DESC
LIMIT 20;

Zie de tabelwaardefunctie voor de volledige SQL-verwijzingcloud_files_state.

Auto Loader bewaken in Lakeflow-pijplijnen

Databricks raadt aan om Lakeflow-pijplijnen te gebruiken voor Auto Loader-pijplijnen in productie. Om te profiteren van de ingebouwde bewakingsmogelijkheden:

• Sla het gebeurtenislogboek van Lakeflow-pijplijnen op in een Delta-tabel, zodat deze kan worden opgevraagd voor waarneembaarheidsgegevens. Configureer dit via de geavanceerde instellingen van de pijplijn of de API. Zie het gebeurtenislogboek van de pijplijn voor meer informatie.

• Structureer uw pijplijn voor waarneembaarheid. Een goed gestructureerde AutoLoader-pijplijn in Lakeflow-pijplijnen bevat een {table}_source weergave (de brondefinitie van de autolader), een {table}_bronze streamingtabel (onbewerkte gegevensopname met _rescued_data en _corrupt_record kolommen), een corrupt_records_sink pijplijn die rijen met onparse gegevens in quarantaine plaatst en een {table} schone weergave voor downstreamverbruik.

• Definieer verwachtingen voor uw bronze-streamingtabellen om schemadrift en gegevenscorruptie te monitoren. _rescued_data IS NULL detecteert onverwachte schemawijzigingen en _corrupt_record IS NULL detecteert onherstelbare gegevens. Lakeflow-pijplijnen evalueren deze verwachtingen wanneer gegevens binnenkomen en genereren een waarneembaarheidsspoor. U kunt verwachtingen configureren om een waarschuwing te geven, rijen te verwijderen of de pijplijn te laten mislukken.

Nadat u de event_log_raw-weergave voor uw pijplijn hebt gemaakt, gebruikt u de volgende query’s voor Auto Loader-specifieke metriek.

Doorvoer voor opname per stroom bewaken:

SELECT
  origin.flow_name,
  origin.update_id,
  timestamp,
  TRY_CAST(details:flow_progress.metrics.num_output_rows AS BIGINT) AS rows_written
FROM event_log_raw
WHERE event_type = 'flow_progress'
ORDER BY timestamp DESC;

Gegevensachterstand per stroom bewaken:

SELECT
  origin.flow_name,
  timestamp,
  DOUBLE(details:flow_progress.metrics.backlog_bytes) AS backlog_bytes
FROM event_log_raw
WHERE event_type = 'flow_progress'
  AND details:flow_progress.metrics.backlog_bytes IS NOT NULL
ORDER BY timestamp DESC;

Samenvatting van schendingen van de verwachting om schemadrift en beschadigde gegevens te detecteren:

SELECT
  origin.flow_name,
  explode(from_json(
    details:flow_progress.data_quality.expectations,
    'array<struct<name:string, dataset:string, passed_records:bigint, failed_records:bigint>>'
  )) AS expectation
FROM event_log_raw
WHERE event_type = 'flow_progress'
  AND details:flow_progress.data_quality.expectations IS NOT NULL;

Raadpleeg Pijplijnen bewaken en Gebeurtenislogboek van de pijplijn voor algemene richtlijnen voor het bewaken van Lakeflow-pijplijnen.

Auto Loader bewaken met Structured Streaming

Wanneer u Auto Loader buiten Lakeflow-pijplijnen uitvoert, gebruikt u de volgende methoden om Structured Streaming te monitoren.

• Implementeer een StreamingQueryListener om Auto Loader-specifieke metrieke gegevens van elke batch vast te leggen door uit source.metrics te lezen.

from pyspark.sql.streaming import StreamingQueryListener

class AutoLoaderMonitor(StreamingQueryListener):
    def onQueryStarted(self, event):
        pass

    def onQueryProgress(self, event):
        for source in event.progress.sources:
            if "CloudFilesSource" in source.description:
                metrics = source.metrics
                files_outstanding = metrics.get("numFilesOutstanding", "0")
                bytes_outstanding = metrics.get("numBytesOutstanding", "0")
                rows_per_sec = source.processedRowsPerSecond
                # Push metrics to your monitoring system (for example, write to a Delta table)

    def onQueryIdle(self, event):
        pass

    def onQueryTerminated(self, event):
        pass

spark.streams.addListener(AutoLoaderMonitor())

Opmerking

Verwerkingslogica in listeners kan de verwerking van query's vertragen. Beperk het rekenwerk in listener-callbacks en vermijd daar synchrone externe schrijfbewerkingen; stuur in plaats daarvan lichte telemetrie asynchroon uit of geef metingen door aan een afzonderlijke taak voor opslag.

• Gebruik numInputRows, inputRowsPerSeconden processedRowsPerSecond van de voortgang van de bron tot het berekenen van doorvoer: bestanden per seconde en rijen per seconde voor elke batch.

• Vergelijk create_time en commit_time uit cloud_files_state() om de opnamelatentie voor de end-to-end-latentie te berekenen. Voor de verwerking van latentie gebruikt u de durationMs uitsplitsing (bijvoorbeeld latestOffset, addBatchen andere gerapporteerde batchfasen) om te bepalen welke fase het knelpunt is.

• Gebruik df.observe() om inline gegevenskwaliteitsmetriek rechtstreeks in het streaming-DataFrame te definiëren. Metrische gegevens zijn zichtbaar in StreamingQueryListener voortgangsevenementen onder observedMetrics.

from pyspark.sql.functions import count, lit, col

observed_df = df.observe(
    "auto_loader_quality",
    count(lit(1)).alias("total_rows"),
    count(col("_rescued_data")).alias("rescued_rows"),
    count(col("_corrupt_record")).alias("corrupt_rows")
)

• Gebruik .queryName() deze functie om een unieke naam toe te wijzen aan elke stream, zodat u autoladerstreams gemakkelijker kunt onderscheiden op het tabblad Streaming van de Spark-gebruikersinterface en in bewakingsdashboards.

Zie Structured Streaming-query's bewaken in Azure Databricks voor de volledige referentie over het bewaken van Structured Streaming-query's.

Een dashboard voor waarneembaarheid bouwen

Combineer gegevens uit meerdere bronnen om een uitgebreid waarneembaarheidsdashboard te bouwen voor uw AutoLoader-pijplijnen. In deze tabel worden enkele voorgestelde bronnen weergegeven die u kunt gebruiken om uw waarneembaarheidsdashboard te structuren.

Gegevensbron	Waarneembaarheidsgegevens
cloud_files_state()	Opnamestatus op bestandsniveau: detectie, verwerking, doorvoer en archiveringstijdstempels per bestand
Gebeurtenislogboek van Lakeflow-pijplijnen	Uitvoeringsgeschiedenis van de pijplijn, stroommetingen per batch en resultaten van verwachtingen voor gegevenskwaliteit
Pijplijnuitvoertabellen	Aantal rijen en gegevensvolume geschreven per opgenomen tabel

Vervolgens kunt u waarneembaarheidsgegevens samenvoegen in toegewezen tabellen die fungeren als de basis voor dashboards en waarschuwingen:

• De statussen van pipeline-uitvoeringen (geslaagd of mislukt) in de tijd samenvatten, afgeleid van event_type = 'update_progress'-gebeurtenissen.
• Statistische metrische gegevens over bestandsopname (achterstandsgrootte, doorvoer, latentie per batch), afgeleid van cloud_files_state() en event_type = 'flow_progress' gebeurtenissen.
• Tabelstatistieken ontwikkelen met behulp van rijaantallen en gegevensvolume per tabel, afgeleid van num_output_rows in het gebeurtenislogboek.
• Verzamel foutopsporingsgegevens uit gedetailleerde foutlogboeken en verwachtingsschendingen per update, afgeleid van event_type = 'flow_progress'-gebeurtenissen waarbij data_quality is ingevuld.

Deze samengevoegde tabellen kunnen een AI/BI-dashboard en SQL-waarschuwingen mogelijk maken. Aanbevolen dashboardvensters omvatten tijdlijn voor pijplijnuitvoeringsstatus, opnameachterstandstrend, doorvoertrend, distributie van opnamelatentie, metrische gegevens over gegevenskwaliteit, schemaontwikkelingsgebeurtenissen en status van bestandsarchivering.

Gebeurtenissen met schemawijzigingen bewaken

Gebruik de volgende methoden om schemawijzigingen te detecteren wanneer ze optreden.

• Niet-NULL-waarden in _rescued_data in de aantallen verwachtingsovertredingen duiden op schemadrift. Doorzoek het gebeurtenislogboek voor failed_records > 0 op de verwachting van no rescued data.
• Wijzigingen in de _schemas map in het geconfigureerde cloudFiles.schemaLocation (of alleen binnen het controlepunt wanneer de schemalocatie niet afzonderlijk is ingesteld) geven aan dat de ontwikkeling van het schema is opgetreden. U kunt deze map peilen vanuit een afzonderlijke bewakingstaak.
• Behandel een onQueryTerminated gebeurtenis niet gevolgd door onQueryStarted voor dezelfde stroomnaam als voldoende bewijs van de ontwikkeling van schema's. Streams worden om verschillende redenen opnieuw opgestart (cluster opnieuw opstarten, code-implementaties, tijdelijke opslagfouten). Correleer opnieuw opstarten met onafhankelijke signalen ( _schemas wijzigingen in mappen of _rescued_data schendingen van verwachtingen) voordat wordt geconcludeerd dat de ontwikkeling van het schema is opgetreden.
• Gebruik _metadata.file_path dit om te bepalen welke bestanden schemawijzigingen hebben geïntroduceerd. Voeg dit toe aan cloud_files_state() het path veld om schemawijzigingen te correleren met specifieke bestanden en batches.

Gebruik deze voorbeeldquery om recente schemadrift te detecteren via schendingen van verwachtingen:

SELECT
  timestamp,
  origin.flow_name,
  exp.name AS expectation_name,
  exp.failed_records
FROM (
  SELECT
    timestamp,
    origin,
    explode(from_json(
      details:flow_progress.data_quality.expectations,
      'array<struct<name:string, dataset:string, passed_records:bigint, failed_records:bigint>>'
    )) AS exp
  FROM event_log_raw
  WHERE event_type = 'flow_progress'
    AND details:flow_progress.data_quality.expectations IS NOT NULL
)
WHERE exp.name = '<rescued-data expectation name>'
  AND exp.failed_records > 0
ORDER BY timestamp DESC;

Waarschuwingen instellen voor veelvoorkomende problemen

Gebruik Databricks SQL-waarschuwingen of pijplijnmeldingen om problemen te detecteren voordat ze van invloed zijn op downstreamgebruikers.

De volgende SQL detecteert een groeiende achterstand en kan worden gebruikt als basis voor een Databricks SQL-waarschuwing. Stel in dat dit periodiek wordt uitgevoerd (bijvoorbeeld elke 5 minuten) en geef een waarschuwing wanneer het resultaat niet leeg is.

-- Alert when backlog exceeds threshold or trends upward across recent batches
WITH recent_backlog AS (
  SELECT
    origin.flow_name,
    timestamp,
    DOUBLE(details:flow_progress.metrics.backlog_bytes) AS backlog_bytes,
    ROW_NUMBER() OVER (PARTITION BY origin.flow_name ORDER BY timestamp DESC) AS rn
  FROM event_log_raw
  WHERE event_type = 'flow_progress'
    AND details:flow_progress.metrics.backlog_bytes IS NOT NULL
)
SELECT flow_name, backlog_bytes, timestamp
FROM recent_backlog
WHERE rn = 1
  AND backlog_bytes > 1073741824  -- alert when backlog exceeds 1 GB

De volgende tabel bevat een overzicht van aanbevolen waarschuwingsvoorwaarden:

Wat u moet detecteren	Hoe u dit kunt detecteren	Wanneer moet u een waarschuwing geven
Groeiende achterstand	numFilesOutstanding stijgend	Aanhoudende toename over meerdere batches
Vastgelopen stroom	Geen voortgangsevenementen	Geen gebeurtenissen voor N minuten (op basis van het verwachte triggerinterval)
Hoge opnamelatentie	commit_time - create_time	De SLA-drempelwaarde overschrijdt
Degradatie van gegevenskwaliteit	Verwachtingsfoutpercentage	Het toenemende percentage rijen dat niet aan de verwachtingen voldoet
Gebeurtenis schemaontwikkeling	_rescued_data IS NOT NULL	Alle niet-NULL-waarden in het aantal verwachtingsovertredingen
Traag zoeken naar bestanden	durationMs.latestOffset	Aanzienlijk hoger dan basislijn

Veelvoorkomende problemen oplossen

In de volgende tabel worden veelvoorkomende problemen met de pijplijn voor automatisch laadprogramma's, de waarschijnlijke oorzaken en aanbevolen acties beschreven om ze op te lossen.

Issue	Mogelijke oorzaak	Aanbevolen actie
Achterstand groeit sneller dan verwerking	Te weinig rekenkracht, scheve datadistributie of geknepen aanvraaglimieten	Schaal de rekenkracht op, controleer op scheefheid met de Spark UI en bekijk de maxFilesPerTrigger instellingen om de batchgrootte te beheren
Bestanden worden niet gedetecteerd	Bestandsgebeurtenissen zijn onjuist geconfigureerd, machtigingsprobleem of stream worden niet binnen 7 dagen uitgevoerd	Controleer de machtigingen voor externe opslaglocaties, controleer de configuratie van bestandsgebeurtenissen in de Unity Catalog-interface en zorg ervoor dat de stream minstens elke 7 dagen wordt uitgevoerd om te voorkomen dat de RocksDB-status verloopt
Het opstarten van stream duurt te lang	Grote controlepuntstatus downloaden (RocksDB)	Upgrade naar Databricks Runtime 15.3 en hoger voor asynchroon laden van status, waardoor de opstarttijd met ongeveer 90% wordt verkort
Dubbele bestandsverwerking	Agressieve cloudFiles.maxFileAge instellingen of beschadiging van controlepunt	Gebruik een conservatieve maxFileAge (minimaal 90+ dagen), controleer de integriteit van controlepunten en vermijd levenscyclusbeleid voor controlepuntopslag
Schemaontwikkeling waardoor pijplijn opnieuw wordt opgestart	Regelmatige of niet-compatibele schemawijzigingen	Controleer schemaEvolutionMode, schakel over op addNewColumnsWithTypeWidening voor typepromoties, of gebruik het varianttype voor zeer dynamische schema's
Corrupte gegevens stapelen zich op in sink	Problemen met kwaliteit van brongegevens	Controleer de _corrupt_record quarantaine-sink op patronen, bekijk hoe brongegevens worden gegenereerd en overweeg upstream-validatie toe te voegen
discovery_time en commit_time niet ingevuld	Wordt uitgevoerd op Databricks Runtime lager dan 18.2 zonder cleanSource	Voer een upgrade uit naar Databricks Runtime 18.2 en hoger of schakel cloudFiles.cleanSource in op Databricks Runtime 16.4–18.1

Zie veelgestelde vragen over automatisch laden voor aanvullende probleemoplossing.