Ingestování dat jako částečně strukturovaný typ varianty

Důležité

Tato funkce je ve verzi Public Preview.

V Databricks Runtime 15.3 a novějších můžete použít VARIANT typ k ingestování částečně strukturovaných dat. Tento článek popisuje chování a poskytuje příklady vzorů pro příjem dat z cloudového objektového úložiště pomocí Auto Loaderu a COPY INTO, streamování záznamů z Kafky a příkazů SQL pro vytváření nových tabulek s různými daty nebo vkládání nových záznamů pomocí typu varianty. Následující tabulka shrnuje podporované formáty souborů a podporu verzí Databricks Runtime:

Formát souboru Podporovaná verze Databricks Runtime
JSON 15.3 a vyšší
jazyk XML 16.4 a vyšší
formát CSV 16.4 a vyšší

Viz Data dotazu na varianty.

Vytvoření tabulky s variantovým sloupcem

VARIANT je standardní typ SQL v Databricks Runtime 15.3 a novější a podporuje je tabulky zálohované Delta Lake. Spravované tabulky v Azure Databricks ve výchozím nastavení používají Delta Lake, takže můžete vytvořit prázdnou tabulku s jedním VARIANT sloupcem pomocí následující syntaxe.

CREATE TABLE table_name (variant_column VARIANT)

Alternativně můžete pomocí příkazu CTAS vytvořit tabulku s variantovým sloupcem. Použijte funkci PARSE_JSON k parsování řetězců JSON nebo funkci FROM_XML k parsování řetězců XML. Následující příklad vytvoří tabulku se dvěma sloupci.

  • Sloupec id se extrahuje z řetězce JSON jako STRING typ.
  • variant_column obsahuje celý řetězec JSON kódovaný jako VARIANT typ.
CREATE TABLE table_name AS
  SELECT json_string:id AS id,
    PARSE_JSON(json_string) variant_column
  FROM source_data

Poznámka:

Databricks doporučuje extrahovat často dotazovaná pole a ukládat je jako ne variantní sloupce, aby se urychlily dotazy a optimalizovaly rozložení úložiště.

VARIANT sloupce nelze použít pro clusteringové klíče, oddíly nebo Z-řádové klíče. Datový typ VARIANT nelze použít pro porovnání, seskupování, řazení a operace nastavení. Další informace najdete v tématu Omezení.

Vložení dat pomocí parse_json

Pokud cílová tabulka již obsahuje sloupec kódovaný jako VARIANT, můžete použít parse_json k vložení záznamů řetězce JSON jako VARIANT. Například parsujte řetězce JSON ze json_string sloupce a vložte je do table_name.

SQL

INSERT INTO table_name (variant_column)
  SELECT PARSE_JSON(json_string)
  FROM source_data

Python

from pyspark.sql.functions import col, parse_json

(spark.read
  .table("source_data")
  .select(parse_json(col("json_string")))
  .write
  .mode("append")
  .saveAsTable("table_name")
)

Vložení dat pomocí from_xml

Pokud cílová tabulka již obsahuje sloupec kódovaný jako VARIANT, můžete použít from_xml k vložení záznamů řetězců XML jako VARIANT. Například parsujte řetězce XML ze xml_string sloupce a vložte je do table_name.

SQL

INSERT INTO table_name (variant_column)
  SELECT FROM_XML(xml_string, 'variant')
  FROM source_data

Python

from pyspark.sql.functions import col, from_xml

(spark.read
  .table("source_data")
  .select(from_xml(col("xml_string"), "variant"))
  .write
  .mode("append")
  .saveAsTable("table_name")
)

Vložení dat pomocí from_csv

Pokud cílová tabulka již obsahuje sloupec kódovaný jako VARIANT, můžete použít from_csv k vložení záznamů řetězce CSV jako VARIANT. Například parsujte záznamy CSV ze csv_string sloupce a vložte je do table_name.

SQL

