将 Unity Catalog 与增量实时表管道配合使用

Databricks 建议使用 Unity Catalog 配置增量实时表管道。

配置有 Unity Catalog 的管道将所有定义的具体化视图和流式处理表发布到指定的目录和架构。 Unity Catalog 管道可以从其他 Unity Catalog 表和卷中读取。

若要管理对 Unity Catalog 管道所创建的表的权限，请使用 GRANT 和 REVOKE。

要求

通过增量实时表管道在 Unity Catalog 中创建表需满足以下权限：

• 目标目录中的 USE CATALOG 特权。
• 如果管道创建具体化视图，目标架构中需有 CREATE MATERIALIZED VIEW 和 USE SCHEMA 特权。
• 如果管道创建流式表，目标架构中需有 CREATE TABLE 和 USE SCHEMA 特权。
• 如果管道设置中未指定目标架构，你必须对目标目录中的至少一个架构拥有 CREATE MATERIALIZED VIEW 或 CREATE TABLE 特权。

运行已启用 Unity Catalog 的管道所需的计算：

• 共享访问模式群集。 无法在单用户（“已分配”）群集上运行已启用 Unity Catalog 的管道。 请参阅 Unity Catalog 上的共享访问模式限制。

使用 Unity Catalog（包括流式处理表和具体化视图）查询增量实时表管道创建的表所需的计算包括以下任一项：

• SQL 仓库

• Databricks Runtime 13.3 LTS 或更高版本中的共享访问模式群集。

• 单用户（或“已分配”）访问模式群集，如果在单用户群集上已启用精细访问控制（即，群集在 Databricks Runtime 15.4 或更高版本上运行，且已为工作区启用无服务器计算）。 有关详细信息，请参阅单用户计算上的精细访问控制。

• 仅在表所有者运行查询时，13.3 LTS 到 15.3 上的单用户（或“已分配”）访问模式群集。

适用的其他计算限制。 请参阅下一节。

限制

下面是将 Unity Catalog 与增量实时表配合使用时的限制：

• 默认情况下，只有管道所有者和工作区管理员可以从运行启用了 Unity Catalog 的管道的群集中查看驱动程序日志。 若要允许其他用户访问驱动程序日志，请参阅允许非管理员用户从启用了 Unity 目录的管道查看驱动程序日志。

• 使用 Hive 元存储的现有管道无法升级为使用 Unity Catalog。 若要迁移写入到 Hive 元存储的现有管道，必须创建新管道，并从数据源重新引入数据。

• 不得在附加到元存储（在 Unity Catalog 公共预览期间创建）的工作区中创建已启用 Unity Catalog 的管道。 请参阅升级到特权继承。

• 不支持 JAR。 仅支持第三方 Python 库。 请参阅管理增量实时表管道的 Python 依赖项。

• 不支持修改流式表架构的数据操作语言(DML)查询。

• 在增量实时表管道中创建的具体化视图不能用作该管道外部（例如，在另一个管道或下游笔记本中）的流式处理源。

• 如果某个管道发布到具有托管存储位置的架构，则可以在后续更新中更改架构，但唯一的前提是更新的架构使用与之前指定的架构相同的存储位置。

• 表存储在目标架构的存储位置。 如果未指定架构存储位置，则表存储在目录存储位置。 如果未指定架构和目录存储位置，表将存储在元存储的根存储位置。

• 目录资源管理器“历史记录”选项卡不显示具体化视图的历史记录。

• 定义表时不支持 LOCATION 属性。

• 已启用 Unity Catalog 的管道无法发布到 Hive 元存储。

• Python UDF 支持为公共预览版。

• 不能将 Delta Sharing 用于发布到 Unity Catalog 的增量实时表具体化视图或流式表。

• 不能在管道或查询中使用 event_log 表值函数来访问多个管道的事件日志。

• 不能与其他用户共享基于 event_log 表值函数创建的视图。

