Běžné vzory načítání dat

Automatický zavaděč zjednodušuje řadu běžných úloh příjmu dat. Tato rychlá referenční příručka obsahuje příklady pro několik oblíbených vzorů.

Příjem dat z cloudového úložiště objektů ve formě varianty

Automatický zavaděč může načíst všechna data z podporovaných zdrojů souborů jako jeden VARIANT sloupec v cílové tabulce. Vzhledem k tomu, že VARIANT je flexibilní vůči změnám schématu a typu, zachovává citlivost na velikost písmen a hodnoty obsažené ve zdroji dat, je tento model robustní vůči většině scénářů importu dat. Podrobnosti najdete na téma Import dat z cloudového úložiště objektů ve variantě.

Filtrování adresářů nebo souborů pomocí vzorů globu

Vzory globu lze použít pro filtrování adresářů a souborů, pokud jsou specifikovány v cestě.

Vzor	Popis
`?`	Odpovídá jakémukoli jednomu znaku.
`*`	Shoduje se s nulou nebo více znaky
`[abc]`	Odpovídá jednomu znaku ze znakové sady {a,b,c}.
`[a-z]`	Odpovídá jednomu znaku z rozsahu znaků {a... z}.
`[^a]`	Odpovídá jednomu znaku, který není ze znakové sady nebo rozsahu {a}. Všimněte si, že `^` znak musí nastat okamžitě napravo od levé závorky.
`{ab,cd}`	Odpovídá řetězci ze sady řetězců {ab, cd}.
`{ab,c{de, fh}}`	Odpovídá řetězci ze sady řetězců {ab, cde, cfh}.

path Použijte k poskytování vzorů předpon, například:

Python

df = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", <format>) \
  .schema(schema) \
  .load("/Volumes/catalog_name/schema_name/volume_name/*/files")

Scala

val df = spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", <format>)
  .schema(schema)
  .load("/Volumes/catalog_name/schema_name/volume_name/*/files")

K explicitnímu poskytnutí vzorů přípon musíte použít možnost pathGlobFilter . path poskytuje pouze filtr předpony. Pokud například chcete analyzovat pouze png soubory v adresáři, který obsahuje soubory s různými příponami, můžete udělat toto:

Python

df = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "binaryFile") \
  .option("pathGlobfilter", "*.png") \
  .load("/Volumes/catalog_name/schema_name/volume_name/path")

Scala

val df = spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "binaryFile")
  .option("pathGlobfilter", "*.png")
  .load("/Volumes/catalog_name/schema_name/volume_name/path")

Poznámka:

Výchozí chování globbing Auto Loaderu se liší od výchozího chování jiných zdrojů souborů Spark. Přidejte .option("cloudFiles.useStrictGlobber", "true") do čtení, abyste použili globbing, který odpovídá výchozímu chování Sparku vůči zdrojům souborů. Další informace o globbingu najdete v následující tabulce:

Vzor	Cesta k souboru	Výchozí globber	Striktní globber
/a/b	/a/b/c/file.txt	Ano	Ano
/a/b	/a/b_dir/c/file.txt	Ne	Ne
/a/b	/a/b.txt	Ne	Ne
/a/b/	/a/b.txt	Ne	Ne
/a/*/c/	/a/b/c/file.txt	Ano	Ano
/a/*/c/	/a/b/c/d/file.txt	Ano	Ano
/a/*/c/	/a/b/x/y/c/file.txt	Ano	Ne
/a/*/c	/a/b/c_file.txt	Ano	Ne
/a/*/c/	/a/b/c_file.txt	Ano	Ne
/a/*/c/	/a/*/cookie/file.txt	Ano	Ne
/a/b*	/a/b.txt	Ano	Ano
/a/b*	/a/b/file.txt	Ano	Ano
/a/{0.txt,1.txt}	/a/0.txt	Ano	Ano
/a/*/{0.txt,1.txt}	/a/0.txt	Ne	Ne
/a/b/[cde-h]/i/	/a/b/c/i/file.txt	Ano	Ano

Povolení snadného ETL

