Gegevens opnemen met Auto Loader

Wanneer nieuwe gegevensbestanden continu binnenkomen in cloudopslag, hebt u een efficiënte manier nodig om ze te verwerken zonder handmatig bij te houden welke bestanden zijn opgenomen. Auto Loader lost deze uitdaging op door automatisch nieuwe bestanden te detecteren en op te nemen zodra ze verschijnen, waarbij u exacte eenmalige garanties krijgt zonder dat u de status of controlepunten handmatig hoeft te beheren.

Begrijpen hoe automatisch laadprogramma werkt

Auto Loader is een structured streaming-bron die een opslaglocatie in de cloud bewaakt en incrementeel nieuwe bestanden verwerkt. In plaats van hele mappen herhaaldelijk te scannen, behoudt Auto Loader de status van de bestanden die het al heeft verwerkt, zodat elk bestand precies eenmaal wordt opgenomen.

Wanneer u een automatische laadstroom start, kan deze bestaande bestanden in de map verwerken en vervolgens continu op nieuwe aankomsten letten. In de stream worden voortgangsgegevens opgeslagen op een controlepuntlocatie, zodat deze kan worden hervat vanaf de locatie waar deze is gestopt als deze wordt onderbroken.

Auto Loader detecteert nieuwe bestanden met behulp van een van de twee modi:

• Modus voor adreslijstvermelding: Automatisch laden geeft periodiek een lijst van de invoermap voor het detecteren van nieuwe bestanden. Voor deze aanpak is geen aanvullende configuratie vereist buiten de toegang tot de opslag.
• Bestandsmeldingsmodus: Automatisch laden maakt gebruik van cloudmeldingsservices om gebeurtenissen te ontvangen wanneer nieuwe bestanden binnenkomen. Deze modus is efficiënter voor grootschalige workloads, omdat hiermee herhaalde adreslijstvermeldingen worden vermeden.

Voor de meeste productieworkloads raadt Databricks de modus voor bestandsmeldingen aan met bestandsevenementen ingeschakeld op uw externe locatie in Unity Catalog. Met bestandsevenementen ontvangt Auto Loader meldingen rechtstreeks wanneer bestanden binnenkomen, wat de latentie en cloud-API-kosten vermindert.

Automatisch laadprogramma configureren voor bestandsopname

Als u gegevens wilt opnemen met automatisch laden, gebruikt u de cloudFiles indeling met spark.readStream. In het volgende voorbeeld worden JSON-bestanden uit Azure Data Lake Storage gelezen en naar een Unity Catalog-tabel geschreven:

base_path = "abfss://container@storage.dfs.core.windows.net/autoloader/orders"
schema_path = f"{base_path}/schema"
checkpoint_path = f"{base_path}/checkpoint"

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", schema_path)
  .load("abfss://container@storage.dfs.core.windows.net/incoming/orders/")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable("sales.bronze.orders"))

Met cloudFiles.format de optie geeft u de bronbestandsindeling op. Auto Loader ondersteunt JSON-, CSV-, Parquet-, Avro-, ORC-, XML-, tekst- en binaire bestanden. Het cloudFiles.schemaLocation slaat het afgeleide schema op, waarmee Auto Loader schemawijzigingen in de loop van de tijd kan bijhouden. Hoewel de schemalocatie en controlepuntlocatie hetzelfde kunnen zijn, kunt u een stroom resetten zonder dat het afgeleide schema verloren gaat.

De trigger(availableNow=True) instelling verwerkt alle momenteel beschikbare bestanden en stopt vervolgens. Dit patroon werkt goed voor geplande batchtaken. Voor continue verwerking kunt u de trigger weglaten of gebruiken trigger(processingTime="1 minute") om bestanden in intervallen te verwerken.

SQL-syntaxis gebruiken met read_files

Auto Loader biedt ook ondersteuning voor SQL-syntaxis via de read_files tabelwaardefunctie. Deze methode is handig wanneer u liever SQL gebruikt of wanneer u in SQL-notebooks werkt:

SELECT * FROM read_files(
  'abfss://container@storage.dfs.core.windows.net/incoming/orders/',
  format => 'json',
  schemaHints => 'order_id INT, amount DECIMAL(10,2)'
)

