Presto、Trino 和 Athena 使用清单与 Delta Lake 集成

重要

Presto、Trino 和 Athena 都原生支持 Delta Lake。支持情况如下：

• Presto 版本 0.269 及更高版本原生支持读取 Delta Lake 表。有关使用原生 Delta Lake 连接器的详细信息，请参阅 Delta Lake 连接器 - Presto。对于低于 0.269 的 Presto 版本，您可以使用本文中基于清单的方法。
• Trino 版本 373 及更高版本原生支持读取和写入 Delta Lake 表。有关使用原生 Delta Lake 连接器的详细信息，请参阅 Delta Lake 连接器 - Trino。对于低于 版本 373 的 Trino 版本，您可以使用本文中详细介绍的基于清单的方法。
• Athena 版本 3 及更高版本原生支持读取 Delta Lake 表。有关使用原生 Delta Lake 连接器的详细信息，请参阅 查询 Delta Lake 表。对于低于 版本 3 的 Athena 版本，您可以使用本文中详细介绍的基于清单的方法。

Presto、Trino 和 Athena 支持使用清单文件读取外部表，清单文件是包含用于查询表的数据文件列表的文本文件。当在 Hive metastore 中使用清单文件定义外部表时，Presto、Trino 和 Athena 可以使用清单中的文件列表，而不是通过目录列表查找文件。本文介绍如何使用清单文件设置 Presto、Trino 和 Athena 与 Delta Lake 的集成以及查询 Delta 表。

设置 Presto、Trino 或 Athena 与 Delta Lake 的集成并查询 Delta 表

您可以通过以下步骤设置 Presto、Trino 或 Athena 与 Delta Lake 的集成。

步骤 1：使用 Apache Spark 生成 Delta 表的清单

使用已配置了 Delta Lake 的 Spark，在位于 <path-to-delta-table> 的 Delta 表上运行以下任何命令。

SQL

GENERATE symlink_format_manifest FOR TABLE delta.`<path-to-delta-table>`

Scala

val deltaTable = DeltaTable.forPath(<path-to-delta-table>)
deltaTable.generate("symlink_format_manifest")

Java

DeltaTable deltaTable = DeltaTable.forPath(<path-to-delta-table>);
deltaTable.generate("symlink_format_manifest");

Python

deltaTable = DeltaTable.forPath(<path-to-delta-table>)
deltaTable.generate("symlink_format_manifest")

有关详细信息，请参阅生成清单文件。

generate 命令在 <path-to-delta-table>/_symlink_format_manifest/ 生成清单文件。换句话说，此目录中的文件将包含应读取的数据文件（即 Parquet 文件）的名称，以读取 Delta 表的快照。

步骤 2：配置 Presto、Trino 或 Athena 以读取生成的清单

1. 使用格式 SymlinkTextInputFormat 和清单位置 <path-to-delta-table>/_symlink_format_manifest/ 在连接到 Presto、Trino 或 Athena 的 Hive metastore 中定义一个新表。

CREATE EXTERNAL TABLE mytable ([(col_name1 col_datatype1, ...)])
[PARTITIONED BY (col_name2 col_datatype2, ...)]
ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'
STORED AS INPUTFORMAT 'org.apache.hadoop.hive.ql.io.SymlinkTextInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION '<path-to-delta-table>/_symlink_format_manifest/'  -- location of the generated manifest

SymlinkTextInputFormat 配置 Presto、Trino 或 Athena 通过读取清单文件而不是使用目录列表查找数据文件来计算 mytable 的文件拆分。将 mytable 替换为外部表的名称，将 <path-to-delta-table> 替换为 Delta 表的绝对路径。

重要

• mytable 必须与 Delta 表具有相同的模式和分区。
• PARTITIONED BY 列集必须与非分区列集不同。此外，您不能使用 AS <select-statement> 指定分区列。

• 您不能在 Apache Spark 中使用此表定义；它只能由 Presto、Trino 和 Athena 使用。

  您用于运行命令的工具取决于 Apache Spark 和 Presto、Trino 或 Athena 是否使用相同的 Hive metastore。

  • 相同的 metastore：如果 Apache Spark 和 Presto、Trino 或 Athena 使用相同的 Hive metastore，您可以使用 Apache Spark 定义表。

• 不同的 metastore：如果 Apache Spark 和 Presto、Trino 或 Athena 使用不同的 metastore，您必须使用其他工具定义表。

  • Athena：您可以在 Athena 中定义外部表。
  • Presto：Presto 不支持语法 CREATE EXTERNAL TABLE ... STORED AS ...，因此您必须使用连接到与 Presto 相同 metastore 的其他工具（例如 Spark 或 Hive）来创建表。