Jednoduchý způsob, jak dostat data do Delta Lake bez ztráty dat, je použít následující vzor a povolit odvozování schématu pomocí Auto Loaderu. Databricks doporučuje spustit následující kód v úloze Azure Databricks, aby automaticky restartoval datový proud, když se změní schéma zdrojových dat. Ve výchozím nastavení je schéma odvozeno jako typy řetězců, všechny chyby analýzy (pokud vše zůstane jako řetězec) budou směřovat do _rescued_dataa všechny nové sloupce způsobí selhání datového proudu a vývoj schématu.

Python

spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "json") \
  .option("cloudFiles.schemaLocation", "<path-to-schema-location>") \
  .load("/Volumes/catalog_name/schema_name/volume_name/source_data") \
  .writeStream \
  .option("mergeSchema", "true") \
  .option("checkpointLocation", "<path-to-checkpoint>") \
  .start("<path_to_target>")

Scala

spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", "<path-to-schema-location>")
  .load("/Volumes/catalog_name/schema_name/volume_name/source_data")
  .writeStream
  .option("mergeSchema", "true")
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path_to_target>")

Prevence ztráty dat ve dobře strukturovaných datech

Pokud znáte schéma, ale chcete zachytit neočekávaná data, databricks doporučuje použít rescuedDataColumn.

Python

spark.readStream.format("cloudFiles") \
  .schema(expected_schema) \
  .option("cloudFiles.format", "json") \
  # will collect all new fields as well as data type mismatches in _rescued_data
  .option("cloudFiles.schemaEvolutionMode", "rescue") \
  .load("/Volumes/catalog_name/schema_name/volume_name/source_data") \
  .writeStream \
  .option("checkpointLocation", "<path-to-checkpoint>") \
  .start("<path_to_target>")

Scala

spark.readStream.format("cloudFiles")
  .schema(expected_schema)
  .option("cloudFiles.format", "json")
  // will collect all new fields as well as data type mismatches in _rescued_data
  .option("cloudFiles.schemaEvolutionMode", "rescue")
  .load("/Volumes/catalog_name/schema_name/volume_name/source_data")
  .writeStream
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path_to_target>")

Pokud chcete, aby stream přestal zpracovávat, pokud se zavádí nové pole, které neodpovídá vašemu schématu, můžete přidat:

.option("cloudFiles.schemaEvolutionMode", "failOnNewColumns")

Umožněte flexibilní částečně strukturované datové procesy

Když dostáváte data od dodavatele, který do informací, které poskytují, zavádí nové sloupce, nemusíte vědět přesně, kdy to dělají, nebo možná nebudete mít šířku pásma pro aktualizaci datového kanálu. Nyní můžete využít vývoj schématu k restartování streamu a nechat Auto Loader aktualizovat odvozené schéma. Můžete také využít schemaHints pro některá z polí "bez schématu", která může dodavatel poskytnout.

Python

spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "json") \
  # will ensure that the headers column gets processed as a map
  .option("cloudFiles.schemaHints",
          "headers map<string,string>, statusCode SHORT") \
  .load("/Volumes/catalog_name/schema_name/volume_name/api/requests") \
  .writeStream \
  .option("mergeSchema", "true") \
  .option("checkpointLocation", "<path-to-checkpoint>") \
  .start("<path_to_target>")

Scala

spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  // will ensure that the headers column gets processed as a map
  .option("cloudFiles.schemaHints",
          "headers map<string,string>, statusCode SHORT")
  .load("/Volumes/catalog_name/schema_name/volume_name/api/requests")
  .writeStream
  .option("mergeSchema", "true")
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path_to_target>")

Transformace vnořených dat JSON

Jelikož Auto Loader interpretuje sloupce JSON nejvyšší úrovně jako řetězce, mohou vám zůstat vnořené objekty JSON, které vyžadují další transformace. Pomocí částečně strukturovaných rozhraní API pro přístup k datům můžete dále transformovat složitý obsah JSON.

Python

spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "json") \
  # The schema location directory keeps track of your data schema over time
  .option("cloudFiles.schemaLocation", "<path-to-checkpoint>") \
  .load("/Volumes/catalog_name/schema_name/volume_name/nested_json") \
  .selectExpr(
    "*",
    "tags:page.name",    # extracts {"tags":{"page":{"name":...}}}
    "tags:page.id::int", # extracts {"tags":{"page":{"id":...}}} and casts to int
    "tags:eventType"     # extracts {"tags":{"eventType":...}}
  )

Scala

spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  // The schema location directory keeps track of your data schema over time
  .option("cloudFiles.schemaLocation", "<path-to-checkpoint>")
  .load("/Volumes/catalog_name/schema_name/volume_name/nested_json")
  .selectExpr(
    "*",
    "tags:page.name",     // extracts {"tags":{"page":{"name":...}}}
    "tags:page.id::int",  // extracts {"tags":{"page":{"id":...}}} and casts to int
    "tags:eventType"      // extracts {"tags":{"eventType":...}}
  )

Odvození vnořených dat JSON

Pokud máte vnořená data, můžete použít možnost cloudFiles.inferColumnTypes k odvození vnořené struktury dat a dalších typů sloupců.

Python

spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "json") \
  # The schema location directory keeps track of your data schema over time
  .option("cloudFiles.schemaLocation", "<path-to-checkpoint>") \
  .option("cloudFiles.inferColumnTypes", "true") \
  .load("/Volumes/catalog_name/schema_name/volume_name/nested_json")

Scala

spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  // The schema location directory keeps track of your data schema over time
  .option("cloudFiles.schemaLocation", "<path-to-checkpoint>")
  .option("cloudFiles.inferColumnTypes", "true")
  .load("/Volumes/catalog_name/schema_name/volume_name/nested_json")

Načtení souborů CSV bez hlaviček

Následující příklad ukazuje, jak načíst soubory CSV bez hlaviček pomocí Auto Loader. Slouží rescuedDataColumn k zachycení všech dat, která neodpovídají zadanému schématu.

Python

df = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .option("rescuedDataColumn", "_rescued_data") \ # ensure that you don't lose data
  .schema(<schema>) \ # provide a schema here for the files
  .load(<path>)

Scala

val df = spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .option("rescuedDataColumn", "_rescued_data") // makes sure that you don't lose data
  .schema(<schema>) // provide a schema here for the files
  .load(<path>)

Vynucení schématu u souborů CSV s hlavičkami

Následující příklad ukazuje, jak vynutit schéma u souborů CSV, které obsahují hlavičky. Slouží rescuedDataColumn k zachycení všech dat, která neodpovídají zadanému schématu.

Python

df = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .option("header", "true") \
  .option("rescuedDataColumn", "_rescued_data") \ # makes sure that you don't lose data
  .schema(<schema>) \ # provide a schema here for the files
  .load(<path>)

Scala

val df = spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .option("header", "true")
  .option("rescuedDataColumn", "_rescued_data") // makes sure that you don't lose data
  .schema(<schema>) // provide a schema here for the files
  .load(<path>)

Ingestování obrázků nebo binárních dat do Delta Lake pro ML

Jakmile jsou data uložená v Delta Lake, můžete na datech spustit distribuované odvozování. Viz Provádění distribuovaných odvozování pomocí funkce UDF pandas.

Python

spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "binaryFile") \
  .load("/Volumes/catalog_name/schema_name/volume_name/images") \
  .writeStream \
  .option("checkpointLocation", "<path-to-checkpoint>") \
  .start("<path_to_target>")

Scala

spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "binaryFile")
  .load("/Volumes/catalog_name/schema_name/volume_name/images")
  .writeStream
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path_to_target>")

Syntaxe Auto Loaderu pro pipeline Lakeflow

Pipeline Lakeflow poskytují mírně upravenou syntaxi jazyka Python pro Auto Loader a přidávají podporu SQL pro Auto Loader. Následující příklady používají Auto Loader k vytvoření datových sad ze souborů JSON pomocí ukázkové cestovní rezervační datové sady Wanderbricks.

Python

@dp.table
def booking_updates():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .option("multiLine", "true")
      .load("/Volumes/my_catalog/my_schema/my_volume/wanderbricks/booking_updates")
  )

