[FANNG1]

背景

对照 Daft v0.7.15 源码 review 后发现：项目里定义的大部分调优参数从未真正传给 Daft，而且大规模在 Ray 上跑时并行度会坍缩成单个 task。

• config.py 里的 DAFT_* 变量是死配置——项目从未调用 daft.set_execution_config，而 Daft 原生只从环境读取少数几个 DAFT_* 变量（DaftExecutionConfig::from_env）。
• daft_io_config() 没有把任何 S3 调优字段传进去（max_connections、重试、超时、retry mode）。
• manifest 是几 MB 的小文件，按字节的 scan task 切分不会生效：整条 download+分析链路在 Ray 上只有 1 个 task，集群再大也吃不满。
• Daft 默认 morsel 131072 行，对携带图片/音频字节的行来说是内存炸弹。

改动

接通死配置

• configure_daft_runner() 对 native 和 Ray 两种 runner 都调用 daft.set_execution_config(...)（morsel size、dynamic batching、scan task 切分/合并、JSON 目标文件大小、maintain_order、actor 就绪超时）。
• S3Config 补齐 max_connections / num_tries / 退避 / 连接+读取超时 / retry_mode。

修复并行度坍缩

• 新增 ANALYZE_NUM_PARTITIONS（默认 auto）：spread_partitions() 在下载前切分 manifest；Ray 模式自动推导为 2× 集群 CPU 数。image 和 audio 两个 analyze 都已接入。
• coalesce_for_write() 在 write_json 前把分区收敛到约 N/8，高并行度不会转化为大量小输出文件。
• 四处 write_lance 调用全部传入 LANCE_MAX_ROWS_PER_FILE / LANCE_MAX_BYTES_PER_FILE（此前也是死配置），在 pylance 7 禁用 blob compaction 期间由写入侧控制 fragment 大小。

稳定性

• 四个模型 UDF（ImageQualityUDF、_ImageEmbedUDF、AsrUDF、_EmbedUDF）加 max_retries=2。
• DAFT_ACTOR_UDF_READY_TIMEOUT 默认 300 → 600（大量 actor 同时冷启动 + 首次下载模型的场景）。
• DeepSeek llm_prompt 增加 concurrency=DEEPSEEK_CONCURRENCY（新配置，默认 8）和 on_error="ignore"，由已有的 try_deserialize/fill_null 兜底 null。
• README 新增 "Tuning for large Ray runs" 小节，含 Daft 事件日志环境变量。

验证

• 全量测试通过：87 个，其中 6 个为新增（S3 配置接通、spread_partitions 各分支、coalesce_for_write）。
• 手工确认 set_execution_config 生效（default_morsel_size=32、actor_udf_ready_timeout=600 等）。
• 本地 Ray 实测：20 行 manifest 读入为 1 个分区 → 自动切分为 20（2×10 CPU），行数不丢。