Konfigurace odvození schématu a vývoje v automatickém zavaděči

Automatický zavaděč můžete nakonfigurovat tak, aby automaticky rozpoznal schéma načtených dat, což umožňuje inicializovat tabulky bez explicitního deklarování schématu dat a vyvíjet schéma tabulky při zavádění nových sloupců. To eliminuje potřebu ručního sledování a použití změn schématu v průběhu času.

Automatický zavaděč může také "zachránit" data, která byla neočekávaná (například různých datových typů) ve sloupci typu JSON blob, který můžete zobrazit později pomocí částečně strukturovaných rozhraní API pro přístup k datům.

Auto Loader podporuje následující formáty pro odvozování a vývoj schématu:

formát souboru Podporované verze
JSON Všechny verze
CSV Všechny verze
XML Databricks Runtime 14.3 LTS a vyšší
Avro Databricks Runtime 10.4 LTS a vyšší
Parquet Databricks Runtime 11.3 LTS a vyšší
ORC Nepodporované
Text Nepoužitelné (pevné schéma)
Binaryfile Nepoužitelné (pevné schéma)

Syntaxe pro odvozování a vývoj schématu

Určení cílového adresáře pro možnost cloudFiles.schemaLocation umožňuje odvozování a vývoj schématu. Můžete použít stejný adresář, který zadáte pro checkpointLocation. Pokud používáte kanály Lakeflow, Azure Databricks automaticky spravuje umístění schématu a další informace o kontrolních bodech.

Poznámka:

Pokud do cílové tabulky načítáte více než jedno zdrojové umístění dat, každá úloha zpracování Auto Loader vyžaduje samostatný kontrolní bod streamování.

Následující příklad používá parquet pro cloudFiles.format. Použijte csv, avronebo json pro jiné zdroje souborů. Všechna ostatní nastavení pro čtení a zápis zůstávají stejná pro výchozí chování jednotlivých formátů.

Python

(spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  # The schema location directory keeps track of your data schema over time
  .option("cloudFiles.schemaLocation", "<path-to-schema>")
  .load("<path-to-source-data>")
  .writeStream
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path-to-target>")
)

Scala

spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "parquet")
  // The schema location directory keeps track of your data schema over time
  .option("cloudFiles.schemaLocation", "<path-to-schema>")
  .load("<path-to-source-data>")
  .writeStream
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path-to-target>")

Jak funguje automatické odvozování schématu u Auto Loaderu?

Pro odvození schématu při prvním čtení dat auto loader vzorkuje prvních 50 GB nebo 1000 souborů, které zjistí, podle toho, který limit je překročen jako první. Auto Loader ukládá informace o schématu v adresáři _schemas nakonfigurované cloudFiles.schemaLocation tak, aby sledoval změny schématu vstupních dat v průběhu času.

Poznámka:

Pokud chcete změnit velikost použité ukázky, nastavte konfigurace SQL:

spark.databricks.cloudFiles.schemaInference.sampleSize.numBytes

(bajtový řetězec, například 10gb)

a

spark.databricks.cloudFiles.schemaInference.sampleSize.numFiles

(celé číslo)

Ve výchozím nastavení se při odvozování schématu automatického zavaděče snaží vyhnout problémům s vývojem schématu kvůli neshodám typů. U formátů, které nekódují datové typy (JSON, CSV a XML), auto loader odvodí všechny sloupce jako řetězce (včetně vnořených polí v souborech JSON). U formátů se zadaným schématem (Parquet a Avro) auto loader vzorkuje podmnožinu souborů a slučuje schémata jednotlivých souborů. Následující tabulka shrnuje toto chování.

formát souboru Výchozí odvozený datový typ
JSON řetězec
CSV řetězec
XML řetězec
Avro Typy kódované ve schématu Avro
Parquet Typy kódované ve schématu Parquet