Bij gebruik van read_files met streamingtabellen in Lakeflow-pijplijnen wordt de Auto Loader-functionaliteit automatisch ingeschakeld. Gebruik het STREAM trefwoord om incrementele verwerking in te schakelen:

CREATE OR REFRESH STREAMING TABLE bronze_orders
AS SELECT * FROM STREAM read_files(
  'abfss://container@storage.dfs.core.windows.net/incoming/orders/',
  format => 'json'
)

In deze context beheert de pijplijn automatisch controlepunten en schemaontwikkeling.

Schemadeductie en evolutie verwerken

Automatisch laden kan automatisch het schema van uw bronbestanden detecteren, waardoor schema's niet handmatig hoeven te worden gedefinieerd. Wanneer schema-inferentie is ingeschakeld, wordt Auto Loader gebruikt om bestanden te bemonsteren en kolomnamen en gegevenstypen vast te stellen.

Voor JSON- en CSV-bestanden worden standaard alle kolommen als tekenreeksen afgeleid. Deze conservatieve benadering voorkomt niet-overeenkomende typen die gegevensverlies kunnen veroorzaken. Als u automatische typedetectie wilt inschakelen, stelt u de cloudFiles.inferColumnTypes optie in:

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.inferColumnTypes", "true")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load(source_path))

Wanneer uw brongegevens nieuwe kolommen introduceert, detecteert Auto Loader de wijziging en kan het doelschema automatisch worden ontwikkeld. De cloudFiles.schemaEvolutionMode optie bepaalt dit gedrag:

De opgeslagen gegevenskolom (_rescued_data) legt alle gegevens vast die niet overeenkomen met het verwachte schema, inclusief typefouten en onverwachte kolommen. Deze functie voorkomt gegevensverlies wanneer brongegevens niet voldoen aan de verwachtingen. Auto Loader voegt deze kolom automatisch toe bij het afleiden van het schema.

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .option("cloudFiles.schemaEvolutionMode", "rescue")
  .schema(expected_schema)
  .load(source_path))

Als u de naam van de kolom met geredde gegevens wilt wijzigen, gebruikt u de rescuedDataColumn volgende optie:

.option("rescuedDataColumn", "my_rescued_data")

Als u weet dat bepaalde kolommen specifieke typen moeten hebben, gebruikt u schemahints om de uitgestelde typen te overschrijven:

.option("cloudFiles.schemaHints", "order_date DATE, amount DECIMAL(10,2)")

Auto Loader-streams bewaken en optimaliseren

Voor productieworkloads helpt bewaking u bij het bijhouden van de opnamevoortgang en het identificeren van problemen. Auto Loader biedt een SQL-functie om een query uit te voeren op de status van gedetecteerde bestanden:

SELECT * FROM cloud_files_state('/path/to/checkpoint');

Deze query retourneert metagegevens over bestanden die automatisch laden heeft gedetecteerd, inclusief de verwerkingsstatus. U kunt deze informatie gebruiken om te controleren of bestanden worden opgenomen zoals verwacht.

Auto Loader rapporteert metrische gegevens via het dashboard voor voortgang van structured streaming, waaronder:

• numFilesOutstanding: Gedetecteerde bestanden maar nog niet verwerkt
• numBytesOutstanding: Totaal aantal bytes dat moet worden verwerkt

Voor workloads met grote volumes bepaalt u de verwerkingssnelheid met behulp van cloudFiles.maxFilesPerTrigger of cloudFiles.maxBytesPerTrigger. Met deze instellingen voorkomt u dat één microbatch te groot wordt:

.option("cloudFiles.maxFilesPerTrigger", "1000")
.option("cloudFiles.maxBytesPerTrigger", "1g")

Als u toegang wilt krijgen tot metagegevens op bestandsniveau in uw opgenomen gegevens, selecteert u de _metadata kolom. Deze verborgen kolom bevat informatie over elk bronbestand, inclusief file_path, file_nameen file_sizefile_modification_time:

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  .load(source_path)
  .select("*", "_metadata.file_path", "_metadata.file_modification_time"))

Auto Loader kan naadloos worden geïntegreerd met Lakeflow-pijplijnen, waar het automatisch checkpointing en schemabeheer afhandelt. In de volgende les leert u hoe u automatisch laden kunt gebruiken in declaratieve pijplijndefinities voor end-to-end gegevensopnamewerkstromen.