Aanbevolen procedures voor automatisch laden

Op deze pagina worden best practices beschreven die u kunt toepassen om Auto Loader zo te configureren dat het voor uw gebruiksscenario betrouwbaar, kosteneffectief en op schaal wordt uitgevoerd.

Deze best practices verminderen operationele overhead en voorkomen veelvoorkomende problemen die moeilijk te diagnosticeren zijn in productie, zoals: onnodige LIST API-kosten van volledige directoryscans, stille gegevensverlies van schemadrift en pijplijnherstel veroorzaakt door onjuiste configuratie van controlepunten.

Zie Auto Loader configureren voor productieworkloads voor details over de productieconfiguratie. Zie Auto Loader controleren en observeren voor controle en waarneembaarheid.

Het juiste uitvoeringsframework kiezen

Het beste uitvoeringsframework voor uw use-case is afhankelijk van hoeveel controle u nodig hebt over de pijplijn en hoeveel operationele overhead u wilt beheren. Voor de meeste gebruikers en productiepijplijnen is Auto Loader met Lakeflow-pijplijnen geschikt. Als u echter maximale controle en aanpassing nodig hebt, gebruikt u autolader met gestructureerd streamen. Gebruik voor de eenvoudigste installatie met een beheerde ervaring een beheerde LakeFlow-connector wanneer deze beschikbaar is.

Lakeflow-pijplijnen breiden Structured Streaming uit met automatisch schalen, controles van gegevenskwaliteit, verwerking van schemaontwikkeling en bewaking via het gebeurtenislogboek. Databricks raadt Lakeflow-pipelines aan voor de meeste productie-ingestieworkloads.

Het juiste plannings- en triggertype kiezen

Het beste plannings- en triggertype voor uw use-case is afhankelijk van uw latentievereisten en patronen voor bestands aankomst. Voor de meeste gebruiksscenario's raadt Databricks een trigger voor bestands aankomst aan met bestandsevenementen ingeschakeld. Dit zorgt voor opname met lage latentie tegen lage kosten, omdat berekening alleen wordt uitgevoerd wanneer nieuwe bestanden binnenkomen. De drie triggertypen verschillen in wanneer en hoe vaak de pijplijn wordt gestart:

• Doorlopend: De pijplijn draait zonder te stoppen. Gebruik alleen wanneer latentie van sub-seconde een harde vereiste is, omdat doorlopende rekenkosten meer kosten. Koppelen aan bestandsevenementen.
• Trigger voor aankomst van bestand: de pijplijn wordt gestart wanneer nieuwe bestanden op de bronlocatie terechtkomen. Het meest geschikt voor lage tot gemiddelde latentie of onregelmatige bestands aankomstpatronen. Vereist dat bestandsevenementen zijn ingeschakeld. Zie activeringsopdrachten wanneer nieuwe bestanden aankomen.
• Gepland: De pijplijn wordt uitgevoerd volgens een schema op basis van tijd (bijvoorbeeld elk uur). Gebruik deze functie wanneer latentievereisten soepel zijn (minuten tot uren). Werkt met directoryvermeldingen, maar bestandsgebeurtenissen verlagen de kosten zelfs in de geplande modus door volledige scans van de directory te voorkomen.

Zie Trigger.AvailableNow voor meer informatie over het gebruik van voor batchplanning.

De juiste bestandsdetectiemodus kiezen

Auto Loader ondersteunt drie bestandsdetectiemodi met verschillende afwegingen in de complexiteit, schaalbaarheid en kosten van de installatie.

Bestandsevenementen consolideren resources voor cloudopslag door één abonnement en één wachtrij per externe locatie te gebruiken in plaats van één per gegevensstroom. Het prestatieverschil is op grote schaal aanzienlijk: het weergeven van de mapinhoud moet bij elke activering de volledige bronmap scannen, waardoor de inleestijd toeneemt naarmate de map groter is. Bestandsgebeurtenissen leveren rechtstreeks nieuwe bestandsmeldingen, dus de opnametijd blijft laag, ongeacht het aantal objecten in de map.

Bestandsevenementen inschakelen

Bestandsgebeurtenissen vereisen een eenmalige cloudmachtiging en een externe locatie die is geconfigureerd voor het gebruik van de service voor beheerde bestandsgebeurtenissen. Zodra de installatie is uitgevoerd, kunnen alle automatisch laadprogramma's die vanaf die externe locatie lezen bestandsevenementen gebruiken zonder extra configuratie.

1. Verdeel de vereiste cloudmachtigingen aan de cloudproviderzijde. De vereisten variëren per cloudprovider. Zie Bestandsevenementen instellen voor een externe locatie.

