Ingeste arquivos como o tipo FILE

O FILE tipo armazena e consulta referências a arquivos não estruturados (documentos, imagens e áudio) em tabelas. Esta página mostra como descobrir arquivos, fazer a ingestão deles como referências FILE e fazer a ingestão incremental de novos arquivos à medida que eles chegam.

Para obter referência sobre o tipo FILE, consulte FILE tipo. Para uma visão geral das abordagens para ingerir dados não estruturados, veja TIPO de ARQUIVO e dados não estruturados.

Modos de armazenamento

Uma FILE referência pode ser armazenada em um de dois modos:

• FILE MANAGEDarmazena cópias de arquivos no armazenamento gerenciado pelo Unity Catalog: as permissões são gerenciadas pela tabela, e excluir linhas torna os arquivos referenciados elegíveis para coleta de lixo, para que a tabela e seus arquivos permaneçam sincronizados. Arquivos de fontes fora de volumes, como SharePoint, Google Drive ou SFTP, devem ser ingeridos e armazenados como FILE MANAGED.
• FILE EXTERNAL referencia arquivos que já existem em um volume do Unity Catalog. O Databricks não suporta armazenar FILE EXTERNAL referências para arquivos armazenados fora de volumes.

O Azure Databricks recomenda FILE MANAGED para cargas de trabalho que se beneficiam de permissões em nível de arquivo e conformidade embutida. Para uma comparação entre governança e comportamento do ciclo de vida, veja tipo FILE e dados não estruturados.

Uso list_files para descobrir arquivos

Use a função com valor de tabela list_files função com valor de tabela para descobrir os arquivos disponíveis em um caminho. Ele retorna uma linha por arquivo com seu path, size, modification_time e uma referência FILE:

SELECT * FROM list_files('/Volumes/my_catalog/my_schema/raw_files/');

Para descobrir arquivos em uma fonte que requer uma conexão com o Unity Catalog, como SharePoint, Google Drive ou SFTP, adicione o connection parâmetro:

SELECT * FROM list_files('https://example.sharepoint.com/sites/my-site/', connection => 'my_sharepoint_connection');

list_files descobre arquivos recursivamente por padrão. Para saber mais, veja list_files função com valores de tabela.

Ingerir arquivos como referências FILE

Selecione uma abordagem de ingestão com base em onde você armazena seus arquivos. Para ingerir arquivos de uma fonte externa, copie-os para o armazenamento gerenciado como FILE MANAGED. Para referenciar arquivos já em um volume do Unity Catalog sem copiá-los, use FILE EXTERNAL.

Ingira arquivos fonte externos como FILE MANAGED

Para gerar referências FILE para arquivos em uma origem como SharePoint, Google Drive ou SFTP, primeiro faça a ingestão dos arquivos e armazene-os como FILE MANAGED. FILE EXTERNAL Não é suportado para arquivos armazenados fora de volumes.

O exemplo a seguir ingire arquivos do SharePoint em uma FILE MANAGED tabela:

SQL

CREATE TABLE managed_documents (
  file_name STRING,
  path STRING,
  size BIGINT,
  modification_time TIMESTAMP,
  file FILE MANAGED
) USING DELTA
  TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/');

INSERT INTO managed_documents
  SELECT _metadata.file_name, *
  FROM read_files(
    'https://example.sharepoint.com/sites/my-site/',
    connection => 'my_sharepoint_connection',
    format => 'file');

Python

(spark.read.format("file")
  .option("databricks.connection", "my_sharepoint_connection")
  .load("https://example.sharepoint.com/sites/my-site/")
  .selectExpr("_metadata.file_name", "*")
  .writeTo("managed_documents").append())

Scala

spark.read.format("file")
  .option("databricks.connection", "my_sharepoint_connection")
  .load("https://example.sharepoint.com/sites/my-site/")
  .selectExpr("_metadata.file_name", "*")
  .writeTo("managed_documents").append()