Apache Spark DataFrameReader používá pro odvozování schématu různé chování, výběr datových typů pro sloupce ve zdrojích JSON, CSV a XML na základě ukázkových dat. Chcete-li toto chování povolit pomocí Auto Loaderu, nastavte možnost cloudFiles.inferColumnTypes na true.

Poznámka:

Při odvozování schématu pro data ve formátu CSV Auto Loader předpokládá, že soubory obsahují záhlaví. Pokud soubory CSV neobsahují záhlaví, zadejte možnost .option("header", "false"). Kromě toho Auto Loader slučuje schémata všech vzorových souborů, aby vzniklo globální schéma. Auto Loader pak může číst každý soubor podle záhlaví a správně parsovat CSV.

Poznámka:

Pokud sloupec obsahuje různé datové typy ve dvou souborech Parquet, automatický zavaděč vybere nejširší typ. Tuto volbu můžete přepsat pomocí funkce schemaHints . Když zadáte rady schématu, Auto Loader nepřetypuje sloupec na zadaný typ, ale instruuje čtenáře Parquet, aby sloupec přečetl jako zadaný typ. V případě neshody Auto Loader zachrání data umístěním do sloupce pro zachráněná data.

Jak funguje vývoj schématu Auto Loaderu?

Auto Loader zjistí přidání nových sloupců při zpracování dat. Když Auto Loader zjistí nový sloupec, datový proud se zastaví s UnknownFieldException. Než váš datový proud vyvolá tuto chybu, Auto Loader provede odvozování schématu na nejnovější mikrodávce dat a aktualizuje lokaci schématu pomocí nejnovějšího schématu sloučením nových sloupců na konec schématu. Datové typy existujících sloupců zůstávají beze změny.

Databricks doporučuje nakonfigurovat Auto Loader streamy pomocí úloh Lakeflow, aby se po takových změnách schématu automaticky restartovaly.

Auto Loader podporuje pro vývoj schématu následující režimy, které jste nastavili v cloudFiles.schemaEvolutionMode této možnosti:

Režim Chování při čtení nového sloupce
addNewColumns (výchozí) Stream selže s UnknownFieldException poté, co Auto Loader přidá do schématu nové sloupce. Restartováním streamu se obnoví zpracování s aktualizovaným schématem. Existující sloupce nemění datové typy. Azure Databricks doporučuje konfigurovat streamy Auto Loaderu pomocí Lakeflow Jobs, aby se automaticky restartovaly.
addNewColumnsWithTypeWidening Stejné chování jako addNewColumns, ale Auto Loader také rozšiřuje podporované datové typy (například int na long). Změny nepodporovaného typu (například int do string) se přidají do sloupce zachráněných dat.
rescue Auto Loader nikdy nevyvíjí schéma a datový proud nezastaví fungování kvůli změnám schématu. Auto Loader zaznamenává všechny nové sloupce ve sloupci zachráněných dat.
failOnNewColumns Stream selže a nerestartuje, pokud neaktualizujete zadané schéma nebo neodeberete chybný datový soubor. Schéma se neaktualizuje automaticky.
none Nevyvíjí schéma, nové sloupce se ignorují a data se nezachovají, pokud není nastavená rescuedDataColumn možnost. Stream neselže kvůli změnám schématu.

Poznámka:

addNewColumns režim je výchozí, pokud schéma není zadané, ale none je výchozí při zadání schématu. addNewColumns není povoleno, pokud je poskytnuto schéma pro stream, ale funguje, pokud zadáte své schéma jako nápovědu schématu .

Auto Loader také podporuje automatické rozšíření typu pomocí režimu vývoje schématu addNewColumnsWithTypeWidening . Tento režim automaticky rozšiřuje datové typy (například int na long nebo float na double) bez nutnosti přepsání dat nebo zásahu uživatele. Tato funkce je ve verzi Public Preview ve verzi Databricks Runtime 16.4 a vyšší. Viz Automatické rozšíření typů pomocí Auto Loaderu.

