파일 형식으로 파일 인제스트

Important

이 기능은 베타 버전으로 제공됩니다. 작업 영역 관리자는 미리 보기 페이지에서 이 기능에 대한 액세스를 제어할 수 있습니다. Azure Databricks 미리 보기 관리를 참조하세요.

이 타입은 FILE 비구조화 파일(문서, 이미지, 오디오)에 대한 참조를 테이블에 저장하고 쿼리합니다. 이 페이지는 파일을 발견하고, 참조로 FILE 수집하며, 새 파일이 도착할 때마다 점진적으로 인징하는 방법을 보여줍니다.

유형에 대한 FILE 참고문헌은 유형을 참조하세요FILE. 비정형 데이터 수집 방법에 대한 개요는 FILE 유형과 비정형 데이터를 참조하세요.

메모

FILE 열에는 정해진 순서가 없습니다. 열을 FILE 분할 열, 클러스터링 열, Z-순서 키로 사용할 수 없습니다. 자세한 내용은 한도를 참조하세요.

스토리지 모드

참조는 FILE 두 가지 모드 중 하나로 저장할 수 있습니다:

FILE EXTERNAL Unity 카탈로그 볼륨에 이미 존재하는 파일을 참조합니다. Databricks는 볼륨 외부에 저장된 파일에 대한 참조 저장을 FILE EXTERNAL 지원하지 않습니다.
FILE MANAGED Unity 카탈로그가 관리하는 저장소에 파일 복사본을 저장합니다. SharePoint, Google Drive, SFTP 등 볼륨 외부의 소스에서 온 파일은 반드시 수집하여 저장FILE MANAGED해야 합니다.

파일 발견에 사용하기 `list_files`

table-value(table-value) 함수를list_files 사용하여 경로에서 이용 가능한 파일을 발견하세요. 파일당 , , modification_time, , FILE 그리고 참조를 포함한 pathsize한 행을 반환합니다:

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

SharePoint, Google Drive, SFTP 등 Unity 카탈로그 연결이 필요한 소스에서 파일을 찾으려면, 다음 매개변수를 추가 connection 하세요:

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

list_files 기본적으로 재귀적으로 파일을 발견합니다. 더 자세한 내용은 테이블 값 함수를 참조하세요list_files.

파일 참조로 파일을 인제스트하기

파일을 저장하는 위치에 따라 수집 방식을 선택하세요. Unity 카탈로그 볼륨 FILE EXTERNAL에 이미 있는 파일을 참조하려면 . 외부 소스에서 파일을 인징하려면 관리 저장소에 FILE MANAGED복사하세요.

FILE EXTERNAL 파일 파일을 인제스트하기

Unity 카탈로그 볼륨 CREATE TABLE AS SELECT 에 이미 존재하는 파일을 인제스트하려면 (CTAS) 문 list_files과 . 이 방법은 각 파일을 참조하는 열이 있는 FILE EXTERNAL 테이블을 생성하며, 파일을 복사하지 않습니다. 다음 예시는 각 파일의 파일 이름, 메타데이터, FILE 참조가 포함된 테이블을 생성합니다documents:

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

FILE MANAGED로 외부 소스 파일을 인징하기.

SharePoint, Google Drive, SFTP와 같은 소스에서 파일의 참조를 생성 FILE 하려면 먼저 파일을 인징하여 FILE MANAGED. FILE EXTERNAL 볼륨 외부에 저장된 파일에는 지원되지 않습니다.

다음 예시는 SharePoint에서 파일을 테이블로 FILE MANAGED 인제스트합니다:

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

파이프라인을 사용해 새로운 파일을 점진적으로 인징합니다

새 파일이 도착할 때마다 인징하려면, Lakeflow 파이프라인에서 소스 STREAM read_files(..., format => 'file')를 읽는 스트리밍 테이블을 사용하세요. 각 파이프라인 업데이트는 마지막 업데이트 이후 추가된 파일만 처리합니다. 선언적 파이프라인을 참고 read_files 하고 점화하세요.

Google Drive와 같은 소스에서 파일을 점진적으로 스트리밍하려면:

파이프라인 채널을 .로 PREVIEW설정하세요. 파이프라인에서 참조를 인제스트하려면 FILE 채널이 필요합니다.PREVIEW

소스를 STREAM read_files(..., format => 'file')읽는 스트리밍 테이블을 정의하라: 다음 코드에서:

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

AUTO CDC에 업데이트 및 삭제를 적용하세요

스트리밍 인제스트는 새 파일을 추가하지만 소스에서 업데이트나 삭제를 캡처하지는 않습니다. 변경 사항을 적용하려면 소스 변경 피드 AUTO CDC를 읽으세요.

경고

Databricks는 변경 데이터를 먼저 관리되는 테이블에 넣고, 다음 예시와 같이 그 테이블에 적용 AUTO CDC 할 것을 권장합니다. 직접 적용하면 AUTO CDCSTREAM read_files(..., readChangeFeed => true) 각 하류 흐름의 소스 변경 피드를 다시 읽게 되어 처리 비용이 증가할 수 있습니다.

변경 피드를 두 단계로 인징하세요. 다음 예시는 SharePoint에서 변경 피드를 인징한 후, 이를 대상 스트리밍 테이블에 SCD 타입 1로 적용합니다:

변경 데이터를 관리 파일이 포함된 스트리밍 테이블에 다음 코드와 같이 쓰세요. 변경 read_files 피드를 반환하도록 설정 readChangeFeed => true 하여 , _is_deleted_sequence메타데이터 열을 포함_file_id합니다.

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

다음 코드처럼 해당 테이블에서 변경 사항을 타겟 스트리밍 테이블에 적용하는 데 사용 AUTO CDC 하세요. 키를 사용하고, _sequence 시퀀스 열로 사용 _file_id 하며, _is_deleted 결손을 식별할 수 있습니다.

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
)

인라인 바이너리 데이터를 FILE 참조로 변환하기

테이블이 이미 파일 내용을 인라인 이진 데이터로 저장하고 있다면, 함수를 사용해 create_file 해당 데이터를 저장소에 쓰고 참조를 FILE 생성합니다.

다음 예시들은 사용자가 생성한 테이블 를 사용하며, raw_documents열과 namecontent 이진 데이터를 담은 열을 사용합니다.

볼륨에 이진 데이터를 FILE EXTERNAL 형식으로 쓰기

Unity 카탈로그 볼륨에 파일을 외부 파일로 쓰려면, 다음 코드와 같이 a destination_pathcreate_file를 에 전달하세요:

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

관리 저장소에 이진 데이터를 FILE MANAGED로 쓰기

관리 파일 create_file 로 저장하려면 이진 콘텐츠만 입력하면 됩니다. 를 누락 destination_path하면 Unity 카탈로그가 콘텐츠를 관리 저장 위치로 업로드합니다:

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

다음 단계

FILE 유형
FILE 타입과 비정형 데이터
튜토리얼: FILE 타입으로 파일 처리 파이프라인을 구축하세요
자동 로더에 대해 자세히 알아봅니다. 오토 로더란 무엇인가?를 참조하세요.

피드백

이 페이지가 도움이 되었나요?

Last updated on 2026-08-03

파일 형식으로 파일 인제스트

스토리지 모드

파일 발견에 사용하기 list_files

파일 참조로 파일을 인제스트하기

FILE EXTERNAL 파일 파일을 인제스트하기

FILE MANAGED로 외부 소스 파일을 인징하기.

SQL

Python

Scala

파이프라인을 사용해 새로운 파일을 점진적으로 인징합니다

SQL

Python

AUTO CDC에 업데이트 및 삭제를 적용하세요

SQL

Python

SQL

Python

인라인 바이너리 데이터를 FILE 참조로 변환하기

볼륨에 이진 데이터를 FILE EXTERNAL 형식으로 쓰기

SQL

Python

Scala

관리 저장소에 이진 데이터를 FILE MANAGED로 쓰기

SQL

Python

Scala

다음 단계

피드백

추가 리소스

파일 발견에 사용하기 `list_files`