• 已启用 Unity Catalog 的管道不支持单节点群集。 由于增量实时表可能会创建单节点群集以运行较小的管道，因此管道可能会失败并显示引用single-node mode的错误消息。 如果发生这种情况，在配置计算时至少指定一个辅助角色。 请参阅配置增量实时表管道的计算。

是否可以结合使用 Hive 元存储和 Unity Catalog 管道？

工作区可以包含使用 Unity Catalog 和旧版 Hive 元存储的管道。 但是，单个管道无法写入 Hive 元存储和 Unity Catalog。 写入 Hive 元存储的现有管道无法升级为使用 Unity Catalog。

未使用 Unity Catalog 的现有管道不受配置有 Unity Catalog 的新管道的影响。 这些管道继续使用配置的存储位置，将数据保存到 Hive 元存储。

除非本文档中另有说明，否则使用 Unity Catalog 的管道支持所有现有的数据源和增量实时表功能。 使用 Unity Catalog 的管道支持 Python 和 SQL 接口。

对现有功能的更改

当增量实时表配置为将数据保存到 Unity Catalog 时，表的生命周期由增量实时表管道管理。 因为管道管理表生命周期和权限：

• 从增量实时表管道定义中删除某个表时，相应的具体化视图或流式表条目会在下一次管道更新时从 Unity Catalog 中删除。 实际数据会保留一段时间，以便误删时可以恢复。 可通过将具体化视图或 > 添加回管道定义来恢复数据。
• 删除增量实时表管道会导致删除该管道中定义的所有表。 由于此项更改，增量实时表 UI 已更新，会提示是否确认删除管道。
• 用户无法直接访问内部支持表，包括用于支持 APPLY CHANGES INTO 的内部支持表。

从增量实时表管道将表写入 Unity Catalog

要将表写入 Unity Catalog，必须配置管道，以便通过工作区与之协同工作。 创建管道时，请在“存储选项”下选择“Unity Catalog”，在“目录”下拉菜单中选择一个目录，然后在“目标架构”下拉菜单中选择一个现有架构或输入新架构的名称。 若要了解 Unity Catalog 目录，请参阅 Azure Databricks 中的目录是什么？。 若要了解 Unity Catalog 中的架构，请参阅 Azure Databricks 中的架构是什么？。

将数据引入 Unity Catalog 管道

配置为使用 Unity Catalog 的管道可从以下位置读取数据：

• Unity Catalog 托管和外部表、视图、具体化视图和流式表。
• Hive 元存储表和视图。
• 自动加载程序使用 read_files() 函数从 Unity Catalog 外部位置读取数据。
• Apache Kafka 和 Amazon Kinesis。

下面是从 Unity Catalog 和 Hive 元存储表读取数据的示例。

从 Unity Catalog 表批量引入

SQL

CREATE OR REFRESH MATERIALIZED VIEW
  table_name
AS SELECT
  *
FROM
  my_catalog.my_schema.table1;

Python

@dlt.table
def table_name():
  return spark.read.table("my_catalog.my_schema.table")

流式处理 Unity Catalog 表中的更改

SQL

CREATE OR REFRESH STREAMING TABLE
  table_name
AS SELECT
  *
FROM
  STREAM(my_catalog.my_schema.table1);

Python

@dlt.table
def table_name():
  return spark.readStream.table("my_catalog.my_schema.table")

从 Hive 元存储引入数据

使用 Unity Catalog 的管道可以使用 hive_metastore 目录从 Hive 元存储表读取数据：

SQL

CREATE OR REFRESH MATERIALIZED VIEW
  table_name
AS SELECT
  *
FROM
  hive_metastore.some_schema.table;

Python

@dlt.table
def table3():
  return spark.read.table("hive_metastore.some_schema.table")

从自动加载程序引入数据

SQL

CREATE OR REFRESH STREAMING TABLE
  table_name
AS SELECT
  *
FROM
  read_files(
    <path-to-uc-external-location>,
    "json"
  )

Python

@dlt.table(table_properties={"quality": "bronze"})
def table_name():
  return (
     spark.readStream.format("cloudFiles")
     .option("cloudFiles.format", "json")
     .load(f"{path_to_uc_external_location}")
 )