2. Stel cloudFiles.useManagedFileEvents in op true in je Auto Loader-query.

   df = (spark.readStream
     .format("cloudFiles")
     .option("cloudFiles.format", "json")
     .option("cloudFiles.useManagedFileEvents", "true")
     .load("/path/to/data/dir"))
   

   Zie Migreren naar Automatisch laden met bestandsevenementen voor volledige installatiestappen.

Wanneer u geen bestandsevenementen kunt gebruiken

Mogelijk kunt u geen bestandsevenementen gebruiken wanneer:

• De externe locatie is niet geconfigureerd met bestandsgebeurtenissen.
• Beveiligingsbeleid voor organisaties staat het inschakelen van bestandsevenementen op een gedeelde externe locatie niet toe.

In deze gevallen gebruikt u de klassieke modus voor bestandsmeldingen of de modus voor het weergeven van mappen. Zie De modi voor bestandsdetectie vergelijken voor een volledige vergelijking van de bestandsdetectiemodi voor automatisch laden.

Ontwikkeling van schema's beheren

Automatisch laden zorgt ervoor dat het schema automatisch wordt afgeleid, maar hoe u de evolutie van het schema configureert, is van invloed op de volledigheid en de stabiliteit van de pijplijn. Gebruik de volgende tabel om een strategie te kiezen.

Nadat u een strategie hebt gekozen, past u de volgende procedures toe om de werking van schemaontwikkeling af te stemmen.

Schemahints gebruiken voor bekende veldtypen

Gebruik de cloudFiles.schemaHints optie om typen af te dwingen voor velden die u van tevoren kent, terwijl u schemadeductie voor andere velden nog steeds toestaat.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaHints", "id long, amount double")
  .load("/path/to/data/dir"))

Typebreiding gebruiken voor compatibele typewijzigingen

De addNewColumnsWithTypeWidening modus schemaontwikkeling verbreedt automatisch compatibele typen (bijvoorbeeld int naar long) in plaats van gegevens naar de _rescued_data kolom te routeren. Dit voorkomt dat nabewerkingstaken eenvoudige typepromoties hoeven af te handelen.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  .option("cloudFiles.schemaEvolutionMode", "addNewColumnsWithTypeWidening")
  .load("/path/to/data/dir"))

Opnemen als Variant type voor onvoorspelbare schema's

Wanneer uw gegevens niet voldoen aan een specifiek schema of het schema continu wordt gewijzigd, neemt u de gegevens op als een Variant type.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("singleVariantColumn", "data")
  .load("/path/to/data/dir"))

Variant biedt schema-on-read tijdens het uitvoeren van query's, maar is minder efficiënt dan query's op gestructureerde kolommen. Zie Schemadeductie en evolutie configureren in AutoLoader voor de volledige mechanica van schemadeductie en evolutie.

Slechte gegevens en gegevenskwaliteit verwerken

Met de volgende procedures kunt u slechte gegevens detecteren, vastleggen en isoleren voordat deze worden doorgegeven aan downstreamlagen.

Inschakelen _rescued_data en _corrupt_record

Auto Loader biedt twee kolommen voor het vastleggen van gegevens die niet op schone wijze kunnen worden geparseerd.

• _rescued_data legt velden vast die niet overeenkomen met het huidige schema. Het wordt automatisch toegevoegd door Auto Loader.
• _corrupt_record legt rijen vast die helemaal niet kunnen worden geparseerd. Schakel het in met behulp van columnNameOfCorruptRecord:

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaHints", "_corrupt_record string")
  .option("columnNameOfCorruptRecord", "_corrupt_record")
  .load("/path/to/data/dir"))

Databricks raadt columnNameOfCorruptRecord aan boven badRecordsPath om mogelijke racecondities te voorkomen waardoor beschadigde records mogelijk niet worden gedetecteerd.

Verwachtingen van Lakeflow-pijplijnen gebruiken voor bewaking

Stel de verwachtingen van Lakeflow-pijplijnen in om te controleren of _rescued_data en _corrupt_record zich NULL onder normale omstandigheden bevinden. Niet-NULL-waarden geven schemadrift of beschadiging van gegevens aan.

import dlt

@dlt.table
@dlt.expect("no rescued data", "_rescued_data IS NULL")
@dlt.expect("no corrupt records", "_corrupt_record IS NULL")
def bronze_table():
    return (spark.readStream
        .format("cloudFiles")
        .option("cloudFiles.format", "json")
        .option("cloudFiles.schemaHints", "_corrupt_record string")
        .option("columnNameOfCorruptRecord", "_corrupt_record")
        .load("/path/to/data/dir"))

Beschadigde gegevens isoleren

Isoleer rijen met gegevens die niet kunnen worden geparseerd in een aparte bestemming voor onderzoek. Hiermee voorkomt u dat beschadigde gegevens worden doorgegeven aan downstreamlagen.

