架构实施

Azure Databricks 通过对 Delta Lake 表实施写入时架构强制检查来确保数据质量。 架构强制执行不适用于使用非 Delta 格式的表，例如云存储中的 CSV 或 JSON 文件。

针对 INSERT 操作的架构强制执行

Azure Databricks将数据插入表中时强制实施以下规则：

• 所有插入的列都必须存在于目标表中。
• 所有列数据类型必须与目标表中的列数据类型匹配。

INSERT 示例

以下示例将数据写入名为 enforce_demo 的托管 Delta Lake 表。 要创建它，请运行以下操作：

CREATE OR REPLACE TABLE main.default.enforce_demo (id INT, name STRING, amount BIGINT);

以下 INSERT 不成立，因为 unknown_column 不存在于 enforce_demo。 Azure Databricks 返回UNRESOLVED_COLUMN.WITH_SUGGESTION错误（SQLSTATE 42703），其中会提示有效的列名：

INSERT INTO main.default.enforce_demo (id, unknown_column) VALUES (1, 'value');

以下 INSERT 成功执行。 Azure Databricks 安全地将整数 42 转换为 amount 列的 BIGINT 类型：

INSERT INTO main.default.enforce_demo (id, amount) VALUES (1, 42);

针对 MERGE 操作的架构强制执行

Azure Databricks在插入或更新数据作为 MERGE 操作的一部分时强制实施以下规则：

• 如果源语句中的数据类型与目标列不匹配，MERGE 会尝试安全地强制转换列数据类型以匹配目标表。
• UPDATE 或 INSERT 操作的目标列必须存在于目标表中。
• 使用 INSERT * 或 UPDATE SET *：
  • 源数据集必须包含目标表中存在的所有列。
  • 强制执行会忽略源数据集中目标表里不存在的列。

MERGE 示例

以下示例复用了上一节中的 enforce_demo 表，以及一个名为 enforce_source 且多了一个列的源表。 要创建源表，请执行以下操作：

CREATE OR REPLACE TABLE main.default.enforce_source (id INT, name STRING, amount BIGINT, extra_col STRING);

INSERT INTO main.default.enforce_source VALUES (1, 'Alice', 100, 'x'), (2, 'Bob', 200, 'y');

以下 MERGE 失败了，因为它给 unknown_column 赋值，而 unknown_column 在 enforce_demo 中不存在。 Azure Databricks 返回一个 DELTA_MERGE_UNRESOLVED_EXPRESSION 错误，其中会列出其可解析的列名：

MERGE INTO main.default.enforce_demo AS t
USING main.default.enforce_source AS s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.unknown_column = s.name
WHEN NOT MATCHED THEN INSERT (id, unknown_column) VALUES (s.id, s.name);

enforce_source该表包含一extra_col列，但enforce_demo没有。 以下使用 INSERT * 的 MERGE 之所以成功，是因为源中包含所有目标列。 执法部门 extra_col忽视：

MERGE INTO main.default.enforce_demo AS t
USING main.default.enforce_source AS s
ON t.id = s.id
WHEN NOT MATCHED THEN INSERT *;

修改表架构

可以使用显式 ALTER TABLE 语句或自动架构演变来更新表的架构。 请参阅 更新具有架构演变的表架构。

例如，若要显式添加一列：

ALTER TABLE catalog.schema.table_name ADD COLUMN new_column STRING;

若要为写入操作启用自动架构演进，请设置 mergeSchema 选项：

SQL

SET spark.databricks.delta.schema.autoMerge.enabled = true;
INSERT INTO catalog.schema.table_name SELECT * FROM source_table;

Python

df.write.option("mergeSchema", "true").mode("append").saveAsTable("catalog.schema.table_name")

架构演变对于 INSERT 和 MERGE 操作具有特殊的语义。 请参阅启用架构演变。

外部表

如果您在 Azure Databricks 之外直接使用外部客户端修改外部表的元数据，或通过基于路径的访问方式进行修改，Unity Catalog 不会自动将这些更新同步到其架构中。 这可能会导致架构约束无法正确生效。

运行 MSCK REPAIR TABLE <table-name> SYNC METADATA 以将架构与 Unity 目录同步。 请参阅 REPAIR TABLE。