Importar ficheiros como o tipo FILE

O FILE tipo armazena e consulta a ficheiros não estruturados (documentos, imagens e áudio) em tabelas. Esta página mostra como descobrir ficheiros, ingeri-los como referências FILE e ingerir novos ficheiros de forma incremental à medida que chegam.

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

Modos de armazenamento

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

• FILE MANAGEDarmazena cópias de ficheiros no armazenamento gerido pelo Unity Catalog: as permissões são geridas através da tabela, e eliminar linhas torna os ficheiros referenciados elegíveis para recolha de lixo, para que a tabela e os seus ficheiros permaneçam sincronizados. Ficheiros provenientes de fontes fora dos volumes, como SharePoint, Google Drive ou SFTP, devem ser ingeridos e armazenados como FILE MANAGED.
• FILE EXTERNAL faz referência a ficheiros que já existem num volume do Unity Catalog. O Databricks não suporta armazenar FILE EXTERNAL referências para ficheiros armazenados fora de volumes.

O Azure Databricks recomenda FILE MANAGED para cargas de trabalho que beneficiam de permissões ao nível de ficheiro e conformidade incorporada. Para uma comparação entre governação e comportamento ao longo do ciclo de vida, veja Tipo de ficheiro e dados não estruturados.

Usar list_files para descobrir ficheiros

Use a list_files função com valor de tabela função com valor de tabela para descobrir os ficheiros disponíveis num caminho. Retorna uma linha por ficheiro com o seu path, size, modification_time e uma referência FILE:

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

Para descobrir ficheiros numa fonte que requer uma ligação ao 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 ficheiros recursivamente por defeito. Para saber mais, veja list_files função com valores de tabela.

Importar ficheiros como referências FILE

Selecione uma abordagem de ingestão com base no local onde armazena os seus ficheiros. Para ingerir ficheiros de uma fonte externa, copie-os para armazenamento gerido como FILE MANAGED. Para referenciar ficheiros já num volume do Unity Catalog sem os copiar, use FILE EXTERNAL.

Importar ficheiros de origem externos como FILE MANAGED

Para gerar FILE referências para ficheiros numa fonte como SharePoint, Google Drive ou SFTP, ingera primeiro os ficheiros e armazene-os como FILE MANAGED. FILE EXTERNAL Não é suportado para ficheiros armazenados fora de volumes.

O exemplo seguinte ingere ficheiros do SharePoint numa 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()

Importar ficheiros de volumes como FICHEIRO EXTERNO

Para ingerir ficheiros que já existem num volume do Unity Catalog, use uma instrução CREATE TABLE AS SELECT (CTAS) com list_files. Isto cria uma tabela com uma FILE EXTERNAL coluna que faz referência a cada ficheiro no local, sem copiar o seu conteúdo. O exemplo seguinte cria uma documents tabela com o nome do ficheiro, metadados e uma FILE referência para cada ficheiro:

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

Utilize pipelines para ingerir novos ficheiros de forma incremental

Para ingerir novos ficheiros à medida que chegam, utilize uma tabela de transmissão contínua num pipeline do Lakeflow que lê a origem com STREAM read_files(..., format => 'file'). Cada atualização do pipeline processa apenas os ficheiros adicionados após a última atualização. Veja read_files e Spark Declarative Pipelines on Lakeflow.

Para transmitir ficheiros de forma incremental a partir de uma fonte como o Google Drive:

1. Defina o canal do pipeline como PREVIEW. A ingestão de referências FILE num pipeline requer o canal PREVIEW.

2. Defina uma tabela de streaming que lê 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")
     )
   

Aplicar atualizações e eliminações com o AUTO CDC

Uma ingestão de streaming adiciona novos ficheiros mas não captura atualizações ou eliminações da fonte. Para aplicar essas alterações, leia o feed de alterações da origem com AUTO CDC.

Ingira o feed de alterações em dois passos. O exemplo seguinte ingere o feed de alterações do SharePoint e aplica-o a uma tabela de streaming alvo como SCD tipo 1:

1. Escreva os dados de alteração numa tabela de streaming com ficheiros geridos, como no código seguinte. Defina readChangeFeed => true em read_files para devolver 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, como no código seguinte. Use _file_id como chave, _sequence como coluna de sequência, e _is_deleted para identificar eliminaçõ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 FICHEIRO

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

Os exemplos seguintes utilizam uma tabela gerada pelo utilizador, raw_documents, com uma name coluna e uma content coluna que armazenam os dados binários.

Escrever dados binários para armazenamento gerido como FILE MANAGED

Para armazenar os ficheiros como ficheiros geridos, chame create_file apenas com o conteúdo binário. Quando omites destination_path, o Unity Catalog carrega o conteúdo para o local de armazenamento gerido:

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

Escrever dados binários num volume como FICHEIRO EXTERNO

Para escrever os ficheiros num volume do Unity Catalog como ficheiros externos, passe destination_path para create_file, como 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()

Passos seguintes

• FILE tipo
• Tipo de ficheiro e dados não estruturados
• Tutorial: Construa um pipeline de processamento de ficheiros com o tipo FILE
• Saiba mais sobre o Auto Loader. Veja O que é o Auto Loader?.