ingestr 把 MySQL 订单表按天搬进 DuckDB：默认 replace 和增量策略的取舍

ingestr 是一个命令行式的数据摄取工具：用 flag 描述源、目标和表，不用写代码。文档在这里。

安装有两条路：

curl -LsSf https://getbruin.com/install/ingestr | sh
# 或
pip install ingestr

要用 Python SDK（Arrow IPC 流）就 pip install 'ingestr[sdk]'，可以在代码里 ingestr.ingest([{...}], dest_uri="duckdb:///tmp/warehouse.duckdb", dest_table="main.people")，也可以用 context manager 分批喂数据，大块物化数据走 transport="mmap"；不想离开 Python 调 CLI 就用 ingestr.run([...])。

支持的源和目标是同一套数据库清单：Postgres、MySQL、SQL Server、MongoDB、SQLite、DuckDB、ClickHouse、Snowflake、BigQuery、Redshift、Databricks、Trino、StarRocks、Cassandra、DynamoDB、Elasticsearch、Oracle、SAP Hana、Vertica、CrateDB、PlanetScale、Vitess、MotherDuck，Iceberg 只能当目标；文件侧支持 CSV/JSONL/Parquet；SaaS 侧有 100 多个，Stripe、Shopify、HubSpot、Notion、GitHub、Slack、Salesforce 都在内。MySQL、Postgres、MongoDB、SQL Server、Vitess、PlanetScale 带 CDC 列。

场景：线上 MySQL 的 orders，按天进分析库

线上库里有一张 orders，每天凌晨把前一天的数据搬到分析侧。本地先用 DuckDB 验证，跑通一条命令：

ingestr ingest \
  --source-uri 'mysql://user:pass@127.0.0.1:3306/shop?sslmode=disable' \
  --source-table 'orders' \
  --dest-uri 'duckdb:///tmp/warehouse.duckdb' \
  --dest-table 'main.orders' \
  --incremental-key created_at \
  --incremental-strategy delete+insert \
  --interval-start 2024-06-01 \
  --interval-end 2024-06-02

--source-uri 也可以用环境变量 SOURCE_URI 或 INGESTR_SOURCE_URI 传。--source-table 除了表名，还支持 query:SELECT ... 形式，可以只挑需要的列或者带上源端过滤；配套还有 --sql-limit、--sql-exclude-columns、--columns（类型覆盖/改名，比如 'id:bigint,first_name:varchar(50):fname,email::eml'）。CDC 场景下 --source-table 可以省略，一次读多表。

跑完之后目标表会多一列 _ingestr_loaded_at，记录这行是什么时候被加载进来的。不需要的话加 --no-load-timestamp。想跳过确认直接跑，加 --yes。

增量策略怎么选

不传 --incremental-strategy 时，默认是 replace——每次把目标表整个换掉。这个默认值最容易踩：数据量小的维表没问题，订单表按天跑几次就会发现目标库里只剩最后一次的结果。可选值有 replace、truncate+insert、append、delete+insert、merge、scd2、none。

delete+insert 是按区间覆盖，适合订单表这种「某天的数据可能被修正」的场景。它需要一个定义切片的 --incremental-key，必须是 date/timestamp/numeric 类型，比如分区日期、批次号。用 id 这种行主键通常是错的。流程是先把读到的行放进 staging，算出这批行覆盖的区间：

start_bound = COALESCE(interval-start, MIN(staging.incremental_key))
end_bound   = COALESCE(interval-end,   MAX(staging.incremental_key))
DELETE ... WHERE incremental_key BETWEEN start_bound AND end_bound
INSERT SELECT FROM staging

传了 --interval-start / --interval-end 的话，ingestr 还会把边界下推到源端读取。两个边界里有一个缺失、又推不出来，就会跳过 delete。数值型 key 只有在能从 staging 行推断出边界时才支持；显式传的 --interval-start / --interval-end 会按 datetime 解析，所以显式边界主要给日期/时间戳 key 用。另外几点：不保证行顺序；不会按 incremental_key 去重；--primary-key 的去重行为取决于目标端；原子性也取决于目标端，ClickHouse 是先 ALTER TABLE DELETE mutation 再插入。CrateDB 已经不支持 delete+insert（之前是非原子的，属于破坏性变更），Trino 则通过 catalog 事务支持。

merge 依赖目标端有没有目标侧的 SQL match 条件。ClickHouse、StarRocks、CrateDB、Cassandra、DynamoDB、MongoDB、Iceberg、OneLake、discard 这些目标端的 merge 没有这个条件，所以不接受 --incremental-predicate（这个参数会拼到 merge 的 join 条件里）。要往这些库 merge，只能不加 predicate，靠 --primary-key 去重。

scd2 保留历史版本，适合用户画像、价格表这种要追字段变化的维度；append 只追加，适合日志、事件流这种源端不改历史行的；--full-refresh 直接重建目标表。

大表与并行

--page-size 默认 25000，控制每次从源端拉多少行。流式写入时 --flush-interval 默认 30s、--flush-records 默认 50000 攒一批。

一次读几千万行的时候，用 --extract-partition-by 指定源端的 date/time 或整型列，把有界读取切成并行窗口，--extract-partition-interval 控制窗口大小，可以是 1h、7d、整型步长或 auto，--extract-parallelism 控并行度。按天搬订单表，--extract-partition-by created_at --extract-partition-interval 7d 就能把一批拆成多窗口。大表写 BigQuery 之类的目标端时，--staging-bucket 指向 gs:// 或 s3://，配合 --staging-dataset 先把数据落 staging，再进目标表。分区和聚簇用 --partition-by / --cluster-by。

--progress 默认 interactive，可选 log 或 json，接到 CI 或者采集日志时用后者。--stream 走 CDC 源和消息队列（Kafka），配合 --flush-interval / --flush-records；--metrics-addr 暴露 expvar 的 /debug/vars，只在 --stream 下有效。

脱敏与 schema 契约

--mask 的格式是 column:algorithm[:param]，算法包括 hash、sha256、md5、hmac、email、phone、credit_card、ssn、redact、stars、fixed、random、partial、first_letter、uuid、sequential、round、range、noise、date_shift、year_only、month_year。手机号、邮箱这类直接 --mask 'phone:phone'、--mask 'email:email'。

--schema-contract 默认 evolve：源端加列，目标表跟着加。另外三个值：freeze 忽略新列，discard_row 遇到新列整行丢掉，discard_value 只丢新列的值。--schema-naming 默认 direct，可选 snake_case、auto。--no-inference 关掉类型推断，--trim-whitespace 去掉字符串首尾空格。

什么时候不该用它

需要跨源 join、窗口函数或复杂转换的时候，ingestr 只做搬运，不在中间写逻辑。delete+insert 的原子性取决于目标端，ClickHouse 走 mutation，执行期间读到中间状态是可能的，对强一致有要求就得换路子。没有可靠的日期/批次列做 --incremental-key 的表，增量只能退化成 replace 或 append，每次全量重跑。

还有一条许可证边界：ingestr 采用 Functional Source License 1.1，属于 source-available。内部生产、开发、测试、教育、研究、专业服务都可以免费用；不能用来提供商业性的竞品摄取/ELT/connector/托管数据管道产品。每个版本发布两年后自动转为 Apache 2.0。