DuckDB访问数据湖

DuckDB 不仅可以方便访问 Pandas DataFrame， CSV、Parquet、PyArrow，甚至可以方便地访问数据湖。

安装需要的包

pip install duckdb deltalake

DeltaTable 表示特定版本的增量表的状态。这包括哪些文件是当前表的一部分、表的结构以及其他元数据，例如创建时间。

加载本地文件系统的数据湖（当前版本）

path = "/tmp/iris_delta"
from deltalake import DeltaTable
dt = DeltaTable(path)

本次使用的测试数据由 PySpark +Delta Lake 生成，更多相关信息可以访问 Delta Lake 快速入门-PySpark 版

使用 DuckDB 访问

import duckdb
con=duckdb.connect()
def dsql(sql):
    return con.execute(sql).df()

ds = dt.to_pyarrow_dataset()
sql = "select * from ds limit 3"
dsql(sql)

当前版本只有 50 条信息

sql = "select count(*) from ds"
dsql(sql)

时间旅行(Time Travel)

除了可以访问最新版本的数据，还可以通过提供要加载的版本号来加载相应的版本：

dt = DeltaTable(path, version=1)
ds = dt.to_pyarrow_dataset()
sql = "select count(*) from ds"
dsql(sql)

也可以在加载表格后，通过使用版本号或日期时间字符串更改版本：

dt.load_version(1)
#等价于👇的
from datetime import datetime
timestamp =(datetime
            .fromtimestamp(1650358332868 / 1e3)
            .astimezone()
            .isoformat())
dt.load_with_datetime(timestamp)
dt.files()

这里的时间戳可以通过下面的命令获得，

#版本历史信息
dt.history()

[{'timestamp': 1650358316226,
  'operation': 'CREATE TABLE',
  'operationParameters': {'isManaged': 'false',
   'description': None,
   'partitionBy': '[]',
   'properties': '{}'},
  'isolationLevel': 'Serializable',
  'isBlindAppend': True,
  'operationMetrics': {},
  'engineInfo': 'Apache-Spark/3.2.1 Delta-Lake/1.2.0',
  'txnId': '0b21cbea-5672-4367-84be-5d6c2fd07aef'},
 {'timestamp': 1650358332868,
  'operation': 'WRITE',
  'operationParameters': {'mode': 'Append', 'partitionBy': '[]'},
  'readVersion': 0,
  'isolationLevel': 'Serializable',
  'isBlindAppend': True,
  'operationMetrics': {'numFiles': '1',
   'numOutputRows': '150',
   'numOutputBytes': '2792'},
  'engineInfo': 'Apache-Spark/3.2.1 Delta-Lake/1.2.0',
  'txnId': 'aea8a270-4f3d-4ad3-8a63-2d9aebedefa7'},
 {'timestamp': 1650358460951,
  'operation': 'CREATE OR REPLACE TABLE',
  'operationParameters': {'isManaged': 'false',
   'description': None,
   'partitionBy': '["Species"]',
   'properties': '{}'},
  'readVersion': 1,
  'isolationLevel': 'Serializable',
  'isBlindAppend': False,
  'operationMetrics': {},
  'engineInfo': 'Apache-Spark/3.2.1 Delta-Lake/1.2.0',
  'txnId': '679a731d-9456-4ed0-aa63-3ce8a4ece8bc'},
 {'timestamp': 1650358516037,
  'operation': 'WRITE',
  'operationParameters': {'mode': 'Append', 'partitionBy': '[]'},
  'readVersion': 2,
  'isolationLevel': 'Serializable',
  'isBlindAppend': True,
  'operationMetrics': {'numFiles': '3',
   'numOutputRows': '150',
   'numOutputBytes': '5612'},
  'engineInfo': 'Apache-Spark/3.2.1 Delta-Lake/1.2.0',
  'txnId': '443c8613-6dd7-4442-9111-adfd8dca7a67'},
 {'timestamp': 1650358641212,
  'operation': 'DELETE',
  'operationParameters': {'predicate': '[]'},
  'readVersion': 3,
  'isolationLevel': 'Serializable',
  'isBlindAppend': False,
  'operationMetrics': {'numRemovedFiles': '3',
   'executionTimeMs': '194',
   'scanTimeMs': '192',
   'rewriteTimeMs': '0'},
  'engineInfo': 'Apache-Spark/3.2.1 Delta-Lake/1.2.0',
  'txnId': 'c6f4373f-3f7b-41ca-8fa1-11501dbd2715'},
 {'timestamp': 1650358662597,
  'operation': 'WRITE',
  'operationParameters': {'mode': 'Append', 'partitionBy': '[]'},
  'readVersion': 4,
  'isolationLevel': 'Serializable',
  'isBlindAppend': True,
  'operationMetrics': {'numFiles': '3',
   'numOutputRows': '50',
   'numOutputBytes': '5031'},
  'engineInfo': 'Apache-Spark/3.2.1 Delta-Lake/1.2.0',
  'txnId': 'cee79e6b-63bc-48a1-9818-ccb37e17d811'}]

多种文件系统支持

除了本地文件系统，可以通过 storage_options 来配置存储后端，如 AWS S3，

storage_options = {"AWS_ACCESS_KEY_ID": "AWS_ACCESS_KEY_ID",
     "AWS_SECRET_ACCESS_KEY":"AWS_SECRET_ACCESS_KEY"}
dt = DeltaTable("../rust/tests/data/delta-0.2.0",
  storage_options=storage_options)

或者，如果您有一个数据目录，您可以通过引用数据库和表名来加载它。目前仅支持 AWS Glue。

对于 AWS Glue 目录，使用 AWS 环境变量进行身份验证。

from deltalake import DeltaTable
from deltalake import DataCatalog
database_name = "simple_database"
table_name = "simple_table"
data_catalog = DataCatalog.AWS
dt = DeltaTable.from_data_catalog(data_catalog=data_catalog,
    database_name=database_name,
    table_name=table_name)
dt.to_pyarrow_table().to_pydict()
{'id': [5, 7, 9, 5, 6, 7, 8, 9]}

除了本地文件系统，还支持以下后端：

AWS S3，由前缀检测 s3://。可以使用与 CLI 相同的方式使用环境变量指定 AWS 凭证。
Azure Data Lake Storage Gen 2，由前缀检测 adls2://。请注意，必须按照说明设置 Azure 存储帐户^[1]。
Google Cloud Storage，由前缀检测 gs://。

更多的访问方式

除了使用 SQL 的方式访问，可以可以这样的方式

import duckdb
from deltalake import DeltaTable
dt = DeltaTable(path)
ds = dt.to_pyarrow_dataset()
ex_data = duckdb.arrow(ds)
(ex_data
 .filter("Species = 'virginica' and Sepal_Length > 7")
 .project("Sepal_Length")
 .to_df())
#返回
 Sepal_Length
0 7.2
1 7.1
2 7.2