Ingerir dados como tipo de variante semiestruturada

Visualização

Esse recurso está em Prévia Pública.

Em Databricks Runtime 15.3 e acima, o senhor pode usar o tipo VARIANT para ingerir dados semiestruturados. Este artigo descreve o comportamento e fornece padrões de exemplo para a ingestão de dados do armazenamento de objetos cloud usando Auto Loader e COPY INTO, registros de transmissão de Kafka e SQL comando para criar novas tabelas com dados de variantes ou inserir novos registros usando o tipo de variante.

Consulte Consultar dados de variantes.

Criar uma tabela com uma coluna de variantes

VARIANT é um tipo padrão SQL em Databricks Runtime 15.3 e acima e suportado por tabelas apoiadas por Delta Lake. Gerenciar tabelas em Databricks usa Delta Lake por default, portanto, o senhor pode criar uma tabela vazia com uma única coluna VARIANT usando a seguinte sintaxe:

CREATE TABLE table_name (variant_column VARIANT)

Como alternativa, o senhor pode usar a função PARSE_JSON em uma cadeia de caracteres JSON para usar uma instrução CTAS para criar uma tabela com uma coluna de variantes. O exemplo a seguir cria uma tabela com duas colunas:

  • A coluna id extraída das cadeias de caracteres JSON como um tipo STRING.

  • A coluna variant_column contém todas as cadeias de caracteres JSON codificadas como tipo VARIANT.

CREATE TABLE table_name AS
  SELECT json_string:id AS id,
    PARSE_JSON(json_string) variant_column
  FROM source_data

Observação

VARIANT não podem ser usadas para a chave clustering, partições ou chave Z-order. Os dados armazenados com o tipo VARIANT não podem ser usados para comparações e ordenação.

Databricks recomenda a extração e o armazenamento de campos como colunas não variantes que o senhor planeja usar para acelerar as consultas e otimizar a disposição do armazenamento.

Inserir uso de dados parse_json

Se a tabela de destino já contiver uma coluna codificada como VARIANT, o senhor poderá usar parse_json para inserir registros de cadeias de caracteres JSON como VARIANT, como no exemplo a seguir:

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

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

Ingerir dados do armazenamento de objetos na nuvem como variante

Em Databricks Runtime 15.3 e acima, o senhor pode usar Auto Loader para carregar todos os dados de JSON fontes como uma única coluna VARIANT em uma tabela de destino. Como VARIANT é flexível a mudanças de esquema e tipo e mantém a sensibilidade a maiúsculas e minúsculas e NULL valores presentes na fonte de dados, esse padrão é robusto para a maioria dos cenários de ingestão, com as seguintes ressalvas:

  • Registros JSON malformados não podem ser codificados usando o tipo VARIANT.

  • VARIANT só pode conter registros de até 16 MB.

Observação

A variante trata registros excessivamente grandes de forma semelhante a registros corrompidos. No modo de processamento default PERMISSIVE, registros excessivamente grandes são capturados na coluna _malformed_data juntamente com registros JSON malformados.

Como todos os dados da fonte JSON são registrados como uma única coluna VARIANT, não ocorre nenhuma evolução do esquema durante a ingestão. O exemplo a seguir pressupõe que a tabela de destino já existe com uma única coluna VARIANT.

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

O senhor também pode especificar VARIANT ao definir um esquema ou passar schemaHints. Os dados no campo de origem referenciado devem conter uma cadeia de caracteres JSON válida. Os exemplos a seguir demonstram essa sintaxe:

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

Use COPY INTO com a variante

A Databricks recomenda o uso do Auto Loader em vez de COPY INTO quando disponível.

COPY INTO suporta a ingestão de todo o conteúdo de uma JSON fonte de dados como uma única coluna. O exemplo a seguir cria uma nova tabela com uma única coluna VARIANT e, em seguida, usa COPY INTO para ingerir registros de uma fonte de arquivo JSON.

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

O senhor também pode definir qualquer campo em uma tabela de destino como VARIANT. Quando o senhor executa COPY INTO, os campos correspondentes na fonte de dados são ingeridos e convertidos para o tipo VARIANT, como nos exemplos a seguir:

-- Extracts the `address` field from the JSON record and casts to variant
CREATE TABLE table_name (address VARIANT);
COPY INTO table_name
  FROM '/Volumes/catalog_name/schema_name/volume_name/path'
  FILEFORMAT = JSON

-- Extracts `name` and `address` from the JSON record and casts `address` to variant
CREATE TABLE table_name (name STRING, address VARIANT);
COPY INTO table_name
  FROM '/Volumes/catalog_name/schema_name/volume_name/path'
  FILEFORMAT = JSON

transmissão Kafka dados como variante

Muitas transmissões Kafka codificam suas cargas úteis usando JSON. A ingestão da transmissão Kafka usando VARIANT torna essas cargas de trabalho robustas em relação às alterações de esquema.

O exemplo a seguir demonstra como ler uma fonte de transmissão Kafka, converter o key em STRING e o value em VARIANT e gravar em uma tabela de destino.

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