dlt项目中使用DuckDB/MotherDuck作为数据源的解决方案

问题背景

在使用dlt(Data Load Tool)项目时，许多开发者尝试将DuckDB或MotherDuck作为数据源，通过sql_database()和sql_table()函数加载数据时遇到了问题。这些问题主要表现为dlt无法正确解析DuckDB的目录结构，导致表名和列名限定符无法正确识别。

问题表现

当开发者尝试从DuckDB/MotherDuck加载数据时，会遇到两种典型的错误：

1. 表不存在错误：dlt生成的SQL语句中表名没有正确限定，导致DuckDB无法识别表的位置。错误信息会提示类似"Table with name foo does not exist! Did you mean 'db_name.schema_name.foo'?"。

2. 反射失败错误：当尝试使用完全限定的表名(db_name.schema_name.table_name)时，SQLAlchemy会抛出反射错误，提示无法在引擎中找到请求的表。

解决方案

临时解决方案

开发者可以提供一个查询适配器回调函数来手动指定完全限定的表名：

def query_adapter_callback(query, table) -> TextClause:
    t_query = sa.text(f"SELECT * FROM {table.fullname}")
    return t_query

这个回调函数会覆盖默认的查询生成逻辑，确保使用完全限定的表名。

推荐解决方案

实际上，dlt项目已经支持直接从DuckDB读取数据。以下是一个完整的工作示例：

import dlt
from dlt.sources.sql_database import sql_database

# 配置源和目标数据库路径
SOURCE_DB_FILE_NAME = "./db.duckdb"
DESTINATION_DB_FILE_NAME = "./db_out.duckdb"

# 创建示例数据表
pipeline = dlt.pipeline(
    pipeline_name="source_pipe",
    dataset_name="source_dataset",
    destination=dlt.destinations.duckdb(SOURCE_DB_FILE_NAME)
pipeline.run([1,2,3], table_name="source_table")

# 从源数据库读取数据
table_source = sql_database(
    "duckdb:///" + SOURCE_DB_FILE_NAME,
    schema="source_dataset",
    reflection_level="full",
).with_resources('source_table')

# 将数据加载到目标数据库
dest_pipeline = dlt.pipeline(
    pipeline_name="dest_pipe",
    dataset_name="dest_dataset",
    destination=dlt.destinations.duckdb(DESTINATION_DB_FILE_NAME))
dest_pipeline.run(table_source)

# 验证结果
print(dest_pipeline.dataset().source_table.df())

关键点说明

1. 连接字符串格式：对于本地DuckDB数据库，使用"duckdb:///"前缀加上文件路径；对于MotherDuck，使用"duckdb:///md:"前缀。

2. schema参数：在DuckDB中，schema参数对应于数据集名称(dataset_name)，这是DuckDB组织表的方式。

3. 反射级别：reflection_level="full"确保获取完整的表结构信息。

4. 资源选择：.with_resources()方法可以明确指定要加载的表名。

最佳实践建议

1. 对于简单的数据迁移场景，可以考虑直接使用dlt的DuckDB目的地功能，而不是通过SQL数据库源。

2. 当确实需要使用SQL数据库源时，确保：

   • 连接字符串格式正确
   • schema参数与DuckDB中的数据集名称匹配
   • 表名大小写一致

3. 对于生产环境，建议添加错误处理和日志记录，特别是在处理大型数据库时。

通过以上方法，开发者可以顺利地在dlt项目中使用DuckDB和MotherDuck作为数据源，实现数据的提取和加载操作。