# App-DuckETL **Repository Path**: skyter10086/App-DuckETL ## Basic Information - **Project Name**: App-DuckETL - **Description**: No description available - **Primary Language**: Unknown - **License**: MIT - **Default Branch**: master - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 0 - **Created**: 2026-10-30 - **Last Updated**: 2026-11-05 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README # App::DuckETL SQL-first ETL on DuckDB. Raku only plans SQL and renders results; rows stay inside DuckDB. 当前进度：**M0 引擎 + M1 只读 TUI + M2 编辑闭环 + M3 生产化**（`ducketl run/tui/worker`、增量水位、Parquet 分区写、`--isolated` worker、DuckDB 扩展装载、Pipeline 表单 / `$EDITOR` / 保存 JSON、history、失败回滚、原子落盘、运行取消）。设计依据见 [TUI-ETL-design-analysis.md](TUI-ETL-design-analysis.md)。 ## 快速开始 ```sh raku -Ilib bin/ducketl validate examples/pipelines/nightly_orders.json raku -Ilib bin/ducketl run examples/pipelines/nightly_orders.json # => examples/out/orders.parquet raku -Ilib bin/ducketl run examples/pipelines/orders_table.json # => examples/out/ducketl.duckdb, mart.orders（upsert） raku -Ilib bin/ducketl tui examples/pipelines/nightly_orders.json # => 全屏 TUI：Dashboard / Sources / Pipeline / Run raku -Ilib bin/ducketl run examples/pipelines/orders_incremental.json raku -Ilib bin/ducketl run examples/pipelines/orders_incremental.json # => 第二次按 created_at 水位追加 0 行 raku -Ilib bin/ducketl run examples/pipelines/partitioned_orders.json # => examples/out/orders_by_country/country=*/data_0.parquet ducketl extensions # 列出 DuckDB 扩展 ducketl tui --isolated pipeline.json # worker 进程独占数据库 ``` 用 zef 安装： ```sh zef install . ducketl run pipeline.json ducketl tui pipeline.json ``` 依赖装在默认 `site` 仓库以外的仓库时（例如 `RAKULIB=inst#/path/to/repo`），`zef` 的安装预编译看不到那个仓库，会报 `Could not find Terminal::UI` / `Could not find Terminal::ANSI` （单独装 `zef install Terminal::UI` 也一样）。把发行版装进同一个仓库即可： ```sh zef install --install-to=inst#/path/to/repo . ``` Windows 上 `zef` 解压 `DB::DuckDB` 0.9.1 时报 `Malformed UTF-8 near bytes ...`：该归档带中文文件名，bsdtar 列目录时输出非法 UTF-8。手动解包后从目录安装该依赖，再装本发行版： ```sh zef install <解包后的 DB::DuckDB 目录> zef install . ``` 依赖：Rakudo 2026.09、`DB::DuckDB` 0.8.3、`Alien::Libduckdb` 1.1.0（DuckDB 2.x）、`JSON::Fast`、`Terminal::UI` 0.1.14（`--backend terminal` 默认 TUI）、Python Textual v8.2.8（`--backend textual`，用 `ducketl tui --install` 或 `pip install textual` 安装）。 ## 命令 ```text ducketl run [--db PATH] [--history PATH] [--threads N] [--memory-limit SIZE] [--keep-temp] [--json] [--quiet] ducketl tui [pipeline.json] [--db PATH] [--history PATH] [--threads N] [--memory-limit SIZE] [--isolated] [--backend terminal|textual] [--python-bin PATH] ducketl tui --install # 安装 Python Textual（backend=textual 需要） ducketl worker [--db PATH] [--history PATH] [--threads N] [--memory-limit SIZE] [--keep-temp] ducketl extensions [--json] ducketl validate ducketl list [--history PATH] [--limit N] [--json] ducketl show [--history PATH] [--json] ducketl query "" [--db PATH] [--limit N] [--json] ducketl version ``` `worker` 是 `tui --isolated` 使用的子进程入口：stdout 每行一个 JSON 事件，stdin 收到 `cancel` 时取消当前运行。 - `--db` 缺省时使用 pipeline 的 `database`，再缺省为 `:memory:`。 - history 默认 `~/.local/share/ducketl/history.duckdb`，与数据库分开文件。 - `run` 的进度写 stderr，结果写 stdout；`--json` 输出完整运行结果。 - `query` 对数据数据库执行一条 SQL：`select` / `with` / `values` 最多取 `--limit` 行（默认 1000），`show` / `describe` / `explain` 原样执行，其余语句执行并报告影响行数。结果表格写 stdout（数值右对齐、`NULL` 原样），`N rows in Xms` 写 stderr；`--json` 输出 `columns` / `rows` / `limited` / `affected` / `elapsed_ms`。 ## Pipeline 格式 canonical 格式是 JSON。路径相对 pipeline 文件所在目录解析。 ```json { "name": "nightly_orders", "database": "../out/ducketl.duckdb", "defaults": { "threads": 4, "memory_limit": "1GB" }, "steps": [ { "id": "extract", "type": "source", "format": "csv", "path": "../data/orders.csv", "options": { "header": true, "delim": ",", "sample_size": 100000 } }, { "id": "clean", "type": "sql", "sql": "select id, cast(amount as decimal(18,2)) as amount from extract where amount is not null" }, { "id": "check", "type": "assert", "sql": "select count(*) from clean where amount < 0", "expect": 0 }, { "id": "load", "type": "sink", "target": "mart.orders", "mode": "upsert", "key": ["id"] } ] } ``` ### 步骤类型 | 类型 | 作用 | 关键字段 | |---|---|---| | `source` | CSV / Parquet / JSON / 表 / SQL | `format`、`path`、`table`、`sql`、`options` | | `sql` | 任意 SELECT | `sql`、`materialize` | | `assert` | 数据质量断言 | `sql`、`expect`、`op`（`eq/ne/lt/lte/gt/gte`） | | `sink` | 表 / Parquet / CSV / 分区 Parquet 目录 | `target`、`mode`（`replace/append/upsert`）、`key`、`watermark`、`partition_by` | - `source` 支持 `csv`、`parquet`、`json`、`table`、`sql`。 - 文件路径作为绑定参数传给 `read_csv_auto(?)` / `read_parquet(?)` / `read_json_auto(?)`。 - `path` 也可以是路径列表。 - `sql` / `assert` 用步骤 id 引用前一步的数据：`from extract`、`join customers`。只有 `source` 与 `sql` 步骤产出数据；引用尚未产出或非数据步骤会校验失败。 - 中间结果默认物化为临时表。`"materialize": false` 改为临时视图（DuckDB 会把视图内联进下游查询），适合把整条链压成一个查询。 - `sink` 的 `target` 以 `.parquet` / `.csv` 结尾时写文件，否则写表（`table` 或 `schema.table`）。 - 文件先写同目录 staged 文件，成功提交后再 `rename`；失败不会碰目标文件。 - `upsert` 需要 `key`；首次运行会建表并创建唯一索引，后续按 `key` 更新。 - `append` 在目标不存在时自动建表。 - 表 sink 在事务内执行；任一步失败都会回滚，目标表回到运行前状态。 ### 增量水位、分区写与扩展 ```json { "name": "incremental", "extensions": ["parquet"], "steps": [ { "id": "extract", "type": "source", "format": "csv", "path": "raw/events.csv" }, { "id": "load", "type": "sink", "target": "mart.events", "mode": "append", "watermark": "created_at" } ] } ``` - `watermark` 只用于**表 sink** 的 `append` / `upsert`：首次运行全量，之后自动加 `where > `；值存在 history 的 `_etl.watermarks` （`pipeline + step_id + column_name`）。水位只在数据 commit（以及文件安装）成功后更新，失败重跑不会跳过数据。重复运行同一份输入插入 **0 行**。 - `partition_by` 用于 Parquet 分区目录：`target` 是目录路径（含 `/`、以 `~` 或 `.` 开头），`mode` 只能是 `replace`。支持 `options.compression`（默认 `zstd`）。写入先落到 `.tmp-…`，commit 后整体替换目录；读取用 `read_parquet('/**/*.parquet')`。重复运行是整目录替换，不会累加重复行。 - `extensions` 是顶层字符串数组（如 `["parquet"]`、`["json"]`），运行前在 job 连接上执行 `load `；名称做字符白名单校验。`ducketl extensions` 列出 DuckDB 已安装/已加载的扩展。远程源（httpfs/sqlite/postgres/excel）仍需先自行 `INSTALL` 对应 DuckDB 扩展，项目不内置连接器。 ## TUI（M1 + M2 + M3） `ducketl tui` 打开工作台。默认用 `--backend terminal`（Raku `Terminal::UI` 自绘 TUI），`--backend textual` 切换到 Python Textual v8.2.8 TUI。两个 backend 共享同一套 core（Duck / Pipeline / Executor / Store / Config / SQL），Textual 版通过 JSON bridge + 子进程协同，core 零侵入。终端图表（sparkline / 条形图）、自绘边框、 Terminal 键位布局仅在 `--backend terminal` 下可用；Textual 版用原生 DataTable / ProgressBar / TabbedContent。 ### `--backend terminal`（默认，Raku Terminal::UI 自绘）最小 50×10；宽度 ≥100 时显示左侧导航，≥84 时显示右侧运行面板，窗口缩放会重建布局。界面由 TUI 自己绘制：圆角边框 + 内联标题、truecolor 配色（`NO_COLOR=1` 或 `DUCKETL_MONO=1` 关闭）、聚焦面板边框高亮、整行反色的选择条；表格列宽按内容自适应、超宽用 `…` 截断；运行/步骤状态用颜色区分（ok 绿、failed 红、canceled 黄、running 青）。 `Terminal::UI` 只提供备用屏与按键解析。终端图表是纯文本：Dashboard 表格下方是最近运行的 `rows/run`、`time/run` 迷你折线（sparkline + 范围），Run 屏步骤表下方是按耗时等比缩放的条形图（步骤失败时变红）。实现见 `lib/App/DuckETL/TUI/Chart`，不依赖任何绘图库。 | 屏 | 内容 | |---|---| | Dashboard | 最近 20 次运行（来自 history）；`Enter` 查看步骤、行数与错误 | | Sources | 当前数据库 catalog，加上当前 pipeline 的 `source` 步骤；`Enter`/`p` 预览（`LIMIT 100`） | | Pipeline | 可编辑草稿：步骤表 + 字段表单；`n` 新建、`Enter` 打开文件、`a` 加步骤、`s` 保存 | | Run | 步骤状态、行数、耗时、日志；`Enter` 查看单步 SQL/错误，`Esc` 返回 | | Query | ad-hoc SQL：`i` 在底部写语句、`e` 用 `$EDITOR`、`r`/`Enter` 执行；结果显示为表格（列名 + 行号，最多 200 行），错误同时显示在结果区和 SQL 面板 | | 键 | 作用 | |---|---| | `1` `2` `3` `4` `5` | Dashboard / Sources / Run / Pipeline / Query | | `Tab` / `Shift-Tab` | 在导航、内容、详情面板间切换 | | `j` `k` / `↑` `↓` | 移动选择；表单中移动字段 | | `Enter` | 激活选中项；Pipeline 中编辑当前字段（枚举/布尔值直接切换） | | `a` / `d` | 追加步骤 / 删除选中步骤（二次确认） | | `e` | 用 `$VISUAL` / `$EDITOR` 编辑选中步骤的 SQL，返回后恢复界面 | | `s` | 保存草稿；首次保存会提示路径（默认 `/pipelines/.json`） | | `o` / `n` | 打开选中 pipeline 文件 / 新建草稿（有未保存改动时确认） | | `r` / `F2` | 运行当前 pipeline；在 Pipeline 屏运行**草稿**（无需先保存） | | `c` / `F3` | 取消当前运行 | | `p` | 预览 Sources 中选中的表或 source | | `i` / `e` | Query 屏：底部单行输入 SQL / 用 `$EDITOR` 编辑 SQL | | `r` | Query 屏：执行查询；其他屏运行 pipeline | | `[` `]` | 切换 pipeline（同时刷新 catalog） | | `R` | 重新加载 pipeline、catalog、history | | `q` | 退出；有运行中的任务时先取消并等待回滚，有未保存草稿时确认 | 编辑草稿： - 表单字段随步骤类型/格式变化：`source` 的 `path`/`table`/`sql` 互斥展示，`sink` 的 `key` 只在 `upsert` 下出现。 - 文本字段用底部单行编辑器（插入、删除、←/→、Home/End、Enter 提交、Esc 取消）；`options` 以 JSON 对象文本编辑；`expect` 只接受整数。 - 保存前用临时文件走一遍 `Pipeline.load` 校验（id、引用、类型、必填项），校验通过才原子 `rename` 到目标路径；失败不会覆盖已有文件。 - 草稿标题带 `*` 表示未保存。未保存草稿运行或保存时，相对路径基准为当前工作目录；保存后基准变为 JSON 文件所在目录。 - `$EDITOR` 为空或未设置时回退到 `vi`；外部编辑器退出码非 0 时 SQL 保持不变。线程与连接： - `--backend terminal`：UI 线程只渲染快照和读按键；run 在后台线程执行，事件经队列交给 UI。数据 `Duck` 同时持有 job connection 与 session connection：catalog/预览走 session，不与运行事务抢临时表。运行取消通过驱动已绑定的 `duckdb_interrupt` 打断 job connection；事务回滚、staged 文件删除、history 记为 `canceled`。 - `--backend textual`：Python asyncio 原生——Bridge / WorkerRunner 全用 `asyncio.create_subprocess_exec`，事件通过 Textual `Message` + `post_message` 投递到主循环，UI 回调里 `await bridge.request()` 不阻塞渲染。DuckDB 文件锁冲突靠 bridge `release-all` 关 Duck+Store → Python spawn worker → worker 结束后 `acquire-all` 重开解决（Bridge 内部 `asyncio.Lock()` 保证 request 串行， DuckDB 单连接硬限制）。 - `--db` 优先于 pipeline 的 `database`，再回退到配置或 `:memory:`；`threads` / `memory_limit` 取 CLI、pipeline `defaults`、配置中的第一个可用值。 - catalog 不会为了浏览而创建尚不存在的数据库文件。隔离模式（`ducketl tui --isolated`）： - 每次运行启动一个 `ducketl worker` 子进程；数据库与 history 由 worker 独占，TUI 只消费 JSONL 事件。 - 运行期间 TUI 关闭自己的 history 连接，结束后自动重开、刷新 Dashboard；Catalog 与预览禁用。 - 草稿必须先保存（worker 只接收 pipeline 文件）；history 必须是文件，不能用 `:memory:`。 - `c` 通过 worker stdin 发送 `cancel`；worker 被 `kill -9` 或崩溃时 TUI 继续工作，运行标为 `failed`，未完成步骤标为失败。 - worker 的 stdout/stderr 是独立管道；异常不会写入 TUI 的终端。 ### `--backend textual`（Python Textual v8.2.8，asyncio 原生） DuckETL 核心模块零侵入。Python UI 同时 spawn 两个子进程，全 asyncio 管理： | 进程 | 作用 | 协议 | |------|------|------| | Raku bridge (`script/ducketl_tui_bridge.raku`) | 命令/响应：hello / catalog / query / ddl / history / pipeline-* / release-all / acquire-all / quit | stdin 每行 JSON command → stdout 每行 JSON response（`note` 只写 stderr，`say` 只写 stdout） | | `ducketl worker` | Pipeline 执行 + JSONL 事件流 | worker 是运行时 spawn（`release-all` 释放锁后），结束后 `acquire-all` 回收 | UI 5 Tab：Catalog / SQL / History / Pipeline / Run。事件机制： - Bridge 内部 `asyncio.Lock()` 保证同一时刻一个 request（DuckDB 单连接硬限制）。 - Bridge request 全 async：`await bridge.request("query", sql=..., limit=...)` 不阻塞 Textual 主循环。 - WorkerRunner 事件：`asyncio.create_task` + `run_in_executor` 读 worker stdout， `WorkerEvent(Message)` + `post_message` 投递回 Textual，reactive watcher 自动渲染进度条/日志/状态。 - Worker crash 检测：run 结束后检查 `worker.returncode`，非 0 则 error bar + Run 状态栏同时告警，`acquire-all` 照常执行回收 DuckDB 锁。 - Textual 缺失友好报错：UI 顶部 `try/except ImportError` 包 textual import，未安装时 stderr 打印 `pip install textual` + `ducketl tui --install` 提示后 exit 1。 - SQL Input 自带 `SQLSuggester`：候选 = catalog 表名（`schema.table` + 裸名）+ 常用 SQL 关键字，最多 20 条。文件位置： | 文件 | 行数 | 内容 | |------|------|------| | `script/ducketl_tui.py` | ~550 | Textual App + Bridge / WorkerRunner / WorkerEvent / SQLSuggester | | `script/ducketl_tui_bridge.raku` | ~200 | JSON 协议实现，Raku 端核心零侵入 | | `lib/App/DuckETL/CLI.rakumod` | — | `cmd-tui` 加 `--backend textual` 分支 + `--install` 子命令 | ## 运行语义 - **单进程**：一个数据 `DB::DuckDB` 实例；`run` 使用一条 job connection，TUI 另有一条 session connection；history 用独立 `.duckdb` 文件，避免跨库锁。 - **数据不过 Raku**：只执行 SQL 和 `COUNT(*)`；没有逐行读取。 - **事务**：`source`/`sql`/表 sink 在 run 级事务中；临时表、目标表、唯一索引一起提交或回滚。 - **文件原子性**：staged 文件 + `rename`。 - **history**：`_etl.runs`、`_etl.steps`、`_etl.watermarks`。水位只在对应 step 成功且 commit/文件安装完成后更新。 - **扩展**：pipeline 的 `extensions` 在事务开始前执行 `load`；失败即整次运行失败，不写目标。 - **分区目录**：`COPY ... PARTITION_BY` 先写 `.tmp-*`，成功后整目录替换；失败清理 staging，旧目录保留。 - **隔离 worker**：`worker` 在独立进程中打开数据库与 history，TUI 不打开数据库；worker 退出后才重新打开 history。 - **失败可见**：失败步骤的 SQL 与错误写入 history；`ducketl show ` 查看。 ## 当前边界 - TUI 编辑是单草稿、线性步骤：没有拖拽排序、多选、撤销栈；步骤顺序用删除/追加调整（M2 范围）。 - `options` 在表单里以 JSON 文本编辑，没有逐项控件；长 SQL 用 `$EDITOR`，内联输入是单行编辑器。 - 运行中不显示精确百分比；只显示已完成的步骤、行数与耗时（设计文档明确反对伪造进度）。 - 不做调度、多租户、远程编排、通用 DAG。 - 增量水位只支持表 sink 的 `append`/`upsert`；`replace` 与文件 sink 不支持水位。 - 分区写只支持 Parquet 目录 + `replace`；目录 SQL 需用 `read_parquet('.../**/*.parquet')`。 - 远程 HTTP / SQLite / Postgres / Excel 源需要自行安装对应 DuckDB 扩展；项目只提供 `extensions` 装载入口与 `ducketl extensions` 探测，不内置连接器。 - 隔离模式一次一个 worker，运行期间不能浏览 catalog/预览；草稿需先保存。 - 表名限制为 `[A-Za-z_][A-Za-z0-9_]*` 的 `table` 或 `schema.table`；其他标识符暂不支持。 - 步骤 id 在 SQL 的表位置具有保留含义；引用当前或未来步骤会报错。 ## 目录 ```text bin/ducketl CLI 入口 lib/App/DuckETL/Pipeline 模型、校验、步骤引用检查 lib/App/DuckETL/SQL quote/literal、步骤引用扫描与重写 lib/App/DuckETL/Planner 步骤 -> SQL（纯函数） lib/App/DuckETL/Duck 连接、catalog、schema、预览 lib/App/DuckETL/Executor 事务、事件、取消点、水位、分区安装、失败清理 lib/App/DuckETL/Run 后台运行、事件队列、取消 lib/App/DuckETL/IsolatedRun 子进程 worker、JSONL 事件、取消/崩溃处理 lib/App/DuckETL/Repo 当前 repository 链 -> 子进程 -I 参数 lib/App/DuckETL/Store history / watermarks（线程安全） lib/App/DuckETL/CLI run / tui / query / validate / list / show lib/App/DuckETL/TUI 布局、四屏状态机、键位、表单与草稿 lib/App/DuckETL/TUI/Draft 草稿字段/枚举/序列化/校验辅助 lib/App/DuckETL/TUI/Render 纯渲染：宽度感知表格、状态行 lib/App/DuckETL/TUI/Chart 纯文本图表：sparkline、条形图、区间标签 lib/App/DuckETL/TUI/Theme truecolor 调色板与着色（NO_COLOR 关闭） lib/App/DuckETL/TUI/Region 面板矩形与内边距（自绘边框用） lib/App/DuckETL/REPL 交互式 REPL（dot-command + SQL 行执行） script/ducketl_tui.py Python Textual v8.2.8 TUI（--backend textual，asyncio 原生） script/ducketl_tui_bridge.raku Raku JSON bridge（Textual 版命令/响应协议实现） examples/ 可直接运行的 pipeline 与样例数据 t/ 单元 + 集成测试 ``` ## 测试 ```sh prove6 -l t ``` 覆盖：SQL 引用扫描（函数/字符串/注释/限定名）、pipeline 校验、Planner golden SQL、CSV→Parquet、upsert 幂等、失败回滚、文件原子性、CLI、TUI 纯渲染（含 CJK 列宽）、后台运行事件与取消、草稿字段/枚举/序列化与保存后往返、ad-hoc SQL 查询（表格/JSON/错误/limit）、增量水位幂等、分区目录替换、扩展装载、隔离 worker 事件/取消/崩溃。本机验收： - **M0**：1,000,000 行 CSV → 过滤/转型 → Parquet，约 **346 ms**（Rakudo 2026.09、DuckDB 2.0.0-dev1、WSL Ubuntu；含启动与 history 写入）。 - **M1**：同一 1M 行 pipeline 运行期间切换 Dashboard/Sources/Run 不阻塞，约 **277 ms** 完成；慢查询运行中按 `c` 取消，UI 在 **0.75 s** 内显示 `canceled`，history 记为 `canceled`，目标文件无残留。 - **M2**：在 TUI 从零新建草稿，用表单建 `source → sql → sink` 三步，并用 `$EDITOR` 改写 SQL；保存 JSON 后运行，源 3 行 → 过滤 2 行 → Parquet 2 行（`sum=50`）。再加 `assert` 步骤后保存并重跑，4 步全部 `ok`。 - **M3**：带水位的 append 连续运行插入 **3 → 0 → 1** 行（新增一行后），目标表 4 行、水位推进到 `2026-01-04`；分区写生成 `country=UK` / `country=US` 两个目录，重跑替换后仍为 3 行且无 staging 残留；`extensions: ["parquet"]` 成功，未知扩展使运行失败且不写目标。`--isolated` 下 worker 正常运行、stdin `cancel` 约 **0.16 s** 生效；`kill -9` worker 后 TUI 保持可操作、运行标为 `failed`，退出码 0。