Ingerir arquivos de volume como FILE EXTERNAL

Para fazer a ingestão de arquivos que já existem em um volume do Unity Catalog, use a instrução CREATE TABLE AS SELECT (CTAS) com list_files. Isso cria uma tabela com uma FILE EXTERNAL coluna que faz referência a cada arquivo no local, sem copiar seu conteúdo. O exemplo a seguir cria uma documents tabela com o nome do arquivo, metadados e uma FILE referência para cada arquivo:

CREATE TABLE documents AS
  SELECT _metadata.file_name, *
  FROM list_files('/Volumes/my_catalog/my_schema/raw_files/');

Use pipelines para ingerir novos arquivos de forma incremental

Para ingerir novos arquivos à medida que eles chegam, use uma tabela de streaming em um pipeline do Lakeflow que lê a origem com STREAM read_files(..., format => 'file'). Cada atualização do pipeline processa apenas os arquivos adicionados após a última atualização. Veja read_files e Spark Declarative Pipelines on Lakeflow.

Para transmitir arquivos incrementalmente de uma fonte como o Google Drive:

1. Defina o canal do pipeline para PREVIEW. Ingerir referências FILE em um pipeline requer o canal PREVIEW.

2. Defina uma tabela de streaming que leia a fonte com STREAM read_files(..., format => 'file'), como no seguinte código:

   SQL

   CREATE STREAMING TABLE streaming_documents (
     path STRING,
     size BIGINT,
     modification_time TIMESTAMP,
     file FILE MANAGED
   )
   TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/')
   AS SELECT *
     FROM STREAM read_files(
       'https://drive.google.com/drive/folders/my-folder-id',
       connection => 'my_gdrive_connection',
       format => 'file');
   

   Python

   from pyspark import pipelines as dp
   
   @dp.table(
     name="streaming_documents",
     schema="path STRING, size BIGINT, modification_time TIMESTAMP, file FILE MANAGED",
     table_properties={"databricks.filespace-preview": "/Volumes/my_catalog/my_schema/filespace/"}
   )
   def streaming_documents():
     return (
       spark.readStream.format("cloudFiles")
         .option("cloudFiles.format", "file")
         .option("databricks.connection", "my_gdrive_connection")
         .load("https://drive.google.com/drive/folders/my-folder-id")
     )
   

Aplique atualizações e excluções com o AUTO CDC

Uma ingesta de streaming adiciona novos arquivos, mas não captura atualizações ou excluções da fonte. Para aplicar essas mudanças, leia o feed de alterações da origem com AUTO CDC.

Ingira o feed de alterações em duas etapas. O exemplo a seguir ingere o feed de alterações do SharePoint e o aplica a uma tabela de streaming alvo como SCD tipo 1:

1. Escreva os dados de alteração em uma tabela de streaming com arquivos gerenciados, conforme no código a seguir. Defina readChangeFeed => true como Ativo em read_files para retornar o feed de alterações, que inclui as colunas de metadados _file_id, _sequence e _is_deleted.

   SQL

   CREATE OR REFRESH STREAMING TABLE documents_changes (
     _file_id STRING,
     _sequence BIGINT,
     _is_deleted BOOLEAN,
     path STRING,
     size BIGINT,
     modification_time TIMESTAMP,
     file FILE MANAGED
   )
   TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/')
   AS SELECT *
     FROM STREAM read_files(
       'https://example.sharepoint.com/sites/my-site/',
       connection => 'my_sharepoint_connection',
       format => 'file',
       readChangeFeed => true);
   

   Python

   from pyspark import pipelines as dp
   
   @dp.table(
     name="documents_changes",
     table_properties={"databricks.filespace-preview": "/Volumes/my_catalog/my_schema/filespace/"}
   )
   def documents_changes():
     return (
       spark.readStream.format("cloudFiles")
         .option("cloudFiles.format", "file")
         .option("databricks.connection", "my_sharepoint_connection")
         .option("cloudFiles.readChangeFeed", "true")
         .load("https://example.sharepoint.com/sites/my-site/")
     )
   

