如何优化 DuckDB/Polars 在 AWS Graviton3 上处理 38M+ 行的工作流？

先减少重复处理，再调整内存和磁盘

这条流水线可以继续使用 Polars 和 DuckDB，优先调整以下几处：

• 将 CSV 解析结果保存为可复用的 Parquet 中间层。修改 regex 或关联逻辑时，只重跑受影响的阶段。
• 用容器限制整个 Python 任务的内存，并给 DuckDB 设置更低的内存预算，在本地验证全量数据的溢写行为。
• 按实际磁盘配置 temp_directory，检查关联是否导致数据异常膨胀，再决定是否提高 EBS 性能或更换实例。

MotherDuck 可以用于需要反复调试和查询的数据丰富阶段，是否省钱还要实测。采用 Zero-ETL 后，CSV 清洗、regex 过滤和业务关联也仍需处理。

配置磁盘前要注意：r7g.xlarge 配备 4 vCPU、32 GiB 内存，没有本地实例存储；r7gd 系列才提供本地 NVMe。在 r7g 上看到 /dev/nvme...，也不能据此认定它是本地盘，因为 EBS 同样可能通过 NVMe 接口呈现。AWS 实例规格

如何在 8GB 本地环境验证全量任务？

可以在限制资源后运行全量数据，前提是本地磁盘能放下输入、中间结果和临时文件。100GB CSV 无需一次全部装入内存，但关联、排序等操作可能产生很大的中间状态。

需要同时设置两层限制：

memory_limit 不是整个 Python 进程的内存硬上限。DuckDB 的部分内存分配会绕过缓冲区管理器，Polars 的内存也不受这个参数控制。DuckDB 内存不足排查

如果已构建包含脚本和固定依赖的 cnpj-pipeline:dev 镜像，可以这样运行：

docker run --rm \
  --memory=5g \
  --memory-swap=5g \
  --cpus=2 \
  --mount "type=bind,src=$PWD/data,dst=/data" \
  --mount "type=bind,src=$PWD/scratch,dst=/scratch" \
  -e POLARS_MAX_THREADS=2 \
  cnpj-pipeline:dev \
  python /app/pipeline.py

运行前先在宿主机上创建 data 和 scratch 目录。在支持相应限制的 Linux 环境中，将 --memory 与 --memory-swap 设为相同值后，容器便不能使用 swap，可以避免操作系统换页掩盖任务本身的内存问题。macOS、Windows 使用 Docker Desktop 时，还要考虑其 Linux 虚拟机的资源分配。Docker 资源限制

DuckDB 可以先用以下配置测试：

SET memory_limit = '3GB';
SET threads = 2;
SET preserve_insertion_order = false;
SET temp_directory = '/scratch/duckdb/run-001';
SET max_temp_directory_size = '80GB';

这些数值只是测试起点，并非最优配置。临时空间限额必须小于实际可用空间，还要为输出文件留出余量。部署到 32 GiB 的 r7g.xlarge 后，如果两个引擎分阶段执行，可以从 DuckDB memory_limit = '18GB' 开始，观察内存峰值后逐步调整。

本地低内存测试能暴露溢写、算子内存增长和磁盘容量问题，但无法复现云端 32 GiB 内存下的执行时间，也不能保证任务一定能在云端完成。内存预算、并发度和存储设备的变化，都可能改变执行行为。

日常开发应保留一组小型自动化正确性用例，覆盖空值、超长活动列表、重复关联键及 regex 边界。全量受限运行用于阶段性验收，无需每次修改后都执行。

如何缩短修改 regex 后的迭代周期？

将流水线拆成可以单独重跑的阶段：

原始 CSV
  → 统一类型和编码的 Parquet
  → Polars 过滤后的 Parquet
  → DuckDB 关联后的 Parquet

第一层应保留后续规则可能用到的字段。修改 regex 时，从这一层重新过滤；只修改地理参考表时，从过滤结果重新关联，省去反复解析全部 CSV 的时间。

读取 CSV 时，要明确指定分隔符、表头、字符编码和字段类型。CNPJ、地理代码等标识符应按字符串处理，避免丢失前导零。遇到编码错误，应按源文件编码正确转码，不能用有损 UTF-8 解码绕过。

Polars 过滤阶段可以逐个处理已有分片：

import os
from pathlib import Path

import polars as pl

# 使用已经通过业务用例验证的规则。
pattern = os.environ["ACTIVITY_REGEX"]

for source in sorted(Path("/data/normalized").glob("*.parquet")):
    target = Path("/data/filtered") / source.name
    target.parent.mkdir(parents=True, exist_ok=True)

    (
        pl.scan_parquet(source)
        .select(
            "cnpj",
            "municipality_code",
            "secondary_activities",
        )
        .filter(
            pl.col("secondary_activities")
            .str.contains(pattern)
            .fill_null(False)
        )
        .sink_parquet(
            target,
            compression="zstd",
            engine="streaming",
        )
    )

示例列名需替换为实际 schema。这里用 sink_parquet 直接输出，避免先通过 collect() 生成完整 DataFrame，再交给 DuckDB。流式执行有助于降低内存压力，但无法保证任意查询计划的内存占用都固定，应检查所用版本的执行计划。Polars 流式执行、sink_parquet API

Polars 的 regex 语法与 Python re 有差异。如果目前通过 Python 函数逐行执行规则，应先验证能否改用原生表达式，再核对匹配结果。Polars 字符串处理

保存中间 Parquet 会增加一次读写，因此适合缓存开发中反复使用、重算成本高的阶段，无需每个简单操作都落盘。

38M 行关联为什么会产生大量溢写？

先检查参与关联的数据。仅凭行数无法判断内存需求，宽字符串列、关联键基数和结果行数往往影响更大。

