xdlake

对deltalake的松散实现，并在pyarrow之上用Python编写，专注于可扩展性、可定制性和分布式数据。

这主要受到deltalake包的启发，并且（非常）未经实战检验。然而，由于其Pythonic设计，它更具灵活性。如果你对此感兴趣，不妨试试看，甚至可以帮助使其变得更好。

安装

pip install xdlake

用法

实例化

实例化一个表！这可以是本地或远程。对于远程，你可能需要安装相关的fsspec实现，例如为AWS S3、Google Storage和Azure Storage分别安装s3fs、gcsfs、adlfs。

dt = xdlake.DeltaTable("path/to/my/cool/local/table")
dt = xdlake.DeltaTable("s3://path/to/my/cool/table")
dt = xdlake.DeltaTable("az://path/to/my/cool/table", storage_options=dict_of_azure_creds)

读取

读取数据。对于复杂的过滤和谓词下推等，请使用to_pyarrow_dataset并了解如何过滤pyarrow数据集。

ds = dt.to_pyarrow_dataset()
t = dt.to_pyarrow_table()
df = dt.to_pandas()

写入

DeltaTable的实例是不可变的：任何执行表操作的函数都将返回一个新的DeltaTable。

写入内存数据

从内存中写入数据。数据可以是 pyarrow 表、数据集、记录批次、pandas DataFrame 或这些事物的可迭代对象。

dt = dt.write(my_cool_pandas_dataframe)
dt = dt.write(my_cool_arrow_table)
dt = dt.write(my_cool_arrow_dataset)
dt = dt.write(my_cool_arrow_record_batches)
dt = dt.write(pyarrow.Table.from_pandas(df))

导入外部数据

导入外部数据的引用而不进行复制。数据可以异构地存储在 s3、gs、azure 和本地，并且可以与 DeltaTable 本身不同的方式进行分区。尽情发挥。

如需为不同的存储位置提供不同的凭据，请参阅 凭据。

一次性从多个位置导入数据。此方法仅适用于非分区数据。

dt = dt.import_refs(["s3://some/aws/data", "gs://some/gcp/data", "az://some/azure/data" ])
dt = dt.import_refs(my_pyarrow_filesystem_dataset)

分区数据需要不同的处理方式。首先，您需要阅读有关 pyarrow 分区 的内容来完成此操作。其次，您一次只能导入一个数据集。

foreign_partitioning = pyarrow.dataset.partitioning(...)
ds = pyarrow.dataset.dataset(
    list_of_files,
    partitioning=foreign_partitioning,
    partition_base_dir,
    filesystem=xdlake.storage.get_filesystem(foreign_refs_loc),
)
dt = dt.import_refs(ds, partition_by=my_partition_cols)

删除

使用 pyarrow 表达式 从 DeltaTable 中删除行

import pyarrow.compute as pc
expr = (
    (pc.field("cats") == pc.scalar("A"))
    |
    (pc.field("float64") > pc.scalar(0.9))
)
dt = dt.delete(expr)

删除向量

我非常希望支持 删除向量，但 pyarrow 无法按行索引过滤 parquet 文件（如果你问我的话，这是很基本的）。如果你也希望 xdlake 支持删除向量，请通过在 这里 发声来告知 arrow 的人们。

克隆

您可以对 deltatable 进行克隆。这是一个软克隆（不复制数据，新表只引用数据）。整个版本历史都得到了保留。写入将被写入新位置。

cloned_dt = dt.clone("the/location/of/the/clone")

凭据

引用分布式数据的 DeltaTable 可能需要各种云位置的凭据。

为 s3、gs 等注册默认凭据。

xdlake.storage.register_default_filesystem_for_protocol("s3", s3_creds)
xdlake.storage.register_default_filesystem_for_protocol("gs", gs_creds)
xdlake.storage.register_default_filesystem_for_protocol("az", az_creds)

为各种前缀注册特定凭据。

xdlake.storage.register_filesystem("s3://bucket-doom/foo/bar", s3_creds)
xdlake.storage.register_filesystem("s3://bucket-zoom/biz/baz", other_s3_creds)
xdlake.storage.register_filesystem("az://container-blah/whiz/whaz", az_creds)

链接

项目主页 GitHub
DeltaLake 事务日志 协议

错误

请在 GitHub 上报告错误、问题、功能请求等。

Gitpod 工作区

启动 gitpod 工作区

构建状态