@dp.table
def reviews():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .option("multiLine", "true")
      .load("/Volumes/my_catalog/my_schema/my_volume/wanderbricks/reviews")
  )

SQL

CREATE OR REFRESH STREAMING TABLE booking_updates
AS SELECT * FROM STREAM read_files(
  "/Volumes/my_catalog/my_schema/my_volume/wanderbricks/booking_updates",
  format => "json",
  multiLine => true
)

CREATE OR REFRESH STREAMING TABLE reviews
AS SELECT * FROM STREAM read_files(
  "/Volumes/my_catalog/my_schema/my_volume/wanderbricks/reviews",
  format => "json",
  multiLine => true
)

Pro Auto Loader můžete použít podporované možnosti formátu. Možnosti read_files jsou páry klíč-hodnota. Podrobnosti o podporovaných formátech a možnostech najdete v tématu Možnosti.

CREATE OR REFRESH STREAMING TABLE my_table
AS SELECT *
  FROM STREAM read_files(
    "/Volumes/my_volume/path/to/files/*",
    option-key => option-value,
    ...
  )

Následující příklad čte víceřádkové soubory JSON s povoleným odvození typu sloupce:

CREATE OR REFRESH STREAMING TABLE booking_updates
AS SELECT * FROM STREAM read_files(
  "/Volumes/my_catalog/my_schema/my_volume/wanderbricks/booking_updates",
  format => "json",
  multiLine => true,
  inferColumnTypes => true
)

K ručnímu zadání formátu můžete použít schema; Je nutné zadat schema pro formáty, které nepodporují odvození schématu:

Python

@dp.table
def booking_updates_raw():
  return (
    spark.readStream.format("cloudFiles")
      .schema("booking_id LONG, booking_update_id LONG, user_id LONG, property_id LONG, status STRING, guests_count INT, total_amount DOUBLE, check_in DATE, check_out DATE, created_at TIMESTAMP, updated_at TIMESTAMP")
      .option("cloudFiles.format", "json")
      .option("multiLine", "true")
      .load("/Volumes/my_catalog/my_schema/my_volume/wanderbricks/booking_updates")
  )

SQL

CREATE OR REFRESH STREAMING TABLE booking_updates_raw
AS SELECT *
FROM STREAM read_files(
  "/Volumes/my_catalog/my_schema/my_volume/wanderbricks/booking_updates",
  format => "json",
  multiLine => true,
  schema => "booking_id LONG, booking_update_id LONG, user_id LONG, property_id LONG, status STRING, guests_count INT, total_amount DOUBLE, check_in DATE, check_out DATE, created_at TIMESTAMP, updated_at TIMESTAMP"
)

Poznámka:

Lakeflow Pipelines při použití Auto Loaderu ke čtení souborů automaticky konfigurují a spravují schéma a adresáře kontrolních bodů. Pokud však ručně nakonfigurujete některý z těchto adresářů, provedení úplné aktualizace nemá vliv na obsah nakonfigurovaných adresářů. Databricks doporučuje používat automaticky nakonfigurované adresáře, aby se zabránilo neočekávaným vedlejším účinkům během zpracování.

Další zdroje informací

Váš názor

Byla tato stránka užitečná?

Last updated on 2026-07-10

Běžné vzory načítání dat

Příjem dat z cloudového úložiště objektů ve formě varianty

Filtrování adresářů nebo souborů pomocí vzorů globu

Python

Scala

Python

Scala

Povolení snadného ETL

Python

Scala

Prevence ztráty dat ve dobře strukturovaných datech

Python

Scala

Umožněte flexibilní částečně strukturované datové procesy

Python

Scala

Transformace vnořených dat JSON

Python

Scala

Odvození vnořených dat JSON

Python

Scala

Načtení souborů CSV bez hlaviček

Python

Scala

Vynucení schématu u souborů CSV s hlavičkami

Python

Scala

Ingestování obrázků nebo binárních dat do Delta Lake pro ML

Python

Scala

Syntaxe Auto Loaderu pro pipeline Lakeflow

Python

SQL

Python

SQL

Další zdroje informací

Váš názor

Další materiály