Mata in data med Auto Loader

När nya datafiler tas emot kontinuerligt i molnlagring behöver du ett effektivt sätt att bearbeta dem utan att manuellt spåra vilka filer som har matats in. Automatisk inläsning löser den här utmaningen genom att automatiskt identifiera och mata in nya filer som de visas, vilket ger garantier exakt en gång utan att du behöver hantera tillstånd eller kontrollpunkter manuellt.

Förstå hur Auto Loader fungerar

Auto Loader är en strukturströmningskälla som övervakar en molnlagringsplats och bearbetar nya filer allt eftersom de tillkommer. I stället för att genomsöka hela kataloger upprepade gånger behåller Auto Loader tillståndet för vilka filer som den redan har bearbetat, vilket säkerställer att varje fil matas in exakt en gång.

När du startar en automatisk inläsningsström kan den bearbeta befintliga filer i katalogen och sedan kontinuerligt titta efter nya ankomster. Strömmen lagrar förloppsinformation på en kontrollpunktsplats, vilket gör att den kan återupptas exakt där den stoppades om den avbröts.

Automatisk inläsning identifierar nya filer med något av två lägen:

• Kataloglistningsläge: Auto Loader listar regelbundet indatakatalogen för att identifiera nya filer. Den här metoden kräver ingen ytterligare konfiguration utöver lagringsåtkomst.
• Läge för filavisering: Auto Loader använder molnmeddelandetjänster för att ta emot händelser när nya filer anländer. Det här läget är effektivare för storskaliga arbetsbelastningar eftersom det undviker upprepade kataloglistor.

För de flesta produktionsarbetsbelastningar rekommenderar Databricks filmeddelandeläge med filhändelser aktiverade på din externa plats i Unity Catalog. Med filhändelser tar Auto Loader emot meddelanden direkt när filer tas emot, vilket minskar svarstiden och api-kostnaderna för molnet.

Konfigurera Auto Loader för filläsning

Om du vill mata in data med Auto Loader använder du formatet med cloudFiles och spark.readStream. I följande exempel läss JSON-filer från Azure Data Lake Storage och de skrivs till en Unity Catalog-tabell:

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"))

Alternativet cloudFiles.format anger källfilformatet. Auto Loader stöder JSON, CSV, Parquet, Avro, ORC, XML, text och binära filer. Lagrar cloudFiles.schemaLocation det härledda schemat, vilket gör att Auto Loader kan spåra schemaändringar över tid. Även om schemaplatsen och kontrollpunktsplatsen kan vara desamma, möjliggör det att hålla dem åtskilda att du kan återställa en ström utan att förlora den härledda schemat.

Inställningen trigger(availableNow=True) bearbetar alla tillgängliga filer och stoppas sedan. Det här mönstret fungerar bra för schemalagda batchjobb. För kontinuerlig bearbetning kan du utelämna utlösaren eller använda trigger(processingTime="1 minute") för att bearbeta filer i intervall.

Använd SQL-syntax med read_files

Auto Loader stöder också SQL-syntax via funktionen read_files tabellvärde. Den här metoden är användbar när du föredrar SQL eller när du arbetar i SQL-notebook-filer:

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

När du använder read_files med strömmande tabeller i Lakeflow-pipelines aktiveras Auto Loader-funktioner automatiskt. Använd nyckelordet STREAM för att aktivera inkrementell bearbetning:

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

I det här sammanhanget hanterar pipelinen kontrollpunkter och schemautveckling automatiskt.

Hantera schemainferens och utveckling

Automatisk inläsare kan automatiskt identifiera schemat för dina källfiler, vilket eliminerar behovet av att definiera scheman manuellt. När schemainferens är aktiverat tar Auto Loader exempelfiler för att fastställa kolumnnamn och datatyper.

För JSON- och CSV-filer härleder Auto Loader alla kolumner som strängar som standard. Den här konservativa metoden förhindrar typkonflikter som kan orsaka dataförlust. Om du vill aktivera automatisk typidentifiering anger du alternativet cloudFiles.inferColumnTypes :

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

När dina källdata introducerar nya kolumner identifierar Auto Loader ändringen och kan utveckla målschemat automatiskt. Alternativet cloudFiles.schemaEvolutionMode styr det här beteendet:

Den räddade datakolumnen (_rescued_data) samlar in alla data som inte matchar det förväntade schemat, inklusive typmatchningar och oväntade kolumner. Den här funktionen förhindrar dataförlust när källdata inte överensstämmer med förväntningarna. Automatisk inläsning lägger automatiskt till den här kolumnen när schemat härleds.

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

Om du vill byta namn på den räddade datakolumnen använder du alternativet rescuedDataColumn :

.option("rescuedDataColumn", "my_rescued_data")

Om du vet att vissa kolumner ska ha specifika typer använder du schematips för att åsidosätta de härledda typerna:

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

Övervaka och optimera Auto Loader-strömmar

För produktionsarbetsbelastningar hjälper övervakning dig att spåra förloppet för inmatning och identifiera problem. Auto Loader tillhandahåller en SQL-funktion för att fråga efter tillståndet för identifierade filer:

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

Den här frågan returnerar metadata om filer som Auto Loader har identifierat, inklusive deras bearbetningsstatus. Du kan använda den här informationen för att kontrollera att filer matas in som förväntat.

Auto Loader rapporterar mått via instrumentpanelen för förlopp för strukturerad direktuppspelning, inklusive:

• numFilesOutstanding: Filer som identifierats men ännu inte bearbetats
• numBytesOutstanding: Totalt antal byte som väntar på att bearbetas

För arbetsbelastningar med stora volymer kontrollerar du bearbetningshastigheten med cloudFiles.maxFilesPerTrigger eller cloudFiles.maxBytesPerTrigger. De här inställningarna förhindrar att en enskild mikrobatch blir för stor:

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

Om du vill komma åt metadata på filnivå i dina inmatade data väljer du _metadata kolumnen. Den här dolda kolumnen innehåller information om varje källfil, inklusive file_path, file_name, file_sizeoch file_modification_time:

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

Auto Loader integreras sömlöst med Lakeflow-pipelines, där den hanterar checkpointhantering och schemahantering automatiskt. I nästa enhet får du lära dig hur du använder Auto Loader i deklarativa pipelinedefinitioner för end-to-end datainmatningsarbetsflöden.