共享具体化视图

默认情况下，只有管道所有者有权查询管道创建的数据集。 可以使用 GRANT 语句授予其他用户查询表的能力，并可以使用 REVOKE 语句撤销查询访问权限。 有关 Unity Catalog 中的特权的详细信息，请参阅管理 Unity Catalog 中的特权。

授予对表的 SELECT 权限

GRANT SELECT ON TABLE
  my_catalog.my_schema.table_name
TO
  `user@databricks.com`

撤销对表的 SELECT 权限

REVOKE SELECT ON TABLE
  my_catalog.my_schema.table_name
FROM
  `user@databricks.com`

授予创建表或创建具体化视图的权限

GRANT CREATE { MATERIALIZED VIEW | TABLE } ON SCHEMA
  my_catalog.my_schema
TO
  { principal | user }

查看管道的世系

增量实时表管道中表的世系在目录资源管理器中可见。 对于已启用 Unity Catalog 的管道中的具体化视图或流式处理表，目录资源管理器世系 UI 会显示上游表和下游表。 若要详细了解 Unity Catalog 世系，请参阅捕获和查看 Unity Catalog 的数据世系。

对于已启用 Unity Catalog 的增量实时表管道中的具体化视图或流式表，如果可以从当前工作区访问管道，则目录资源管理器世系 UI 还会链接到生成了具体化视图或流式表的管道。

在流式表中添加、更改或删除数据

可使用数据操作语言 (DML) 语句（包括插入、更新、删除和合并语句）来修改发布到 Unity Catalog 的流式处理表。 支持对流式处理表进行 DML 查询，这可实现一些用例，例如更新表以使其符合一般数据保护条例 (GDPR)。

下面是用于修改流式表中的记录的 DML 语句示例。

删除具有特定 ID 的记录：

DELETE FROM my_streaming_table WHERE id = 123;

更新具有特定 ID 的记录：

UPDATE my_streaming_table SET name = 'Jane Doe' WHERE id = 123;

发布具有行筛选器和列掩码的表

行筛选器允许指定在表扫描提取行时作为筛选器应用的函数。 这些筛选器可确保后续查询仅返回筛选器谓词的计算结果为 true 的行。

每当表扫描提取行时，列掩码就会让你将列值掩码。 该列的将来查询将返回被评估函数的结果，而不是该列的原始值。 有关使用行筛选器和列掩码的详细信息，请参阅使用行筛选器和列掩码筛选敏感表数据。

管理行筛选器和列掩码

在具体化视图和流式处理表上，应通过 CREATE OR REFRESH 语句添加、更新或删除行筛选器和列掩码。

有关定义具有行筛选器和列掩码的表的详细语法，请参阅增量实时表 SQL 语言参考和增量实时表 Python 语言参考。

行为

以下是在 Delta Live Tables 管道中使用行筛选器或列掩码时的重要详细信息：

• 以所有者身份刷新：当管道更新刷新具体化视图或流式表时，行筛选器和列掩码函数以管道所有者权限运行。 这意味着表刷新使用创建管道的用户的安全上下文。 系统使用管道所有者的用户上下文计算检查用户上下文的函数（例如 CURRENT_USER 和 IS_MEMBER）。
• 查询：查询具体化视图或流式表时，系统使用调用程序用户上下文计算检查用户上下文的函数（例如 CURRENT_USER 和 IS_MEMBER）。 此方法基于当前用户的上下文强制实施特定于用户的数据安全和访问控制。
• 在包含行筛选器和列掩码的源表上创建具体化视图时，具体化视图的刷新始终是完全刷新。 完全刷新使用最新定义重新处理源中的所有可用数据。 此过程检查是否使用了最新的数据和定义来评估和应用源表上的安全策略。

可观察性

使用 DESCRIBE EXTENDED、INFORMATION_SCHEMA 或目录资源管理器检查应用于给定具体化视图或流式处理表的现有行筛选器和列掩码。 此功能允许用户审核和查看有关具体化视图和流式处理表的数据访问和保护措施。