数据工程大数据批处理【免费下载链接】smallpondA lightweight data processing framework built on DuckDB and 3FS.项目地址https://gitcode.com/gh_mirrors/sma/smallpond点击查看免费下载导读本文是 smallpond 官方快速入门文档docs/source/getstarted.rst的完整实战指南覆盖从安装、会话初始化、数据加载、分区、变换、保存到任务监控的完整工作流。smallpond 是一个构建在 DuckDB 计算引擎与 3FS 等共享文件系统之上的轻量级分布式数据处理框架通过本文你将掌握用read_parquet加载数据集、用repartition三种模式手动分区、用 SQL 表达式或 Python 函数做数据变换以及利用 Ray Dashboard 监控分布式任务执行的完整技能。环境要求与安装smallpond 要求Python 3.8README.md 中进一步说明支持 3.8 至 3.12安装只需一条命令pip install smallpond核心运行时依赖见 pyproject.tomlduckdb 1.2.0计算引擎、pyarrow ~ 16.1.0列式数据交换与 Parquet 读写、ray[default] 2.10.0分布式任务调度、polars、pandas、fsspec等。此外还提供了可选扩展组pip install .[dev] # 开发与单元测试pytest 等 pip install .[docs] # 构建 Sphinx 文档 pip install .[warc] # WARC 网页数据解析支持初始化 smallpond 会话初始化是整个框架的入口代码极为简洁import smallpond sp smallpond.init()默认情况下init()会在当前节点启动一个本地 Ray 集群作为任务调度与执行环境。如果需要更多工作节点通过num_executors参数指定sp smallpond.init(num_executors10)从 smallpond/init.py 的init()源码可见所有参数均可选显式传入 环境变量SP_*前缀 默认值。主要参数如下参数环境变量默认值说明job_idSP_JOBID随机 UUID唯一任务标识job_timeSP_JOB_TIME当前时间任务创建时间epoch 秒job_nameSP_JOB_NAME当前脚本文件名任务显示名data_rootSP_DATA_ROOT~/.smallpond/data运行时生成文件的根目录num_executorsSP_NUM_EXECUTORS0执行器数量0表示所有任务在当前节点运行ray_addressSP_RAY_ADDRESS无指定时连接已有 Ray 集群否则新建bind_numa_nodeSP_BIND_NUMA_NODEFalse是否将执行进程绑定到 NUMA 节点memory_allocatorSP_MEMORY_ALLOCATORmimallocworker 进程内存分配器system/jemalloc/mimallocplatformSP_PLATFORM自动检测平台类型目前可选mpi_remove_output_root—True任务结束后是否清理{data_root}/output配置的解析逻辑位于 smallpond/session.py 的Config.from_args_and_env()其优先级为「函数参数 SP_*环境变量 平台默认值」。会话退出时通过atexit自动调用session.shutdown()清理资源见 smallpond/init.py。关于data_root所有运行时数据都集中存放在该目录下结构为data_root/job_time.job_id/{config, log, queue, output, staging, temp}详见 docs/source/internals.rst。加载数据读取 Parquet 文件smallpond 的核心数据集格式是 Parquet读取方式smallpond/dataframe.pydf sp.read_parquet(path/to/dataset/*.parquet)read_parquet支持以下参数参数默认值说明paths必填文件路径、路径列表或 glob 通配模式如[data/100.parquet, /datasetA/*.parquet]recursiveFalse是否递归解析路径模式columnsNone非空时只加载指定列列裁剪union_by_nameFalse按列名合并不同 schema 的文件对应 DuckDB 的 union-by-name 语义其他数据源除 Parquet 外smallpond/dataframe.py 还提供了一组等价的数据加载方法# CSV必须显式给出 schema可自定义分隔符 df sp.read_csv(path/to/*.csv, schema{urlstr: varchar, valstr: varchar}, delim,) # JSON df sp.read_json(path/to/*.json, schema{col: varchar}) # 本地 Python 对象 / pandas / pyarrow df sp.from_items([1, 2, 3]) # 非 dict 元素包装为 {item: x} df sp.from_items([{a: 1}, {a: 2}]) df sp.from_pandas(pandas_df) df sp.from_arrow(arrow_table)这些方法在 tests/test_dataframe.py 中均有对应断言例如test_csv验证读取 1000 行 TSV、test_items验证列表与 dict 列表的转换行为。从源码结构看加载数据时read_*方法会构建一个ParquetDataSet/CsvDataSet等数据集对象再由DataSourceNode包装成逻辑计划节点见 smallpond/dataframe.pyDataSet 负责解析 glob 通配、解析绝对路径列表见 smallpond/logical/dataset.py。数据分区repartition 的三种模式当前版本要求用户手动指定数据分区。分区是分布式计算的关键它决定数据如何拆分给并行任务直接决定partial_sql中 join/group by 能否在分区内本地完成。df df.repartition(3) # 按文件默认 df df.repartition(3, by_rowsTrue) # 按行数均匀切分 df df.repartition(3, hash_byhost) # 按列的哈希值分区注意一个细节原快速入门文档写的是by_rowTrue但当前源码 smallpond/dataframe.py 中repartition的签名是by_rows: bool False以源码为准。从 smallpond/dataframe.py 的实现可以看到三种模式分别映射到不同的逻辑节点repartition(npartitions)生成EvenlyDistributedPartitionNode按文件粒度均分partition_by_rowsFalserepartition(npartitions, by_rowsTrue)仍为EvenlyDistributedPartitionNode但partition_by_rowsTrue按行数均分可避免文件大小悬殊导致的数据倾斜repartition(npartitions, hash_byhost)生成HashPartitionNode将指定列支持字符串或列名列表的哈希值取模分配到各分区相同键的数据必然落在同一分区repartition(npartitions, bybucket)生成ShuffleNode按列值本身分区data_partition_column与hash_by互斥源码中会断言二者不能同时指定。分区是后续 join/聚合能够高效执行的前提。例如 smallpond/dataframe.py 中partial_sql的 join 示例就要求两个数据集按同一列id哈希分区到相同分区数才能实现分区内本地 joina sp.read_parquet(a/*.parquet).repartition(10, hash_byid) b sp.read_parquet(b/*.parquet).repartition(10, hash_byid) c sp.partial_sql(select * from {0} join {1} on a.id b.id, a, b)数据变换SQL 表达式与 Python 函数map逐行变换map是核心变换方法同时接受 SQL 表达式或 Python 函数smallpond/dataframe.py# SQL 表达式推荐更高效走 DuckDB 引擎 df df.map(a b as c) # Python 函数输入/输出均为 dict df df.map(lambda row: {c: row[a] row[b]})使用 Python 函数时可通过schema参数显式指定输出 schema当映射结果无法从首行推断类型时例如首行为None必须提供该参数tests/test_dataframe.py 的test_map对此有完整验证df3 df.map( lambda r: {c: None if r[a] 1 else r[a] r[b]}, schemapa.schema([(c, pa.int64())]), )更多变换方法方法作用示例filter按谓词过滤df.filter(a 1)或df.filter(lambda r: r[a] 1)flat_map一行展开为多行df.flat_map(unnest(array[a, b]) as c)map_batches按批处理 pyarrow Tabledf.map_batches(fn, batch_size122880)partial_sort分区内排序df.partial_sort(by[a, b desc])random_shuffle全局随机打乱df.random_shuffle()limit截断行数不触发执行df.limit(10)partial_sql分区内执行 SQL{0}占位输入 DataFramesp.partial_sql(SELECT ... FROM {0}, df)其中partial_sql支持用{0}、{1}引用多个输入 DataFrame多个输入必须具有相同的分区数见 smallpond/dataframe.py。在 SQL 表达式中使用 UDFmap/filter的 SQL 表达式可通过udfs参数加载用户自定义函数支持 Python 函数与 DuckDB 扩展两种形式smallpond/dataframe.pyfrom smallpond.logical.udf import UDFType, udf udf(params[UDFType.INT, UDFType.INT], return_typeUDFType.INT) def gcd(a: int, b: int) - int: while b: a, b b, a % b return a df df.map(gcd(a, b), udfs[gcd]) # 或从 duckdb extension 文件加载 df df.map(gcd(a, b), udfs[path/to/udf.duckdb_extension])保存与消费数据写出 Parquetdf.write_parquet(path/to/output)write_parquet是阻塞操作内部等价于write_parquet_lazy(path).compute()见 smallpond/dataframe.py。需要并行写出多个结果时用非阻塞版本配合sp.wait一次性触发执行o1 df.write_parquet_lazy(output1) o2 df.map(col1, col2).write_parquet_lazy(output2) sp.wait(o1, o2)结果回读以下操作会触发惰性变换的执行并返回结果smallpond/dataframe.pydf.count() # 行数仅当无法从元数据确定时触发执行 rows df.take(10) # 最多返回 10 行List[Dict] rows df.take_all() # 返回全部行 pandas_df df.to_pandas() # 转为 pandas DataFrame arrow_table df.to_arrow() # 转为 pyarrow Table惰性执行与手动控制smallpond 的 DataFrame 是惰性求值的read_parquet、map、repartition等操作只构建逻辑计划真正计算发生在触发性操作write_parquet、count、to_pandas等调用时。框架会先经Optimizer优化逻辑计划再由Planner生成并调度执行任务见 smallpond/dataframe.py。如需手动控制执行时机可使用以下方法df.compute() # 触发执行 df.is_computed() # 数据是否已落盘 df.recompute() # 强制重算忽略已有结果 sp.wait(df1, df2) # 等待多个 DataFrame 完成recompute()的典型场景是修改变换后复用中间结果input.select(b).recompute()会重算select结果而复用inputsmallpond/dataframe.py。已计算的任务结果会被记录在 session 中后续 DataFrame 可复用任务避免重复计算smallpond/dataframe.py。任务监控与 Ray Dashboardsmallpond 使用Ray Core 作为任务调度器因此可以借助Ray Dashboard实时查看任务执行状态、资源使用与任务时间线。当 smallpond 在本地启动 Ray 集群时日志会打印 Dashboard 地址... Started a local Ray instance. View the dashboard at http://127.0.0.1:8008从 smallpond/session.py 的源码可以看到本地集群固定使用dashboard_host0.0.0.0、dashboard_port8008并通过_metrics_export_port8080暴露 Prometheus 指标若本机安装了 prometheus 与 grafana还会自动启动Grafana 运行于http://localhost:8122。此外session 会启动一个后台线程每分钟导出逻辑计划图graph.png与任务时间线并输出进度日志progress: 45/100 tasks (45.0%)具体实现在 smallpond/session.py可用于长任务的分阶段观察。完整实战示例综合以上步骤一个典型的处理流程如下改编自 README.md 的 Quick Start# 下载示例数据 wget https://duckdb.org/data/prices.parquetimport smallpond # 1. 初始化会话 sp smallpond.init() # 2. 加载数据 df sp.read_parquet(prices.parquet) # 3. 按 ticker 哈希分区 df df.repartition(3, hash_byticker) # 4. 分区内 SQL 聚合 df sp.partial_sql(SELECT ticker, min(price), max(price) FROM {0} GROUP BY ticker, df) # 5. 保存结果 df.write_parquet(output/) # 6. 查看结果 print(df.to_pandas())仓库 examples/sort_mock_urls_v2.py 提供了更完整的端到端示例——对 URL 数据按 host 分区并排序def sort_mock_urls_v2(sp: Session, input_paths, output_path, npartitions): dataset sp.read_csv(input_paths, schema{urlstr: varchar, valstr: varchar}, delimr\t).repartition(npartitions) urls dataset.map( split_part(urlstr, /, 1) as host, split_part(urlstr, , 1) as url, from_base64(valstr) AS payload ) urls urls.repartition(npartitions, hash_byhost) sorted_urls urls.partial_sort(by[host]) sorted_urls.write_parquet(output_path) sp smallpond.init() sort_mock_urls_v2(sp, input_paths[tests/data/mock_urls/*.tsv], output_pathsort_mock_urls, npartitions10)进一步阅读完整 API 参考docs/source/api/dataframe.rst加载/分区/变换/消费/执行方法全集运行时目录结构与故障恢复机制docs/source/internals.rst数据集类型ParquetDataSet/CsvDataSet 等docs/source/api/dataset.rst单元测试示例tests/test_dataframe.py、tests/test_partition.py更多示例脚本examples/shuffle_data.py、shuffle_mock_urls.py、sort_mock_urls.py 等赞分享数据工程大数据批处理【免费下载链接】smallpondA lightweight data processing framework built on DuckDB and 3FS.项目地址https://gitcode.com/gh_mirrors/sma/smallpond点击查看免费下载相关推荐smallpond 快速上手指南基于 DuckDB 与 3FS 的轻量级分布式数据处理框架smallpond 快速上手指南基于 DuckDB 与 3FS 的轻量级分布式数据处理框架 本文以仓库根目录 README.md https://link.g数据工程大数据批处理smallpond 逻辑计划节点Nodes详解构建基于 DuckDB 与 3FS 的数据处理流水线smallpond 逻辑计划节点Nodes详解构建基于 DuckDB 与 3FS 的数据处理流水线 导读 Nodes 是 smallpond 数据处理流数据工程大数据批处理smallpond入门指南5分钟快速上手轻量级数据处理框架smallpond入门指南5分钟快速上手轻量级数据处理框架 smallpond 是一个基于 DuckDB 和 3FS 构建的轻量级数据处理框架专为高性数据工程大数据批处理上一篇OpenUPM架构深度解析从Git标签到UPM包的完整流程下一篇如何用OpCore-Simplify一键搞定黑苹果OpenCore配置完整高效指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/10/13 6:02:16 · 阅读 1.2k smallpond 快速入门：基于 DuckDB 与 3FS 的分布式数据处理实战指南
数据工程大数据批处理【免费下载链接】smallpondA lightweight data processing framework built on DuckDB and 3FS.项目地址https://gitcode.com/gh_mirrors/sma/smallpond点击查看免费下载导读本文是 smallpond 官方快速入门文档docs/source/getstarted.rst的完整实战指南覆盖从安装、会话初始化、数据加载、分区、变换、保存到任务监控的完整工作流。smallpond 是一个构建在 DuckDB 计算引擎与 3FS 等共享文件系统之上的轻量级分布式数据处理框架通过本文你将掌握用read_parquet加载数据集、用repartition三种模式手动分区、用 SQL 表达式或 Python 函数做数据变换以及利用 Ray Dashboard 监控分布式任务执行的完整技能。环境要求与安装smallpond 要求Python 3.8README.md 中进一步说明支持 3.8 至 3.12安装只需一条命令pip install smallpond核心运行时依赖见 pyproject.tomlduckdb 1.2.0计算引擎、pyarrow ~ 16.1.0列式数据交换与 Parquet 读写、ray[default] 2.10.0分布式任务调度、polars、pandas、fsspec等。此外还提供了可选扩展组pip install .[dev] # 开发与单元测试pytest 等 pip install .[docs] # 构建 Sphinx 文档 pip install .[warc] # WARC 网页数据解析支持初始化 smallpond 会话初始化是整个框架的入口代码极为简洁import smallpond sp smallpond.init()默认情况下init()会在当前节点启动一个本地 Ray 集群作为任务调度与执行环境。如果需要更多工作节点通过num_executors参数指定sp smallpond.init(num_executors10)从 smallpond/init.py 的init()源码可见所有参数均可选显式传入 环境变量SP_*前缀 默认值。主要参数如下参数环境变量默认值说明job_idSP_JOBID随机 UUID唯一任务标识job_timeSP_JOB_TIME当前时间任务创建时间epoch 秒job_nameSP_JOB_NAME当前脚本文件名任务显示名data_rootSP_DATA_ROOT~/.smallpond/data运行时生成文件的根目录num_executorsSP_NUM_EXECUTORS0执行器数量0表示所有任务在当前节点运行ray_addressSP_RAY_ADDRESS无指定时连接已有 Ray 集群否则新建bind_numa_nodeSP_BIND_NUMA_NODEFalse是否将执行进程绑定到 NUMA 节点memory_allocatorSP_MEMORY_ALLOCATORmimallocworker 进程内存分配器system/jemalloc/mimallocplatformSP_PLATFORM自动检测平台类型目前可选mpi_remove_output_root—True任务结束后是否清理{data_root}/output配置的解析逻辑位于 smallpond/session.py 的Config.from_args_and_env()其优先级为「函数参数 SP_*环境变量 平台默认值」。会话退出时通过atexit自动调用session.shutdown()清理资源见 smallpond/init.py。关于data_root所有运行时数据都集中存放在该目录下结构为data_root/job_time.job_id/{config, log, queue, output, staging, temp}详见 docs/source/internals.rst。加载数据读取 Parquet 文件smallpond 的核心数据集格式是 Parquet读取方式smallpond/dataframe.pydf sp.read_parquet(path/to/dataset/*.parquet)read_parquet支持以下参数参数默认值说明paths必填文件路径、路径列表或 glob 通配模式如[data/100.parquet, /datasetA/*.parquet]recursiveFalse是否递归解析路径模式columnsNone非空时只加载指定列列裁剪union_by_nameFalse按列名合并不同 schema 的文件对应 DuckDB 的 union-by-name 语义其他数据源除 Parquet 外smallpond/dataframe.py 还提供了一组等价的数据加载方法# CSV必须显式给出 schema可自定义分隔符 df sp.read_csv(path/to/*.csv, schema{urlstr: varchar, valstr: varchar}, delim,) # JSON df sp.read_json(path/to/*.json, schema{col: varchar}) # 本地 Python 对象 / pandas / pyarrow df sp.from_items([1, 2, 3]) # 非 dict 元素包装为 {item: x} df sp.from_items([{a: 1}, {a: 2}]) df sp.from_pandas(pandas_df) df sp.from_arrow(arrow_table)这些方法在 tests/test_dataframe.py 中均有对应断言例如test_csv验证读取 1000 行 TSV、test_items验证列表与 dict 列表的转换行为。从源码结构看加载数据时read_*方法会构建一个ParquetDataSet/CsvDataSet等数据集对象再由DataSourceNode包装成逻辑计划节点见 smallpond/dataframe.pyDataSet 负责解析 glob 通配、解析绝对路径列表见 smallpond/logical/dataset.py。数据分区repartition 的三种模式当前版本要求用户手动指定数据分区。分区是分布式计算的关键它决定数据如何拆分给并行任务直接决定partial_sql中 join/group by 能否在分区内本地完成。df df.repartition(3) # 按文件默认 df df.repartition(3, by_rowsTrue) # 按行数均匀切分 df df.repartition(3, hash_byhost) # 按列的哈希值分区注意一个细节原快速入门文档写的是by_rowTrue但当前源码 smallpond/dataframe.py 中repartition的签名是by_rows: bool False以源码为准。从 smallpond/dataframe.py 的实现可以看到三种模式分别映射到不同的逻辑节点repartition(npartitions)生成EvenlyDistributedPartitionNode按文件粒度均分partition_by_rowsFalserepartition(npartitions, by_rowsTrue)仍为EvenlyDistributedPartitionNode但partition_by_rowsTrue按行数均分可避免文件大小悬殊导致的数据倾斜repartition(npartitions, hash_byhost)生成HashPartitionNode将指定列支持字符串或列名列表的哈希值取模分配到各分区相同键的数据必然落在同一分区repartition(npartitions, bybucket)生成ShuffleNode按列值本身分区data_partition_column与hash_by互斥源码中会断言二者不能同时指定。分区是后续 join/聚合能够高效执行的前提。例如 smallpond/dataframe.py 中partial_sql的 join 示例就要求两个数据集按同一列id哈希分区到相同分区数才能实现分区内本地 joina sp.read_parquet(a/*.parquet).repartition(10, hash_byid) b sp.read_parquet(b/*.parquet).repartition(10, hash_byid) c sp.partial_sql(select * from {0} join {1} on a.id b.id, a, b)数据变换SQL 表达式与 Python 函数map逐行变换map是核心变换方法同时接受 SQL 表达式或 Python 函数smallpond/dataframe.py# SQL 表达式推荐更高效走 DuckDB 引擎 df df.map(a b as c) # Python 函数输入/输出均为 dict df df.map(lambda row: {c: row[a] row[b]})使用 Python 函数时可通过schema参数显式指定输出 schema当映射结果无法从首行推断类型时例如首行为None必须提供该参数tests/test_dataframe.py 的test_map对此有完整验证df3 df.map( lambda r: {c: None if r[a] 1 else r[a] r[b]}, schemapa.schema([(c, pa.int64())]), )更多变换方法方法作用示例filter按谓词过滤df.filter(a 1)或df.filter(lambda r: r[a] 1)flat_map一行展开为多行df.flat_map(unnest(array[a, b]) as c)map_batches按批处理 pyarrow Tabledf.map_batches(fn, batch_size122880)partial_sort分区内排序df.partial_sort(by[a, b desc])random_shuffle全局随机打乱df.random_shuffle()limit截断行数不触发执行df.limit(10)partial_sql分区内执行 SQL{0}占位输入 DataFramesp.partial_sql(SELECT ... FROM {0}, df)其中partial_sql支持用{0}、{1}引用多个输入 DataFrame多个输入必须具有相同的分区数见 smallpond/dataframe.py。在 SQL 表达式中使用 UDFmap/filter的 SQL 表达式可通过udfs参数加载用户自定义函数支持 Python 函数与 DuckDB 扩展两种形式smallpond/dataframe.pyfrom smallpond.logical.udf import UDFType, udf udf(params[UDFType.INT, UDFType.INT], return_typeUDFType.INT) def gcd(a: int, b: int) - int: while b: a, b b, a % b return a df df.map(gcd(a, b), udfs[gcd]) # 或从 duckdb extension 文件加载 df df.map(gcd(a, b), udfs[path/to/udf.duckdb_extension])保存与消费数据写出 Parquetdf.write_parquet(path/to/output)write_parquet是阻塞操作内部等价于write_parquet_lazy(path).compute()见 smallpond/dataframe.py。需要并行写出多个结果时用非阻塞版本配合sp.wait一次性触发执行o1 df.write_parquet_lazy(output1) o2 df.map(col1, col2).write_parquet_lazy(output2) sp.wait(o1, o2)结果回读以下操作会触发惰性变换的执行并返回结果smallpond/dataframe.pydf.count() # 行数仅当无法从元数据确定时触发执行 rows df.take(10) # 最多返回 10 行List[Dict] rows df.take_all() # 返回全部行 pandas_df df.to_pandas() # 转为 pandas DataFrame arrow_table df.to_arrow() # 转为 pyarrow Table惰性执行与手动控制smallpond 的 DataFrame 是惰性求值的read_parquet、map、repartition等操作只构建逻辑计划真正计算发生在触发性操作write_parquet、count、to_pandas等调用时。框架会先经Optimizer优化逻辑计划再由Planner生成并调度执行任务见 smallpond/dataframe.py。如需手动控制执行时机可使用以下方法df.compute() # 触发执行 df.is_computed() # 数据是否已落盘 df.recompute() # 强制重算忽略已有结果 sp.wait(df1, df2) # 等待多个 DataFrame 完成recompute()的典型场景是修改变换后复用中间结果input.select(b).recompute()会重算select结果而复用inputsmallpond/dataframe.py。已计算的任务结果会被记录在 session 中后续 DataFrame 可复用任务避免重复计算smallpond/dataframe.py。任务监控与 Ray Dashboardsmallpond 使用Ray Core 作为任务调度器因此可以借助Ray Dashboard实时查看任务执行状态、资源使用与任务时间线。当 smallpond 在本地启动 Ray 集群时日志会打印 Dashboard 地址... Started a local Ray instance. View the dashboard at http://127.0.0.1:8008从 smallpond/session.py 的源码可以看到本地集群固定使用dashboard_host0.0.0.0、dashboard_port8008并通过_metrics_export_port8080暴露 Prometheus 指标若本机安装了 prometheus 与 grafana还会自动启动Grafana 运行于http://localhost:8122。此外session 会启动一个后台线程每分钟导出逻辑计划图graph.png与任务时间线并输出进度日志progress: 45/100 tasks (45.0%)具体实现在 smallpond/session.py可用于长任务的分阶段观察。完整实战示例综合以上步骤一个典型的处理流程如下改编自 README.md 的 Quick Start# 下载示例数据 wget https://duckdb.org/data/prices.parquetimport smallpond # 1. 初始化会话 sp smallpond.init() # 2. 加载数据 df sp.read_parquet(prices.parquet) # 3. 按 ticker 哈希分区 df df.repartition(3, hash_byticker) # 4. 分区内 SQL 聚合 df sp.partial_sql(SELECT ticker, min(price), max(price) FROM {0} GROUP BY ticker, df) # 5. 保存结果 df.write_parquet(output/) # 6. 查看结果 print(df.to_pandas())仓库 examples/sort_mock_urls_v2.py 提供了更完整的端到端示例——对 URL 数据按 host 分区并排序def sort_mock_urls_v2(sp: Session, input_paths, output_path, npartitions): dataset sp.read_csv(input_paths, schema{urlstr: varchar, valstr: varchar}, delimr\t).repartition(npartitions) urls dataset.map( split_part(urlstr, /, 1) as host, split_part(urlstr, , 1) as url, from_base64(valstr) AS payload ) urls urls.repartition(npartitions, hash_byhost) sorted_urls urls.partial_sort(by[host]) sorted_urls.write_parquet(output_path) sp smallpond.init() sort_mock_urls_v2(sp, input_paths[tests/data/mock_urls/*.tsv], output_pathsort_mock_urls, npartitions10)进一步阅读完整 API 参考docs/source/api/dataframe.rst加载/分区/变换/消费/执行方法全集运行时目录结构与故障恢复机制docs/source/internals.rst数据集类型ParquetDataSet/CsvDataSet 等docs/source/api/dataset.rst单元测试示例tests/test_dataframe.py、tests/test_partition.py更多示例脚本examples/shuffle_data.py、shuffle_mock_urls.py、sort_mock_urls.py 等赞分享数据工程大数据批处理【免费下载链接】smallpondA lightweight data processing framework built on DuckDB and 3FS.项目地址https://gitcode.com/gh_mirrors/sma/smallpond点击查看免费下载相关推荐smallpond 快速上手指南基于 DuckDB 与 3FS 的轻量级分布式数据处理框架smallpond 快速上手指南基于 DuckDB 与 3FS 的轻量级分布式数据处理框架 本文以仓库根目录 README.md https://link.g数据工程大数据批处理smallpond 逻辑计划节点Nodes详解构建基于 DuckDB 与 3FS 的数据处理流水线smallpond 逻辑计划节点Nodes详解构建基于 DuckDB 与 3FS 的数据处理流水线 导读 Nodes 是 smallpond 数据处理流数据工程大数据批处理smallpond入门指南5分钟快速上手轻量级数据处理框架smallpond入门指南5分钟快速上手轻量级数据处理框架 smallpond 是一个基于 DuckDB 和 3FS 构建的轻量级数据处理框架专为高性数据工程大数据批处理上一篇OpenUPM架构深度解析从Git标签到UPM包的完整流程下一篇如何用OpCore-Simplify一键搞定黑苹果OpenCore配置完整高效指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考