开始使用 COPY INTO 加载数据

可以使用 COPY INTO SQL 命令将数据从文件位置加载到 Delta 表中。 COPY INTO 是可重试的且幂等 — 在后续运行中，会跳过已经加载的源位置中的文件。

COPY INTO 提供这些功能：

云存储中可以轻松配置的文件或文件夹筛选器，包括 S3、ADLS、ABFS、GCS 和 Unity Catalog 数据库卷册。
支持多种源文件格式：CSV、JSON、XML、 Avro、 ORC、 Parquet、文本和二进制文件。
默认情况下，恰好一次（幂等）文件处理。
目标表架构推理、映射、合并和演变。

注意

为了获得更具可缩放且可靠的文件引入体验，Databricks 建议 SQL 用户使用流式处理表。有关详细信息，请参阅流式处理表。

Prerequisites

帐户管理员必须按照 “配置数据访问” 中的步骤配置使用云对象存储中的数据的访问权限，然后用户才能在使用 COPY INTO 时加载数据。

将数据加载到无架构 Delta Lake 表中

在 Databricks Runtime 11.3 LTS 及更高版本中，可以创建空的占位符 Delta 表，以便在 COPY INTO 中通过将 mergeSchema 设置为 true 来使用 COPY_OPTIONS 命令推断架构。以下示例使用 Wanderbricks 数据集。将<catalog>、<schema>和<volume>替换为你拥有CREATE TABLE权限的目录、架构和卷。

SQL

CREATE TABLE IF NOT EXISTS <catalog>.<schema>.booking_updates_schemaless;

COPY INTO <catalog>.<schema>.booking_updates_schemaless
FROM '/Volumes/<catalog>/<schema>/<volume>/wanderbricks/booking_updates'
FILEFORMAT = JSON
FORMAT_OPTIONS ('mergeSchema' = 'true', 'multiLine' = 'true')
COPY_OPTIONS ('mergeSchema' = 'true');

Python

table_name = '<catalog>.<schema>.booking_updates_schemaless'
source_data = '/Volumes/<catalog>/<schema>/<volume>/wanderbricks/booking_updates'
source_format = 'JSON'

spark.sql("CREATE TABLE IF NOT EXISTS " + table_name)

spark.sql("COPY INTO " + table_name + \
  " FROM '" + source_data + "'" + \
  " FILEFORMAT = " + source_format + \
  " FORMAT_OPTIONS ('mergeSchema' = 'true', 'multiLine' = 'true')" + \
  " COPY_OPTIONS ('mergeSchema' = 'true')"
)

R

library(SparkR)
sparkR.session()

table_name = "<catalog>.<schema>.booking_updates_schemaless"
source_data = "/Volumes/<catalog>/<schema>/<volume>/wanderbricks/booking_updates"
source_format = "JSON"

sql(paste("CREATE TABLE IF NOT EXISTS ", table_name, sep = ""))

sql(paste("COPY INTO ", table_name,
  " FROM '", source_data, "'",
  " FILEFORMAT = ", source_format,
  " FORMAT_OPTIONS ('mergeSchema' = 'true', 'multiLine' = 'true')",
  " COPY_OPTIONS ('mergeSchema' = 'true')",
  sep = ""
))

Scala(编程语言)

val table_name = "<catalog>.<schema>.booking_updates_schemaless"
val source_data = "/Volumes/<catalog>/<schema>/<volume>/wanderbricks/booking_updates"
val source_format = "JSON"

spark.sql("CREATE TABLE IF NOT EXISTS " + table_name)

spark.sql("COPY INTO " + table_name +
  " FROM '" + source_data + "'" +
  " FILEFORMAT = " + source_format +
  " FORMAT_OPTIONS ('mergeSchema' = 'true', 'multiLine' = 'true')" +
  " COPY_OPTIONS ('mergeSchema' = 'true')"
)

此 SQL 语句是幂等的。这意味着你可以计划它重复运行，并且只会将新数据加载到 Delta 表中。

注意

在COPY INTO之外，空 Delta 表无法使用。 INSERT INTO 和 MERGE INTO 不支持将数据写入无架构 Delta 表。使用 COPY INTO 将数据插入到表中后，该表将变为可查询。

请参阅为 COPY INTO 创建目标表。

设置模式并将数据加载到 Delta Lake 表中

