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 tipoSTRING
.A coluna
variant_column
contém todas as cadeias de caracteres JSON codificadas como tipoVARIANT
.
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 evolução do esquema durante a ingestão e rescuedDataColumn
não é suportado. O exemplo a seguir pressupõe que a tabela de destino já exista 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 sobre 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")
)