INSERT INTO table_name (variant_column)
  SELECT FROM_CSV(csv_string, 'v variant').v
  FROM source_data

Python

from pyspark.sql.functions import col, from_csv

(spark.read
  .table("source_data")
  .select(from_csv(col("csv_string"), "v variant").v)
  .write
  .mode("append")
  .saveAsTable("table_name")
)

Příjem dat z cloudového úložiště objektů jako varianta

Auto Loader lze použít k načtení všech dat z podporovaných zdrojů souborů jako jednoho VARIANT sloupce v cílové tabulce. Vzhledem k tomu, že VARIANT je flexibilní pro změny schématu a typu a udržuje citlivost na velikost písmen a hodnoty NULL existující ve zdroji dat, je tento vzor robustní pro většinu scénářů příjmu dat s následujícími výhradami:

  • Chybně formátované záznamy nelze zakódovat pomocí VARIANT typu.
  • VARIANT typ může obsahovat pouze záznamy o velikosti až 16 MB.

Poznámka:

Variant zpracovává příliš velké záznamy stejně jako poškozené záznamy. Ve výchozím PERMISSIVE režimu zpracování jsou příliš velké záznamy zachyceny v corruptRecordColumn souboru.

Vzhledem k tomu, že se celý záznam zaznamená jako jeden VARIANT sloupec, během příjmu dat nedojde k žádnému vývoji schématu a rescuedDataColumn nepodporuje se. Následující příklad předpokládá, že cílová tabulka již existuje s jedním VARIANT sloupcem.

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("singleVariantColumn", "variant_column")
  .load("/Volumes/catalog_name/schema_name/volume_name/path")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("table_name")
)

Můžete také určit VARIANT při definování schématu nebo předání schemaHints. Data v odkazovaném zdrojovém poli musí obsahovat platný záznam. Následující příklady ukazují tuto syntaxi.

# Define the schema.
# Writes the columns `name` as a string and `address` as variant.
(spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  .schema("name STRING, address VARIANT")
  .load("/Volumes/catalog_name/schema_name/volume_name/path")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("table_name")
)

# Define the schema.
# A single field `payload` containing JSON data is written as variant.
(spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  .schema("payload VARIANT")
  .load("/Volumes/catalog_name/schema_name/volume_name/path")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("table_name")
)

# Supply schema hints.
# Writes the `address` column as variant.
# Infers the schema for other fields using standard rules.
(spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaHints", "address VARIANT")
  .load("/Volumes/catalog_name/schema_name/volume_name/path")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("table_name")
)

Použít COPY INTO s variantou

Databricks doporučuje používat Auto Loader namísto COPY INTO, pokud je k dispozici.

COPY INTO podporuje ingestování celého obsahu podporovaného zdroje dat jako jednoho sloupce. Následující příklad vytvoří novou tabulku s jedním sloupcem VARIANT a pak použije COPY INTO k ingestování záznamů ze zdroje souboru JSON.

CREATE TABLE table_name (variant_column VARIANT);
COPY INTO table_name
  FROM '/Volumes/catalog_name/schema_name/volume_name/path'
  FILEFORMAT = JSON
  FILES = ('file-name')
  FORMAT_OPTIONS ('singleVariantColumn' = 'variant_column')

Streamování dat Kafka jako varianta

Mnoho datových proudů Kafka kóduje své payloady pomocí JSON. Ingestování datových proudů Kafka pomocí VARIANT dělá tyto úlohy odolné vůči změnám schématu.

Následující příklad ukazuje čtení zdroje streamování Kafka, přetypování key jako STRING a value jako VARIANTa zápis do cílové tabulky.

from pyspark.sql.functions import col, parse_json

(spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .option("startingOffsets", "earliest")
  .load()
  .select(
    col("key").cast("string"),
    parse_json(col("value").cast("string"))
  ).writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("table_name")
)

Další kroky

  • Last updated on