Inmata filer som FILE-typen

Typen FILE lagrar och frågar referenser till ostrukturerade filer (dokument, bilder och ljud) i tabeller. Den här sidan visar hur man upptäcker filer, tar in dem som FILE referenser och successivt tar in nya filer när de anländer.

För referensen om FILE typen, se FILE typ. För en översikt över metoder för att ta in ostrukturerad data, se FILTYP och ostrukturerad data.

Lagringslägen

En FILE referens kan lagras i ett av två moder:

• FILE MANAGEDlagrar kopior av filer i Unity Catalog-hanterad lagring: behörigheter hanteras via tabellen, och rader som rader tar bort gör de refererade filerna lämpliga för skräpsamling, så att tabellen och dess filer förblir synkroniserade. Filer från källor utanför volymer, såsom SharePoint, Google Drive eller SFTP, måste hämtas och lagras som FILE MANAGED.
• FILE EXTERNAL refererar till filer som redan finns i en Unity-katalogvolym. Databricks stöder inte lagring FILE EXTERNAL av referenser för filer som lagras utanför volymer.

Azure Databricks rekommenderar FILE MANAGED för arbetsbelastningar som drar nytta av filnivåbehörigheter och inbyggd efterlevnad. För en jämförelse av styrning och livscykelbeteende, se FILTYP och ostrukturerad data.

Använd list_files för att upptäcka filer

Använd tabellvärdsfunktionenlist_files tabellvärd funktion för att upptäcka filerna som finns tillgängliga vid en sökväg. Den returnerar en rad per fil med sin path, size, modification_time, och en FILE referens:

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

För att upptäcka filer i en källkod som kräver en Unity Catalog-anslutning, såsom SharePoint, Google Drive eller SFTP, lägg till parameternconnection:

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

list_files Upptäcker filer rekursivt som standard. För att lära dig mer, se list_files tabellvärdad funktion.

Inmata filer som FILE-referenser

Välj en inskrivningsmetod baserat på var du lagrar dina filer. För att ta in filer från en extern källa, kopiera dem till hanterad lagring som FILE MANAGED. För att referera filer som redan finns i en Unity-katalogvolym utan att kopiera dem, använd FILE EXTERNAL.

Inhämta externa källfiler som FILHANTERAD

För att generera FILE referenser för filer i en källa som SharePoint, Google Drive eller SFTP, importera filerna först och lagra dem som FILE MANAGED. FILE EXTERNAL stöds inte för filer som lagras utanför volymer.

Följande exempel matar in filer från SharePoint i en FILE MANAGED tabell:

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

Inmata volymfiler som EXTERN FIL

För att inge filer som redan finns i en Unity Catalog-volym, använd en CREATE TABLE AS SELECT (CTAS)-sats med list_files. Detta skapar en tabell med en FILE EXTERNAL kolumn som refererar till varje fil på plats, utan att kopiera dess innehåll. Följande exempel skapar en documents tabell med filnamn, metadata och en FILE referens för varje fil:

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

Använd pipelines för att inkrementiellt ta in nya filer

För att ta emot nya filer när de anländer, använd en strömningstabell i en Lakeflow-pipeline som läser källkoden med STREAM read_files(..., format => 'file'). Varje pipelineuppdatering behandlar endast filerna som lagts till efter den senaste uppdateringen. Se read_files och tänd deklarativa pipelines.

För att stegvis strömma filer från en källa som Google Drive:

1. Ställ in pipelinens kanal till PREVIEW. Att ta in FILE referenser i en pipeline kräver kanalen PREVIEW .

2. Definiera en strömningstabell som läser källkoden med STREAM read_files(..., format => 'file'), som i följande kod:

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

Tillämpa uppdateringar och raderingar med AUTO CDC

En streaming-inmatning lägger till nya filer men fångar inte uppdateringar eller raderingar från källan. För att tillämpa dessa ändringar, läs källändringsflödet med AUTO CDC.

Ta in ändringsflödet i två steg. Följande exempel tar in ändringsflödet från SharePoint och applicerar det sedan på en målströmningstabell som SCD typ 1:

1. Skriv ändringsdata i en strömningstabell med hanterade filer, som i följande kod. Sätt readChangeFeed => true på read_files för att returnera ändringsflödet, som inkluderar kolumnerna _file_id, _sequence, och _is_deleted metadata.

   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. Använd AUTO CDC för att tillämpa ändringarna från den tabellen på en målströmningstabell, som i följande kod. Använd _file_id som nyckel, _sequence som sekvenskolumn och _is_deleted för att identifiera raderingar.

   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
   )
   

Konvertera inline-binär data till FILE-referenser

Om en tabell redan lagrar filinnehåll som inline-binär data, använd create_file funktionen för att skriva den datan till lagring och skapa en FILE referens.

Följande exempel använder en användargenererad tabell, raw_documents, med en name kolumn och en content kolumn som innehåller binära data.

Skriv binär data till managed storage som FILE MANAGED

För att lagra filerna som hanterade filer, anropa create_file endast det binära innehållet. När du utelämnar destination_path, laddar Unity Catalog upp innehållet till den hanterade lagringsplatsen:

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

Skriv binär data till en volym som FIL EXTERN

För att skriva filerna till en Unity-katalogvolym som externa filer istället, skicka a destination_path till create_file, som i följande kod:

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

Nästa steg

• FILE typ
• FIL-typ och ostrukturerad data
• Handledning: Bygg en filbehandlingspipeline med FILE-typen
• Läs mer om Auto Loader. Se Vad är en automatisk inläsare?.