2. 如果 Delta 表已分区，请在生成清单后运行 MSCK REPAIR TABLE mytable 以强制 metastore（连接到 Presto、Trino 或 Athena）发现分区。这是必需的，因为分区表的清单本身以与表相同的目录结构进行分区。使用创建表的相同工具运行此命令。此外，您应该运行此命令

   • 每次清单生成后：新分区很可能在清单文件更新后立即可见。但是，过于频繁地执行此操作可能会导致 Hive metastore 负载过高。
   • 与预期新分区的频率一样频繁：例如，如果一个表按日期分区，那么您可以在每天午夜后、在新分区在表中创建并且其相应的清单文件已生成后运行一次修复。

步骤 3：更新清单

当 Delta 表中的数据更新时，您必须使用以下任一方法重新生成清单

• 显式更新：所有数据更新后，您可以运行 generate 操作来更新清单。

• 自动更新：您可以配置 Delta 表，以便对表的所有写入操作都会自动更新清单。要启用此自动模式，请使用以下 SQL 命令设置相应的表属性。

ALTER TABLE delta.`<path-to-delta-table>` SET TBLPROPERTIES(delta.compatibility.symlinkFormatManifest.enabled=true)

要禁用此自动模式，请将此属性设置为 false。此外，对于分区表，您必须运行 MSCK REPAIR 以确保连接到 Presto、Trino 或 Athena 的 metastore 更新分区。

注意

在分区表上启用自动模式后，每个写入操作仅更新与该操作写入的分区对应的清单。这种增量更新确保了写入操作的清单生成开销较低。但是，这也意味着如果其他分区中的清单已过期，启用自动模式将不会自动修复它。因此，建议您在启用自动模式后立即明确运行 GENERATE 以更新整个表的清单。

是自动更新还是显式更新取决于 Delta 表上写入操作的并发性质和所需的数据一致性。例如，如果启用自动模式，并发写入操作会导致清单文件的并发覆盖。在这种无序写入的情况下，写入操作完成后，清单文件不保证指向表的最新版本。因此，如果预期有并发写入并且您想避免过时清单，您应该考虑在预期的写入操作完成后显式更新清单。

限制

Presto、Trino 和 Athena 集成在行为上存在已知的限制。

数据一致性

每当 Delta Lake 生成更新的清单时，它都会原子地覆盖现有清单文件。因此，Presto、Trino 和 Athena 将始终看到数据文件的一致视图；它将看到所有旧版本文件或所有新版本文件。但是，一致性保证的粒度取决于表是否已分区。

• 未分区表：所有文件名都写入一个清单文件，该文件是原子更新的。在这种情况下，Presto、Trino 和 Athena 将看到完整的表快照一致性。
• 分区表：清单文件以与原始 Delta 表相同的 Hive 分区样式目录结构进行分区。这意味着每个分区都是原子更新的，Presto、Trino 或 Athena 将看到每个分区的一致视图，但不会看到跨分区的一致视图。此外，由于所有分区的清单无法一起更新，并发尝试生成清单可能导致不同分区具有不同版本的清单。虽然在数据更改下这种一致性保证弱于使用 Spark 读取 Delta 表，但它仍然强于 Parquet 等格式，因为它们不提供分区级别的一致性。

根据您用于 Delta 表的存储系统，当 Presto、Trino 或 Athena 在清单文件被重写时并发查询清单时，可能会得到不正确的结果。在缺少原子文件覆盖的文件系统实现中，清单文件可能暂时不可用。因此，如果清单更新可能与 Presto、Trino 或 Athena 的查询同时发生，请谨慎使用清单。

性能

大量文件可能会损害 Presto、Trino 和 Athena 的性能。因此，建议您在生成清单之前压缩表的文件。文件数量不应超过 1000 个（对于整个未分区表或分区表中的每个分区）。

模式演变

Delta Lake 支持模式演进，并且对 Delta 表的查询会自动使用最新模式，无论 Hive metastore 中表中定义的模式如何。但是，Presto、Trino 或 Athena 使用 Hive metastore 中定义的模式，并且在重新定义 Presto、Trino 或 Athena 使用的表以具有更新的模式之前，不会使用更新的模式进行查询。

加密表

Athena 不支持读取 CSE-KMS 加密表的清单。有关最新信息，请参阅 AWS 文档。