尤其要确认地理参考表的关联键是否唯一：

SELECT municipality_code, count(*) AS n
FROM read_parquet('/data/reference/geo.parquet')
GROUP BY municipality_code
HAVING count(*) > 1;

如果预期是多对一关联，这个查询应返回零行。发现重复时，要按业务规则修正键或选择参考记录，不能直接用 DISTINCT 隐藏问题。历史参考表可能需要使用“地理代码 + 生效时间”等复合条件。

关联时只选择需要的字段，直接导出结果：

COPY (
    SELECT
        f.cnpj,
        f.municipality_code,
        g.municipality_name
    FROM read_parquet('/data/filtered/*.parquet') AS f
    LEFT JOIN read_parquet('/data/reference/geo.parquet') AS g
        ON f.municipality_code = g.municipality_code
)
TO '/data/output/enriched.parquet'
(FORMAT PARQUET, COMPRESSION ZSTD);

输出目录需提前创建。生产中可以按月份或输入分片分别输出，方便失败后重跑。如果涉及跨分片去重、全局排序或窗口计算，还需要单独的全局阶段。

也要确认“12 个月数据”的含义。如果每个月都是完整快照，直接合并会保留同一主体在多个月份的记录。应先决定生成最新快照还是保留历史，并明确记录粒度。确实需要集合去重时才使用 UNION，普通拼接使用 UNION ALL。

先用 EXPLAIN 查看计划，再用 EXPLAIN ANALYZE 实际执行，观察耗时、过滤位置和关联后的行数。多个关联、排序和聚合集中在同一条查询中时，即使支持溢写，仍可能出现内存不足。DuckDB 工作负载调优

NVMe 和 EBS 的 TEMP_DIRECTORY 分别怎么配置？

temp_directory 决定临时文件的位置。要让 CPU 持续有数据可处理，还需要减少溢写量，并确认底层存储的持续性能。

使用本地 NVMe

如果换用带实例盘的 r7gd，先将实例盘挂载到明确的位置，再将临时目录指向它：

SET temp_directory = '/mnt/local-nvme/duckdb/run-001';
SET max_temp_directory_size = '120GB';

路径和容量只是示例，设置前要核对挂载点与剩余空间。临时目录应放在实际磁盘文件系统上，放到 tmpfs 会继续消耗内存。

本地实例盘适合存放可重建的临时数据。原始文件和最终结果需要持久化，任务完成后再上传或提交最终输出。核算空间时，要考虑输入缓存、溢写和输出同时存在的情况，不能按压缩后的 Parquet 大小推算临时空间。

使用 EBS

在 r7g 上，可以挂载专用 EBS 卷，用来存放临时文件：

SET temp_directory = '/mnt/ebs-scratch/duckdb/run-001';

尽量避免 CSV 读取、临时文件读写和最终结果写入同时争用一块繁忙的卷。增加卷后，仍需遵守实例级 EBS 总带宽限制。

gp3 的基础性能为 3,000 IOPS、125 MiB/s，容量、IOPS 和吞吐量可以分别配置。单纯扩大 gp3 容量不会自动提高吞吐量。EBS gp3 性能说明

还要检查实例上限。AWS 规格表中，r7g.xlarge 的 EBS 基线吞吐量为 156.25 MB/s，最高值为 1,250 MB/s。持续批任务不能直接按最高值估算全程性能。EC2 EBS 性能规格

调优时，结合 iostat -xz 1、任务日志和 CloudWatch 判断瓶颈：

• 吞吐量接近卷的配置值时，可以考虑增加 gp3 吞吐量。
• 延迟和队列增长时，要同时检查 IOPS 与实例限制。
• 实例级能力已经成为瓶颈时，继续增加卷性能的收益有限。
• 磁盘尚有余量，但 regex 或压缩阶段 CPU 已满时，应调整计算阶段。

也可以比较不同线程数的表现。内存紧张时，更多线程可能增加并行状态和溢写量，建议先比较 2 个和 4 个线程下的完整任务耗时。

MotherDuck 或 Zero-ETL 是否更划算？

Zero-ETL 通常针对特定、受支持的数据源和目标，自动处理数据同步。这条流水线仍需执行 CSV 解析、次要活动过滤和参考表关联，采用 Zero-ETL 并不会直接省掉这些处理成本。AWS Zero-ETL 说明

如果清洗后的数据保留在云端，开发者需要频繁修改 SQL、重复做数据丰富处理，或者多人需要查询同一份结果，可以尝试 MotherDuck。其混合执行机制能在本地与云端之间分配查询，但连接 MotherDuck 后，本地 Polars 代码不会自动改为远程执行。应通过执行计划确认计算位置和数据移动情况。MotherDuck 混合执行机制

不同使用方式可以优先尝试以下方案：

比较成本时，应使用同一份完整输入、相同转换规则和相同输出要求。记录总耗时、峰值内存、临时空间峰值及数据传输量，并核对输出行数和关联匹配率。

EC2 侧要计入实例、EBS 容量与额外性能、对象存储、重试和维护时间；MotherDuck 侧要计入所选套餐、计算、存储及导入导出涉及的费用。具体费用按当前方案核算，不能只比较小时单价。MotherDuck 定价

这条流水线建议先做好 Parquet 中间层和自动化部署：锁定依赖，使用同一版本的脚本与容器入口，本地在资源限制下验收，云端按参数执行全量任务。部署到 Graviton 时，使用原生 linux/arm64 构建，并确认依赖及扩展兼容。x86 本地机器上的模拟运行结果不适合作为性能依据。

完成后，再将最终关联阶段单独放到 MotherDuck 做一次对照，分别比较计算费用、等待时间和实例维护工作量的变化。

备注：内容仅供参考。