以下示例创建 Delta 表，并使用 COPY INTO SQL 命令将 Wanderbricks 数据集中的示例数据加载到表中。源文件是存储在 Unity 目录卷中的 JSON 文件。可以从附加到Azure Databricks群集的笔记本中运行示例Python、R、Scala 或 SQL 代码。还可以从 Databricks SQL 中与 SQL 仓库关联的查询运行 SQL 代码。将<catalog>、<schema>和<volume>替换为你拥有CREATE TABLE权限的目录、架构和卷。

SQL

DROP TABLE IF EXISTS <catalog>.<schema>.booking_updates_upload;

CREATE TABLE <catalog>.<schema>.booking_updates_upload (
  booking_id BIGINT,
  user_id BIGINT,
  status STRING,
  total_amount DOUBLE
);

COPY INTO <catalog>.<schema>.booking_updates_upload
FROM '/Volumes/<catalog>/<schema>/<volume>/wanderbricks/booking_updates'
FILEFORMAT = JSON
FORMAT_OPTIONS ('multiLine' = 'true');

SELECT * FROM <catalog>.<schema>.booking_updates_upload;

Python

table_name = '<catalog>.<schema>.booking_updates_upload'
source_data = '/Volumes/<catalog>/<schema>/<volume>/wanderbricks/booking_updates'
source_format = 'JSON'

spark.sql("DROP TABLE IF EXISTS " + table_name)

spark.sql("CREATE TABLE " + table_name + " (" \
  "booking_id BIGINT, " + \
  "user_id BIGINT, " + \
  "status STRING, " + \
  "total_amount DOUBLE)"
)

spark.sql("COPY INTO " + table_name + \
  " FROM '" + source_data + "'" + \
  " FILEFORMAT = " + source_format + \
  " FORMAT_OPTIONS ('multiLine' = 'true')"
)

booking_updates_upload_data = spark.sql("SELECT * FROM " + table_name)

display(booking_updates_upload_data)

R

library(SparkR)
sparkR.session()

table_name = "<catalog>.<schema>.booking_updates_upload"
source_data = "/Volumes/<catalog>/<schema>/<volume>/wanderbricks/booking_updates"
source_format = "JSON"

sql(paste("DROP TABLE IF EXISTS ", table_name, sep = ""))

sql(paste("CREATE TABLE ", table_name, " (",
  "booking_id BIGINT, ",
  "user_id BIGINT, ",
  "status STRING, ",
  "total_amount DOUBLE)",
  sep = ""
))

sql(paste("COPY INTO ", table_name,
  " FROM '", source_data, "'",
  " FILEFORMAT = ", source_format,
  " FORMAT_OPTIONS ('multiLine' = 'true')",
  sep = ""
))

booking_updates_upload_data = tableToDF(table_name)

display(booking_updates_upload_data)

Scala(编程语言)

val table_name = "<catalog>.<schema>.booking_updates_upload"
val source_data = "/Volumes/<catalog>/<schema>/<volume>/wanderbricks/booking_updates"
val source_format = "JSON"

spark.sql("DROP TABLE IF EXISTS " + table_name)

spark.sql("CREATE TABLE " + table_name + " (" +
  "booking_id BIGINT, " +
  "user_id BIGINT, " +
  "status STRING, " +
  "total_amount DOUBLE)"
)

spark.sql("COPY INTO " + table_name +
  " FROM '" + source_data + "'" +
  " FILEFORMAT = " + source_format +
  " FORMAT_OPTIONS ('multiLine' = 'true')"
)

val booking_updates_upload_data = spark.table(table_name)

display(booking_updates_upload_data)

若要清理，请运行以下代码以删除示例表。

SQL

DROP TABLE <catalog>.<schema>.booking_updates_upload

Python

spark.sql("DROP TABLE " + table_name)

R

sql(paste("DROP TABLE ", table_name, sep = ""))

Scala(编程语言)

spark.sql("DROP TABLE " + table_name)

清除元数据文件

可以运行 VACUUM 来清理 Databricks Runtime 15.2 及更高版本中由 COPY INTO 创建的未引用的元数据文件。

其他资源

使用 COPY INTO 和 Unity Catalog 卷或外部位置加载数据
使用 COPY INTO. 的常见数据加载模式。
Databricks Runtime 7.x 及更高版本： COPY INTO

上次更新时间为 2026-10-23