Jak fungují oddíly s Auto Loaderem?

Automatický zavaděč se pokusí odvodit sloupce oddílů ze základní adresářové struktury dat, pokud jsou data rozložená v dělení stylu Hive. Například cesta k souboru base_path/event=click/date=2021-04-01/f0.json vede k tomu, že jsou date a event odvozeny jako sloupce oddílů. Pokud základní adresářová struktura obsahuje konfliktní oddíly Hive nebo neobsahuje dělení ve stylu Hive, Auto Loader ignoruje sloupce přiřazené oddílům.

Binární soubor (binaryFile) a text formáty souborů mají pevná schémata dat, ale podporují odvozování sloupců oddílů. Databricks doporučuje nastavit cloudFiles.schemaLocation pro tyto formáty souborů. Tím se zabrání případným chybám nebo ztrátě informací a zabrání nutnosti odvozovat sloupce oddílů při každém spuštění Auto Loaderu.

Automatický zavaděč nebere v úvahu partiční sloupce při vývoji schématu. Pokud jste měli počáteční adresářovou strukturu jako base_path/event=click/date=2021-04-01/f0.json, a pak začněte přijímat nové soubory jako base_path/event=click/date=2021-04-01/hour=01/f1.json, Automatický Zavaděč ignoruje sloupec hodin. Pokud chcete zaznamenat informace pro nové sloupce oddílů, nastavte cloudFiles.partitionColumns na event,date,hour.

Poznámka:

Tato cloudFiles.partitionColumns možnost přebírá čárkami oddělený seznam názvů sloupců. Auto Loader analyzuje pouze sloupce, které existují jako key=value páry ve vaší adresářové struktuře.

Co je sloupec obnovených dat?

Když Auto Loader odvodí schéma, Auto Loader automaticky přidá do schématu záchranný datový sloupec jako _rescued_data. Sloupec můžete přejmenovat nebo zahrnout, když zadáte schéma, a to nastavením rescuedDataColumn možnosti.

Sloupec zachráněných dat zajišťuje, aby Auto Loader zachraňuje sloupce, které neodpovídají schématu, místo aby je vyhodil. Sloupec zachráněných dat obsahuje všechna data, která nejsou analyzována z následujících důvodů:

  • Ve schématu chybí sloupec.
  • Neshody typů
  • Neshody velkých a malých písmen.

Sloupec zachráněných dat obsahuje objekt blob JSON se záchrannými sloupci a cestou ke zdrojovému souboru záznamu.

Poznámka:

Analyzátory JSON a CSV podporují při analýze záznamů tři režimy: PERMISSIVE, DROPMALFORMEDa FAILFAST. Při použití společně s datovým typem rescuedDataColumn neshody nezpůsobují, že automatický zavaděč zahazuje záznamy v režimu DROPMALFORMED ani nevyvolávají chybu v režimu FAILFAST. Pouze poškozené záznamy selžou nebo můžou vyvolat chyby, jako jsou neúplné nebo poškozené soubory JSON nebo CSV. Pokud použijete badRecordsPath při analýze JSON nebo CSV, Auto Loader nezachází s neshodami datových typů jako se špatnými záznamy při použití rescuedDataColumn. Auto Loader ukládá pouze neúplné a poškozené záznamy JSON nebo CSV v badRecordsPathsouboru .

Změna chování rozlišujícího malá a velká písmena

Pokud není povoleno rozlišování velikosti písmen, Auto Loader považuje sloupce abc, Abc, a ABC jako stejný sloupec pro účely odvozování schématu. Auto Loader libovolně vybírá scénář na základě vzorkovaných dat. Pomocí nápověd schématu můžete určit, který případe by měl být použit. Jakmile auto Loader provede výběr a odvodí schéma, nepovažuje varianty velikostí písmen, které nebyly vybrány v souladu se schématem.

