deltdask

将Delta Lake表读取到Dask DataFrame的连接器。

使用pip install deltdask安装。

如下所示读取Delta Lake到Dask DataFrame

import deltadask

ddf = deltadask.read_delta("path/to/delta/table")

与Parquet数据湖相比，Dask读取Delta表效率更高。当然，Parquet表的速度比CSV表快如本视频所述。Delta表是Dask用户性能的下一个重大改进。

基本用法

假设您有一个包含以下三个版本的Delta表。

以下是读取Delta表的最新版本的方法

deltadask.read_delta("path/to/delta/table").compute()

   id
0   7
1   8
2   9

以下是读取Delta表的版本1的方法

deltadask.read_delta("path/to/delta/table", version=1).compute()

   id
0   0
1   1
2   2
0   4
1   5

Delta Lake使Dask能够轻松地在Delta表的各个版本之间进行时间旅行。

请参阅此笔记本以获取完整的工作示例，以便您可以在自己的机器上复制。

为什么Delta Lake比Parquet更适合Dask

Delta表将数据存储在Parquet文件中，并将元数据存储在事务日志中。元数据包括模式和文件的存储位置。

Dask Parquet数据湖可以以两种不同的方式存储。

1. 带有单个元数据文件的Parquet文件
2. 没有元数据文件的Parquet文件

带有单个元数据文件的Parquet文件由于扩展性限制而不好。

没有元数据文件的Parquet文件需要昂贵的文件列表/ Parquet尾部查询来构建数据湖的整体元数据统计信息。从Delta事务日志中获取这些信息要快得多。

Delta Lake更好，因为事务日志可扩展，并且可以比数据湖快得多地查询。

如何使读取更快

您可以通过使用列投影和谓词下推过滤来加快Delta Lake查询的速度。这些策略允许您向集群发送更少的数据。

以下是一个使用Dask查询Delta表并利用列剪枝和谓词下推过滤的示例。

ddf = deltadask.read_delta(
    "path/to/delta/table", 
    columns=["col1"], # column pruning
    filters=[[('col1', '==', 0)]] # predicate pushdown
)

此查询只将col1发送到集群，而不会发送其他任何列（列投影）。

此查询还使用事务日志来识别至少包含一些col1等于零的数据的文件。如果一个文件不包含匹配的数据，则不会读取该文件。根据数据的组织方式，可以跳过很多文件。您还可以通过分区或Z ORDERING数据来进一步跳过文件数量。

此库的构建方式

delta-rs库使得构建deltadask连接器变得非常容易。

所有事务日志解析逻辑都由delta-rs处理。您只需连接到API，就可以轻松构建Dask连接器。