2. Use AUTO CDC para aplicar as alterações dessa tabela a uma tabela de streaming alvo, conforme no código a seguir. Use _file_id como chave, _sequence como coluna de sequência, e _is_deleted para identificar exclusões.

   SQL

   CREATE OR REFRESH STREAMING TABLE documents
     TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/');
   
   CREATE FLOW documents_cdc AS AUTO CDC INTO
     documents
   FROM STREAM documents_changes
     KEYS (_file_id)
     APPLY AS DELETE WHEN _is_deleted = true
     SEQUENCE BY _sequence
     COLUMNS * EXCEPT (_is_deleted, _sequence)
     STORED AS SCD TYPE 1;
   

   Python

   from pyspark import pipelines as dp
   from pyspark.sql.functions import col, expr
   
   dp.create_streaming_table(
     name="documents",
     table_properties={"databricks.filespace-preview": "/Volumes/my_catalog/my_schema/filespace/"}
   )
   
   dp.create_auto_cdc_flow(
     target = "documents",
     source = "documents_changes",
     keys = ["_file_id"],
     sequence_by = col("_sequence"),
     apply_as_deletes = expr("_is_deleted = true"),
     except_column_list = ["_is_deleted", "_sequence"],
     stored_as_scd_type = 1
   )
   

Converter dados binários inline em referências de ARQUIVO

Se uma tabela já armazena o conteúdo do arquivo como dados binários inline, use a função create_file para gravar esses dados no armazenamento e gerar uma referência FILE.

Os exemplos a seguir usam uma tabela gerada pelo usuário, raw_documents, com uma name coluna e uma content coluna que armazenam os dados binários.

Escreva dados binários no armazenamento gerenciado como FILE MANAGED

Para armazenar os arquivos como arquivos gerenciados, chame create_file apenas com o conteúdo binário. Quando você omite destination_path, o Unity Catalog envia o conteúdo para o local de armazenamento gerenciado:

SQL

CREATE TABLE managed_documents (name STRING, file FILE MANAGED) USING DELTA
  TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/');

INSERT INTO managed_documents (name, file)
  SELECT name, create_file(content => content)
  FROM raw_documents;

Python

(spark.read.table("raw_documents")
  .selectExpr("name", "create_file(content => content) AS file")
  .writeTo("managed_documents").append())

Scala

spark.read.table("raw_documents")
  .selectExpr("name", "create_file(content => content) AS file")
  .writeTo("managed_documents").append()

Gravar dados binários em um volume como ARQUIVO EXTERNO

Para gravar os arquivos em um volume do Unity Catalog como arquivos externos, passe destination_path para create_file, conforme no seguinte código:

SQL

CREATE TABLE documents (name STRING, file FILE EXTERNAL) USING DELTA;

INSERT INTO documents (name, file)
  SELECT
    name,
    create_file(
      content => content,
      destination_path => '/Volumes/my_catalog/my_schema/my_volume/' || name
    )
  FROM raw_documents;

Python

(spark.read.table("raw_documents")
  .selectExpr(
    "name",
    "create_file(content => content, destination_path => '/Volumes/my_catalog/my_schema/my_volume/' || name) AS file")
  .writeTo("documents").append())

Scala

spark.read.table("raw_documents")
  .selectExpr(
    "name",
    "create_file(content => content, destination_path => '/Volumes/my_catalog/my_schema/my_volume/' || name) AS file")
  .writeTo("documents").append()

Próximas Etapas 

• FILE tipo
• Tipo de arquivo e dados não estruturados
• Tutorial: Construa um pipeline de processamento de arquivos com o tipo FILE
• Saiba mais sobre o Carregador Automático. Confira O que é o Carregador Automático?.