Pokud je povolený zachráněný datový sloupec, automatický zavaděč načte pole pojmenovaná v jiném případě, než je schéma, do _rescued_data sloupce. Toto chování změňte nastavením možnosti readerCaseSensitive na false; v takovém případě Auto Loader čte data bez ohledu na velká a malá písmena.

Přepsání odvození schématu pomocí tipů schématu

Pomocí nápovědu ke schématu můžete vynutit informace o schématu, které znáte a očekáváte u odvozeného schématu. Pokud víte, že sloupec je konkrétní datový typ, nebo pokud chcete zvolit obecnější datový typ (například double místo integer), můžete zadat libovolný počet tipů pro datové typy sloupců jako řetězec pomocí syntaxe specifikace schématu SQL, například:

.option("cloudFiles.schemaHints", "tags map<string,string>, version int")

Seznam podporovaných datových typů najdete v tématu Mapování jazyka.

Pokud sloupec není na začátku datového proudu, můžete také použít náznaky schématu k přidání tohoto sloupce do odvozeného schématu.

Následující příklad ukazuje odvozené schéma a výsledek aplikace náznaků schématu.

Odvozené schéma:

|-- date: string
|-- quantity: int
|-- user_info: struct
|    |-- id: string
|    |-- name: string
|    |-- dob: string
|-- purchase_options: struct
|    |-- delivery_address: string

Zadáním následujících tipů schématu:

.option("cloudFiles.schemaHints", "date DATE, user_info.dob DATE, purchase_options MAP<STRING,STRING>, time TIMESTAMP")

získáte:

|-- date: string -> date
|-- quantity: int
|-- user_info: struct
|    |-- id: string
|    |-- name: string
|    |-- dob: string -> date
|-- purchase_options: struct -> map<string,string>
|-- time: timestamp

Poznámka:

Podpora nápovědy pro schéma polí a map je dostupná v Databricks Runtime 9.1 LTS a novějších.

Následující příklad ukazuje odvozené schéma se složitými datovými typy a výsledek použití tipů schématu.

Odvozené schéma:

|-- products: array<string>
|-- locations: array<string>
|-- users: array<struct>
|    |-- users.element: struct
|    |    |-- id: string
|    |    |-- name: string
|    |    |-- dob: string
|-- ids: map<string,string>
|-- names: map<string,string>
|-- prices: map<string,string>
|-- discounts: map<struct,string>
|    |-- discounts.key: struct
|    |    |-- id: string
|    |-- discounts.value: string
|-- descriptions: map<string,struct>
|    |-- descriptions.key: string
|    |-- descriptions.value: struct
|    |    |-- content: int

Zadáním následujících tipů schématu:

.option("cloudFiles.schemaHints", "products ARRAY<INT>, locations.element STRING, users.element.id INT, ids MAP<STRING,INT>, names.key INT, prices.value INT, discounts.key.id INT, descriptions.value.content STRING")

získáte:

|-- products: array<string> -> array<int>
|-- locations: array<int> -> array<string>
|-- users: array<struct>
|    |-- users.element: struct
|    |    |-- id: string -> int
|    |    |-- name: string
|    |    |-- dob: string
|-- ids: map<string,string> -> map<string,int>
|-- names: map<string,string> -> map<int,string>
|-- prices: map<string,string> -> map<string,int>
|-- discounts: map<struct,string>
|    |-- discounts.key: struct
|    |    |-- id: string -> int
|    |-- discounts.value: string
|-- descriptions: map<string,struct>
|    |-- descriptions.key: string
|    |-- descriptions.value: struct
|    |    |-- content: int -> string

Poznámka:

Automatický zavaděč používá nápovědy schématu pouze v případě, že nezadáte schéma. Můžete použít nápovědu schématu, jestli cloudFiles.inferColumnTypes je povolená nebo zakázaná.

Další kroky

  • Last updated on