import dlt

@dlt.table
def corrupt_records_sink():
    return dlt.read_stream("bronze_table").where("_corrupt_record IS NOT NULL")

@dlt.view
def clean_table():
    return dlt.read_stream("bronze_table").where("_corrupt_record IS NULL")

Aantekeningen toevoegen aan gegevens met metagegevens van bronbestand

Neem de _metadata kolom op in uw query's voor automatisch laden. Leg minimaal file_path en file_modification_time vast. Hiermee kunt u gegevensproblemen herleiden tot specifieke bronbestanden en deze koppelen aan cloud_files_state() voor de volledige bestandslevenscyclus.

df = (spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .load("/path/to/data/dir")
  .select("*", "_metadata.file_path", "_metadata.file_modification_time"))

Zie de kolom Bestandsmetagegevens voor meer informatie.

Kosten en prestaties optimaliseren

De volgende werkwijzen verminderen de drie belangrijkste kostenfactoren voor Auto Loader: cloud LIST API-aanroepen, ongebruikte rekenkracht en toename van langetermijnopslag.

• Bestandsevenementen gebruiken om het minimum te beperken LIST API-kosten: bestandsevenementen bieden incrementele bestandsdetectie, waardoor er geen volledige mapvermeldingen meer nodig zijn voor elke uitvoering. Dit is de meest impactvolle kostenoptimalisatie voor Auto Loader.

• Gebruik triggers voor bestands aankomst voor gebeurtenisgestuurde verwerking: met triggers voor bestandsaankomst wordt uw pijplijn alleen gestart wanneer nieuwe bestanden binnenkomen, zodat u niet betaalt voor niet-actieve berekening. Zie activeringsopdrachten wanneer nieuwe bestanden aankomen.

• Verwerkte bestanden archiveren met cloudFiles.cleanSource: hiermee cloudFiles.cleanSource kunt u verwerkte bestanden automatisch verwijderen of verplaatsen. Dit vermindert zowel de opslagkosten als de kosten voor het opvragen van directoryvermeldingen voor langlopende streams. Zie Archiveringsbestanden in de bronmap voor meer informatie om de kosten te verlagen.

  • Gebruik de delete-modus om bestanden na het importeren te verwijderen.
  • Gebruik move de modus voor het archiveren van bestanden op een andere locatie voor naleving of controle.

  df = (spark.readStream
    .format("cloudFiles")
    .option("cloudFiles.format", "json")
    .option("cloudFiles.cleanSource", "delete")
    .load("/path/to/data/dir"))
  

  Warning

  Schakel cloudFiles.cleanSource niet in als meerdere Auto Loader-streams of andere clients uit dezelfde bronmap lezen.

• Profiteer van prestatieverbeteringen: voer een upgrade uit naar de nieuwste Databricks Runtime of gebruik serverloze rekenkracht om te profiteren van recente prestatieverbeteringen in het autolaadprogramma.

Controlepuntbeheer

Het controlepunt slaat de voortgang en bestandsstatus van de stream op. Voor het onjuist configureren of verliezen van het controlepunt is een volledige herstart vereist, dus behandel het als kritieke infrastructuur.

• Pas nooit levenscyclusbeleid voor cloudobjecten toe op controlepuntlocaties. Als controlepuntbestanden worden verwijderd, is de stroomstatus beschadigd en moet u helemaal opnieuw opstarten.
• Gebruik afzonderlijke controlepunten voor elke gegevensstroom en bronmap.
• Overweeg cloudFiles.maxFileAge voor langdurige stromen met een hoog volume om de groei van de toestand te beperken. Gebruik een conservatieve instelling (minimaal 90 dagen aanbevolen). Als u deze waarde te agressief instelt, loopt u een risico op het opnieuw verwerken van bestanden die automatisch laadprogramma al heeft opgenomen als ze buiten het venster vallen.

Zie Het bijhouden van bestandsevenementen voor meer informatie.

Volumes gebruiken voor optimale bestandsdetectie met bestandsevenementen

Voor betere prestaties met bestandsgebeurtenissen maakt u voor elk pad of elke submap waaruit Auto Loader laadt een extern opslagvolume. Geef volumepaden (bijvoorbeeld /Volumes/catalog/schema/volume) op voor Auto Loader in plaats van cloudpaden (bijvoorbeeld s3://bucket/path). Hiermee optimaliseert u bestandsdetectie via een geoptimaliseerd patroon voor gegevenstoegang.

Zie Aanbevolen werkwijzen voor Auto Loader met bestandsgebeurtenissen voor meer aanbevolen werkwijzen voor bestandsgebeurtenissen.