Data-Juicer+Ray+AFlow 多模态数据处理实践：从通算到通智一体的工业级落地

导读

本次分享介绍了通义实验室开源项目 DataJuicer 的发展历程：从 SIGMOD 发表的 1.0 纯文本预训练版本，到 NeurIPS Spotlight 接收的 2.0 多模态增强版。

主要内容包括以下几个部分：

1. Data-juicer  
2. AFlow：多模数据处理新范式  
3. 多模数据处理引擎设计  

分享嘉宾｜通义实验室
内容校对｜郭慧敏
出品社区｜DataFun  

---

Data-juicer

1. Great AI starts with great Data

随着大模型研究的深入，学术界和工业界正经历从“模型为中心”向“数据-模型协同开发”的范式转变。数据显示，2022 年以来，关于“Data & MLLMs”的论文数量持续增长，表明数据在模型训练中的核心地位日益凸显。数据质量、规模与结构已成为决定模型性能的关键因素。

早期大模型发展聚焦于架构优化与参数扩展，但随着模型规模逼近极限，单纯堆叠参数带来的性能提升趋于平缓。研究发现，仅在模型侧优化已出现收益递减，而高质量、多模态、语义丰富的数据成为突破瓶颈的核心驱动力。

未来趋势是构建“Data for Model”与“Model for Data”的双向闭环：一方面用模型指导数据筛选与增强，另一方面用数据反哺模型训练效率与泛化能力。该模式强调全生命周期协同，实现数据与算法的联合优化。

2. 基建侧挑战

在大模型与多模态融合背景下，数据处理面临功能性、可用性/易用性、效率-规模权衡三大基建挑战。传统系统难以应对语义理解、异构计算与高并发需求，亟需构建统一抽象、高效可扩展的新一代数据处理架构。

• 功能性挑战：从结构化到语义驱动的数据处理
  数据类型正从强结构化向半结构化及多模态（文本、图像、语音）演进，对系统的语义理解能力提出更高要求。如何定义“高质量”“多样性”“无害性”等模糊概念，并将其转化为可执行的约束条件，成为关键难点。这背后涉及价值观建模与人机协同标注机制的设计。

• 可用性与易用性挑战：开发生态迁移与统一抽象
  AI 开发主流语言已从 Java/Scala/SQL 转向 Python/PyTorch/HuggingFace，但底层系统仍依赖传统分布式框架。缺乏统一抽象层导致开发者需重复处理异质计算细节（如 IO、网络、异步），形成碎片化开发模式，阻碍工程效率提升与跨团队协作。

• 效率-规模权衡挑战：模型依赖与资源瓶颈并存
  AI 场景中计算日益依赖深度学习模型（如用于数据清洗、打分、去重），导致单节点资源需求激增。同时，长尾任务与大规模并发带来稳定性与弹性调度压力。如何在保证高吞吐的同时满足低延迟、高可用要求，是当前系统设计的核心矛盾。

3. 架构总览

Data-Juicer 是通义实验室开源的多模态数据处理框架，旨在解决大模型训练中的数据分析、清洗、合成与标注等难题。该平台通过原子化算子库、用户友好界面、分布式执行引擎及效用验证四大支柱，实现了从手动工程到自动化流水线的飞跃。

• 系统化与可复用：原子化算子与工具链
  Data-Juicer 提供了近 200 个标准化算子，涵盖过滤、映射、增强、标注等操作，适用于预训练和后训练场景。每个算子被设计为独立单元，支持组合使用，形成高度模块化的数据处理流水线。这种“乐高积木”式设计显著降低了开发门槛，并提高了代码复用性和维护性。此外，Data-Juicer 还提供了丰富的文档和最佳实践指南，帮助开发者快速上手并高效解决问题。

• 用户友好与高灵活性：零代码与智能交互
  为了降低技术门槛，Data-Juicer 支持低代码或无代码配置，提供 Web UI、CLI 工具及 RESTful API 接口。特别值得一提的是其 Agent 风格的交互模式，允许用户通过自然语言描述需求，系统自动匹配相应的算子和参数，并调用相关文档和参考实现，极大提升了用户体验。这种对话式数据处理方式使得即使是非技术人员也能轻松完成复杂任务，加速了项目迭代周期。

• 高效与鲁棒：分布式引擎与自适应优化
  基于 Ray 构建的分布式执行引擎是 Data-Juicer 的核心竞争力之一。它支持 CPU/GPU 混合计算与多粒度并行，内置容错机制与算子融合能力，确保在大规模作业中保持高性能和稳定性。Engine-Agnostic 设计使其能够适配多种底层计算环境，进一步增强了系统的灵活性和扩展性。这些特性共同作用，使 Data-Juicer 成为工业级应用的理想选择。

• 效用验证与易实践：支持数据-模型协同开发
  Data-Juicer 不仅关注数据处理本身，还集成了实验管理与性能评估功能，支持数据与模型联合迭代。譬如在 VBench 和 MMVP Bench 等基准测试，证明了其处理后的数据能显著提升模型性能。此外，Data-Juicer 强调小样本优化策略，能够在有限资源下实现高效试错和快速迭代，从而最大化投资回报率（ROI），助力科研人员和工程师更有效地开展工作。

4. Data-Juicer 算子池：多模态和模型化

Data-Juicer 构建了结构清晰、语义明确的算子体系，通过操作符类型、模态类型、功能类型与实现类型四个维度对 200+ 算子进行系统化组织。该分类体系不仅提升开发效率与可维护性，也为 Agent 自动编程和多模态数据工程提供坚实基础。

• 操作符类型
  Data-Juicer 将算子按行为抽象为 8 大类算子类型。譬如 Mapper 执行数据变换、增强或合成（如文本改写、图像裁剪）；Filter 则基于统计量或语义特征进行筛选（如语言识别、重复检测）。这些算子成数据流水线的基本构建单元，支持链式组合与批处理执行。

• 模态类型
  Data-Juicer 的能力演进经历了从 1.0 纯文本预训练数据处理，到 2.0 全面支持视频（Video）和音频（Audio）等富媒体模态的跨越。当前，平台正进一步探索在具身智能（Embodied AI）、自动驾驶等前沿场景中的应用，Data-Juicer 正逐步实现从“单模态过滤”到“多源感知与生成”的能力升级，为下一代 AI 系统提供高质量、语义丰富的训练数据基底。

• 功能类型
  按功能划分为基础清洗（Deduplication, Language ID）、质量评估（Perplexity, Safety）、数据增强（Back Translation）、合成生成（Synthetic Q&A）

• 实现类型
  每个算子均配备中英文说明、输入输出示例、效果演示、源码链接与单元测试，形成完整知识闭环。这种规范化设计极大提升了 Agent 自动编程能力，使其能高效调用算子库完成复杂任务，推动“对话式数据工程”落地。

5. 运行时层：自适应执行与优化

Data-Juicer 在运行时层通过统一抽象、动态优化与异构资源调度实现高性能数据处理。基于 DJ-Dataset 与 DJ-Operator 构建计算图，支持批处理、算子融合、重排序与容错机制，显著提升吞吐率与资源利用率，满足多模态大模型训练需求。

• 统一抽象：构建显式计算图
  Data-Juicer 在 Ray/HF-Data 等底层引擎之上构建了 DJ-Dataset 与 DJ-Operator 的统一编程接口，将用户逻辑转化为可分析的计算图。该图支持静态优化与动态调度，实现对批处理、样本级容错、算子融合等策略的统一管理，降低开发复杂度并提升执行效率。

• 关键观察：算子间共享与可交换性
  譬如多个算子在上下文或计算过程中存在共享可能，部分算子（如 Filter 类）具备交换性。例如轻量级过滤可前置以减少后续计算负载。同时，不同算子在 CPU/GPU 资源消耗与耗时上差异显著，需进行精细化调度与资源匹配。

• 关键步骤：数据切片与动态调优
  通过数据切片实现批处理与样本级容错，提升鲁棒性；利用运行时探针动态监测各算子开销，实时调整并行度与调度策略。结合算子融合与重排序技术，消除冗余 I/O 与重复计算（如多次抽帧），最大化资源利用率与执行效率。

• 异构计算适配：CPU-GPU 混合调度
  面对算子间资源需求差异（如 CPU 高频任务 vs GPU 模型推理），系统采用流式批处理思想，支持按工作负载自动分配计算资源。通过自适应超参配置与算子级并行度调控，实现异构环境下的高效协同与负载均衡。

6. DataJuicer x Ray

DataJuicer 与 Ray 深度集成，通过 DJRayExecutor 构建了面向多模态、异构计算场景的分布式执行框架。该架构在 Ray.data 基础上扩展了自动路由、格式验证、流式读取与 GPU 算子支持能力，实现从单机到集群的平滑扩展，显著提升大规模数据处理效率。

• 统一执行入口：DJRayExecutor 与 RayLoadStrategy
  DJRayExecutor 作为核心调度器，集成 RayLoadStrategy 实现对不同数据类型（如 PDF、JSON、VKPDA）和来源的自动路由与格式校验。支持字段级、结构级验证，确保输入数据一致性，并通过动态适配机制降低用户接入成本，提升系统健壮性与兼容性。

• 高效数据流处理：RayDataset 与 map_batches 优化
  基于 ray.data.Dataset.map_batches 实现批处理模式，支持 JSON 数据的流式读取与任务分发。将 CPU 算子映射为 Actor，GPU 算子封装为 Task，实现计算资源的自动分配与并行化执行，有效应对海量非结构化数据的高吞吐需求。

• 灵活输出与导出机制：RayExporter 支持多格式写入
  RayExporter 提供对 WebDataset、Parquet、HuggingFace 等主流格式的自动导出支持，结合智能路由策略实现输出路径与格式的动态配置。支持自定义导出逻辑，满足不同下游模型训练框架的数据输入要求，提升工程灵活性与可维护性。

• 算子分类与性能优化：常规 vs Ray 特定算子
  DataJuicer 将算子分为两类：常规算子（如 clean_ip_mapper、alphanumeric_filter）天然兼容 Ray 模式；Ray 特定算子（如 ray_bts_minhash_deduplicator）针对去重等高负载场景进行深度优化，利用 MinHash、文档指纹等算法实现高效去重。

7. 自动并发控制与资源调度

DataJuicer 在 Ray 分布式框架之上，构建了一套细粒度的自动并发控制与资源调度机制，旨在解决高负载场景下的 OOM 风险、资源利用率低与负载不均衡问题。通过资源建模与并行度动态推导，实现从“任务派发”到“智能交通管理”的跃迁。

• 核心目标：细粒度并发控制与资源优化
  系统聚焦于在 Ray 环境中实现对并发度的精细化调控，避免因内存溢出（OOM）导致任务失败，同时提升 GPU/CPU 资源利用率。在数据处理执行前进行一次性计算，预估资源需求，保障 pipeline 流动性与稳定性。

• 资源建模：异构资源映射与模式匹配
  根据算子类型建立资源模型：CUDA 算子映射至显存，采用 actor 模式；CPU 算子映射至内存，采用 task 模式。该建模支持异构计算环境下的资源隔离与高效调度，为后续并发推导提供基础依据。

• Ray 并行度范围自动推导：下界与上界约束
  下界：满足资源约束的最小可行并发，确保 pipeline 不阻塞，公式  

  $$ \sum (G_i \times R_i) \leq A $$  

  其中 $R_i$ 为算子 $i$ 消耗资源占比；$G_i$ 为算子 $i$ 被分配的并行度；$A$ 为剩余可分配资源占比；$\mathrm{std}(R_i/G_i)$ 最小 → 负载均衡。

  上界：单算子理论极限 $\lfloor A/R_i \rfloor$，防止过度并发引发资源争抢。

8. 数据分区、检查点与事件日志

DataJuicer 在 Ray 基础上引入 PartitionedRayExecutor，通过数据分区、检查点机制与事件日志系统实现任务的高可用性与可调试性。该架构支持大规模数据集的容错恢复与实时监控，显著提升系统稳定性与工程效率。

• 容错性与可观测性
  系统构建全面的事件日志记录体系，覆盖所有执行阶段，并支持实时监控与异常告警。结合检查点机制，实现任务中断后的自动恢复，避免因单点失败导致全量重跑。例如在 10T 数据处理中，仅需重放失败分区，大幅降低时间与资源成本。

• 可扩展、性能
  内置检查点管理器支持状态持久化、检查点写入与清理，确保中间结果可持久保存。在任务失败时，系统可从最近检查点恢复，避免重复计算。同时支持多级检查点策略，平衡存储开销与恢复粒度。

  采用分区引擎对输入数据进行自动切片（如按样本划分），支持 JSON/Parquet 等格式的大规模数据集。每个分区独立处理，可在 Ray 集群中并行执行，提升吞吐率。分区策略灵活可配置，适配不同数据规模与业务需求。

9. 典型工作流

DataJuicer 采用“输入 → 分区 → 并行处理 → 合并输出”的典型工作流。将大文件按样本切分为多个分区（如 10 万条→1 万条/块），由 Ray 节点并行执行算子处理，支持中间检查点保存与事件日志记录。最终通过结果合并器完成聚合、一致性校验与性能分析，实现高吞吐、可恢复的数据流水线。

10. 算子优化示例：ray_bts_minhash_deduplicator

ray_bts_minhash_deduplicator 是 DataJuicer 针对大规模去重场景设计的高性能算子，通过算法创新与实现层优化克服 Ray 原生 GroupBy 的瓶颈。该算子融合负载均衡查集（BTS）、Hash 分组与 Map-side 预聚合技术，显著降低通信开销与 head 节点压力。

• 算法层面：引入 BTS 与 Hash 分组
  引入负载均衡并查集（BTS）高效维护数据分片归属，支持低开销合并与分裂操作。基于 Hash 进行聚合分组，缓解热点 Key 导致的数据倾斜问题，提升全局去重效率与资源利用率。

• 实现层面：减少对象注册与通信开销
  避免 Ray 原生 GroupBy 的碎片化问题，通过在每个节点设置 IDGenerator 将 Hash GroupBy 逻辑下沉至 worker 层。采用 Map-side 预聚合策略，减少对全局 Object Store 的访问，降低网络带宽压力。

  针对 MinHash 中的 Hash 计算、Signature 生成等性能瓶颈，使用 Cython 重写关键模块，实现 Python 到 C++ 级别的性能跃升。结合本地化计算与内存优化，大幅缩短处理延迟，提升整体吞吐率。

11. 数据生产：高性能规模化

DataJuicer 在高性能数据生产方面实现了显著突破，通过系统级优化支持超大规模数据处理。实验表明，其在 700 亿样本、6400 CPU 核上仅耗时 2 小时，3200 核下 70 亿样本仅需 0.45 小时，展现卓越可扩展性。针对去重任务，优化后性能较原生 Ray 提升 3.3 倍，1TB 数据在 8×160 CPU 上 30 分钟完成，而 GPU 版本（6×64A100）需约 1.8 小时，展现了优化潜力。结合算子重排、融合与批处理调优，进一步提升资源利用率与吞吐率，为大模型训练提供高效、稳定的数据基础设施支撑。

12. 接口层：智能体工作流

DataJuicer 通过与 AgentScope 深度集成，构建了面向自然语言交互的智能数据处理接口。该接口以大模型为引擎，结合标准化算子库与高质量知识底座，实现从模糊需求到可执行流水线的自动转化，显著降低使用门槛并提升开发效率。

• 智能体自动分析与算子推荐
  用户仅需用自然语言描述目标（如“生成用于 LlamaFactory 的 QA 数据”），智能体即可自动解析意图，检索内置算子库与预置数据菜谱（recipe），匹配最优处理链路。若现有能力不足，系统还能辅助定制新算子，调用编程大模型生成符合规范的代码及单元测试，实现端到端自动化。

• 动态编辑与可视化归因
  用户可在智能体推荐基础上，对数据菜谱进行动态调整。系统提供可视化处理结果（如清洗前后对比、去重率、分布变化）及细粒度数据归因（如某算子导致样本丢失原因），帮助用户理解流水线行为，并快速迭代优化配置，形成“推荐—编辑—验证—反馈”的高效闭环。

13. 应用 1：沙盒实验室 + 数据洞察

DataJuicer Sandbox 将数据处理与模型训练深度耦合，构建了面向多模态数据-模型协同开发的反馈驱动实验平台。通过引入 AB 测试思想，系统将数据集划分为多个可控池，对不同算子组合进行采样式干预，评估其对模型性能的影响。结合 OpenCampass、LlamaFactory 等模型侧开源框架，实现从数据处理到模型评测的闭环。用户无需完整跑完训练流程，即可基于采样结果快速验证数据策略有效性，显著提升研发效率。该方案支持成本可控的数据池构建、重要性分析、相关性与去重分析，推动数据工程从“经验驱动”向“实验驱动”演进，助力高质量模型迭代。

14. 应用 2：从静态先验数据到动态经验

DataJuicer 正从静态数据清洗迈向动态、模型感知的数据处理新范式。基于开源框架 Trinity-RFT（底层依托 Ray），团队实现了 Trainer、环境与 Data Buffer 的彻底解耦，构建支持长周期演进的经验池。该 Data Buffer 可原生接入 DataJuicer 算子，对强化学习中的经验数据进行实时分析、筛选与合成——例如回放高奖励样本、扩增有效经验或基于历史日志优化数据分布。通过引入在线数据选择与离线业务日志反馈机制，系统能动态识别“对当前模型最有价值的数据”，驱动模型持续迭代。该方法已在阿里闪购等场景落地，显著提升模型对话主动性与任务完成率，为模型依赖型数据工程提供了可扩展的实践路径。

15. 更多应用

DataJuicer 正在构建一个社区驱动的开放数据处理生态——DataJuicer Hub，汇聚真实场景下的最佳实践与可复用 recipe。平台系统性整理了面向 LLM/Agent 训练、RAG 优化、数据合成与评估等多类任务的标准化流程，涵盖 QA 生成、对齐理解、知识提取、安全清洗、多模态合成等典型场景。通过分类组织和持续沉淀，Hub 不仅为开发者提供即拿即用的“轮子”，也支持基于已有模板的快速迭代与优化。无论是 Agent 自动编排还是人工开发，用户均可从中获得高效、可靠的数据处理方案，推动从“重复造轮子”向“协同进化”的范式转变，加速大模型时代数据工程的工业化进程。

16. 社区 & 链接汇总

DataJuicer 已发展为一个开放、演进中的数据处理平台，核心包含 Operator 子库与分布式 Engine 两大模块。当前正与蚂蚁、PAI、AnyScale、NVIDIA-NeMo 等深度协作，推动引擎层在多粒度容错、大规模部署与自适应并行方面的系统化优化，致力于实现“透明分布式”体验。在应用层面，已拓展至自动驾驶、具身智能等前沿场景，探索物理引擎驱动的数据合成与验证。DataJuicer、Trinity-RFT 与 AgentScope 项目持续开源迭代，欢迎社区共同参与共建。我们鼓励开发者试用、反馈并贡献代码，一起推动大模型时代数据工程的智能化进程。

---

AFlow：多模数据处理新范式

1. 数据处理发展趋势：从单模态走向多模态

现实世界天然具备多模态特性，视频、图像、音频等非结构化数据与文本、表格等结构化数据共同构成复杂信息体。随着生成式人工智能（GenAI）技术突破，非结构化数据的商业价值被深度释放。DataJuicer 等框架推动多模态数据协同处理，实现跨模态对齐、融合与增强，为智能应用提供高质量输入。

• 现实世界的多模态本质
  真实场景中，信息以多种形态并存，如医疗诊断结合病历（结构化）与影像（非结构化）、零售推荐融合商品图与用户评论。传统单模态处理难以捕捉完整语义，而多模态融合可提升模型理解能力。例如蚂蚁“阿凡”系统通过整合身高、年龄等结构化特征与医学图像，实现精准辅助决策。

• GenAI 驱动非结构化数据价值释放
  生成式 AI 技术突破了传统机器学习对标注数据的依赖，使图像、语音等非结构化数据可通过大模型进行高效编码与推理。这推动企业从“数据孤岛”转向全域数据资产运营，尤其在医疗、金融、智能制造等领域，非结构化数据成为关键竞争力来源。

• 数据质量驱动模型性能：融合发挥更大价值
  数据质量是决定模型性能的核心因素。结构化数据提供精确上下文，非结构化数据补充丰富语义，二者融合可显著提升模型泛化能力。通过 DataJuicer 实现多模态清洗、增强与合成，确保输入数据的一致性与多样性，为下游任务（如分类、生成、检索）提供更优训练基础。

2. 数据处理发展趋势：从通算走向智算到通智一体

数据处理正经历从“通算”到“智算”，再到“通智一体”的范式跃迁。通算阶段以 Spark/Flink 为核心，聚焦结构化数据的规模化流转与高效计算；智算阶段依托 TensorFlow/PyTorch 等框架实现智能决策；而当前目标是打破引擎壁垒，构建统一任务流，实现通算与智算协同，提升端到端效率。

• 通算阶段：规模化数据处理基础
  通算阶段以“数据怎么存、怎么算更快”为核心，依赖 Spark、Flink 等批流一体化引擎处理结构化数据，存储层采用 HDFS 等分布式文件系统。该阶段解决了海量数据的高效读写与基础计算问题，为后续智能分析提供数据底座，但缺乏对非结构化数据的深度处理能力。

• 智算阶段：智能价值提取与模型驱动
  智算阶段聚焦“数据怎么用、怎么决策准”，通过 TensorFlow、PyTorch 等深度学习框架实现图像分类、文本理解等任务，结合 Horovod、DeepSpeed 等训练加速技术，以及 vLLM、SGLang 等推理引擎，推动模型性能持续提升，但通常独立于通算流程，形成“数据-模型”割裂。

• 通智一体阶段：打破割裂，实现协同融合
  通智一体旨在解决跨引擎数据移动与流程割裂问题，通过 Ray 实现异构资源统一调度，DataJuicer 提供多模态算子支持，配合多模态数据湖，将通算与智算任务整合至单一工作流中。AFlow 等框架可在一个任务中表达全链路逻辑，实现“1+1>2”的融合价值。

3. 多模数据处理引擎思考：案例分析

传统多模态数据处理流程虽逻辑清晰，但在实际工程中面临四大核心痛点。以消融实验前置数据流为例，原始数据经过滤、去重、质量评估等步骤后写入外部存储，看似完整，实则存在跨任务、跨平台、异构计算与命令式编程等问题，严重制约系统性能与开发效率。

• 跨任务：临时表串联导致存储爆炸
  数据节点间通过临时表（如中间结果表）进行串行交互，形成“任务链”结构。以网页数据处理为例，20～30 个环节累计产生 5～10PB 级中间存储开销，显著增加 IO 负载与存储成本。缺乏端到端数据流动机制，造成资源浪费和延迟放大。

• 跨平台：多引擎与语言混杂降低协同效率
  不同处理阶段依赖异构平台，如 Spark（Python/SQL）、VR 引擎或自定义脚本，涉及多种语言与接口规范。开发需在多个系统间切换，调试复杂度高，测试周期长，难以实现统一调度与监控，阻碍工程标准化进程。

• 异构计算：GPU/CPU 算子间通信瓶颈突出
  数据流中同时包含 CPU 算子（如文本清洗）与 GPU 算子（如图像增强），但二者通过 OSS/S3 等外部存储交互，引入高延迟与低吞吐的 I/O 开销。缺乏原生异构计算支持，无法实现计算资源高效协同，限制整体吞吐能力。

• 命令式编程：人工编排导致性能未达最优
  流程依赖手动编写执行顺序（如连续 Filter），未考虑操作符组合优化。例如将低选择率过滤器置于前端可大幅减少下游处理量，但人工干预难以全局优化。缺乏自动化的算子融合、重排与批处理机制，造成资源利用率低下。

4. 多模数据处理引擎思考：设计目标

针对传统拼接式数据处理的四大痛点，我们提出新一代多模态数据处理引擎的设计目标。该引擎以 DataJuicer 为核心，融合 Ray 的分布式能力，旨在实现通算与智算一体化、异构计算协同与自动化优化，推动数据工程向声明式、可复用、高性能方向演进。

• 可复用与组合的“通”+“智”算子
  提供标准化、模块化的“通”（如 Spark 式 ETL）与“智”（如 LLM 增强、图像分类）算子库，支持结构化与非结构化数据统一处理。用户可在单一作业中混合调用各类算子，描述端到端业务逻辑，提升代码复用性与系统可维护性。

• 支持 Python + SQL 混合编程
  兼顾算法工程师与数据开发者的使用习惯：Python 用于复杂模型推理与自定义逻辑，SQL 用于高效表达批流处理与聚合操作。在同一个作业中无缝切换语言，降低协作成本，提升表达灵活性与开发效率。

• 统一引擎支持本地调试与线上执行
  对外提供单一接口和运行时引擎，支持从本地快速验证到线上大规模部署的一体化流程。无需频繁切换环境或重构代码，显著缩短迭代周期，提升研发效能，实现“一次编写，随处运行”。

• 声明式接口与自动优化执行计划
  通过声明式 API 描述数据处理流程，由引擎自动进行任务重排、算子融合、资源调度与执行计划生成。例如将低选择率 Filter 提前、合并连续映射操作等，实现性能最优的多任务工作流，摆脱人工编排依赖。

---

多模数据处理引擎设计

1. 多模数据处理引擎设计：功能架构

AFlow 作为新一代多模态数据处理引擎，采用分层架构设计，整合 DataJuicer 智算能力与 Ray 分布式执行底座。从接入 API 到底层数据湖，实现通算与智算一体化、声明式编程与自动优化协同，为大模型训练提供高性能、可扩展的数据基础设施。

• 接入 API 层：支持作业全生命周期管理
  提供标准化接口，支持用户对 AFlow 作业进行提交、暂停、恢复、监控与终止等运维操作。通过 RESTful 或 CLI 接口实现与调度系统集成，便于自动化部署与资源编排，提升平台可观测性与管理效率。

• 开发 API 层：融合通算与智算算子
  集成 AiDataOps（通算类）与 DataJuicer Ops（智算类），支持结构化与非结构化数据处理。同时开放 UDF 扩展机制，允许用户自定义逻辑。开发者可在统一接口中混合使用 Python/SQL 编程，灵活表达复杂业务流程。

• 逻辑计划与优化器：实现智能执行优化
  AFLOW Logical Plan 描述用户定义的处理流程，Optimizer 负责生成最优执行计划。核心能力包括向量化执行（提升 CPU 利用率）、高性能 Shuffle（降低网络开销）与异构资源调度（GPU/CPU 协同），显著提升吞吐与延迟表现。

• 执行引擎与底层支撑：基于 Ray 的统一运行时
  AFLOW Engine 基于 Ray Core 构建，具备分布式任务调度、容错与弹性扩展能力。底层依托多模态数据湖（如 OSS/S3 + 元数据管理），实现统一存储与访问，打破数据孤岛，支持端到端数据流动与高并发处理。

2. 多模数据处理引擎设计：逻辑架构

AFlow 采用分层逻辑架构，实现从用户交互到物理执行的端到端闭环。系统涵盖产品层、开发接口层、逻辑算子层、优化层与物理层，通过统一 API 和智能优化机制，支持通算与智算融合，提升多模态数据处理的可编程性、性能与扩展性。

• 产品层与开发接口：提供高阶接入能力
  产品层为上层平台（如非结构化数据处理平台）提供 RESTful API 接口，实现 AFlow 的无缝集成。开发接口层支持 Python DataFrame 与 SQL 两种模式，兼顾算法工程师与数据开发者的使用习惯，降低学习成本并提升表达灵活性。

• 逻辑算子层：统一通算与智算操作原语
  该层是 DataFrame 与 SQL 的公共抽象，包含通用算子（如 Map、Filter、Agg、Join）和 AI 算子（如 AI_Classify、AI_Extract、AI_Filter）。支持集成 DataJuicer、AiDataOps 等第三方库，并开放 UDF 扩展机制，满足多样化业务需求。

• 优化层：智能执行计划生成与调度
  基于规则与成本模型，优化器执行过滤前置、算子融合、Stage 切分、Codegen 生成及多任务编排等策略。结合 Catalog Service 提供的元数据信息，自动生成高效执行计划，显著减少中间存储与计算开销。

• 物理层：分布式异构执行引擎
  物理层将优化后的逻辑计划映射为可执行任务流，通过 RayCore 实现分布式并行执行。支持 CPU 与 GPU 任务动态调度，涵盖 Decode、Frame Annotation、Classification、Write 等阶段，实现通算与智算算子的协同运行与资源高效利用。

3. 多模数据处理引擎设计：DataFrame API

AFlow 的 DataFrame API 作为核心开发接口，提供与主流 ETL 框架兼容的语义模型，支持结构化与非结构化数据的统一处理。通过标准化算子、灵活 UDF 扩展与跨库兼容机制，实现通算与智算能力的无缝集成，提升开发者效率与系统可维护性。

• 数据过滤：支持条件筛选与列选择
  提供基于谓词的行级过滤（如 df.filter(col > 10)）和列选择操作（如 select("name", "age")），支持复杂逻辑组合与动态条件构建。该功能等价于 SQL 中的 WHERE 和 SELECT 子句，适用于数据清洗与样本筛选场景。

• 数据转换：支持类型转换与自定义函数
  支持列计算、类型转换（如 string → int）及用户自定义转换函数（UDF）。例如对文本字段进行情感分析或图像路径解码，可通过装饰器快速封装为可复用函数，提升数据预处理灵活性与表达能力。

• 分组聚合与数据关联：支持多表联合处理
  内置多种聚合函数（如 sum、mean、count）与窗口函数，支持按键分组后执行统计操作。同时支持多种 Join 类型（内连接、外连接等），实现多源数据融合，满足复杂业务逻辑建模需求。

• UDF 支持与算子库兼容：实现灵活扩展
  通过 udf 装饰器支持用户自定义 Python 函数或类，可直接在 DataFrame 上调用。同时支持第三方算子库（如 DataJuicer、AiDataOps）无缝接入，只要符合标准接口即可无版本限制地使用，增强生态兼容性。

4. 多模数据处理引擎设计：DataFrame API — DataJuicer 算子简介

DataJuicer 作为 AFlow 的核心智算算子库，提供约 200 个标准化 Operator，涵盖聚合（aggregator）、去重（deduplicator）、过滤（filter）、格式化（formatter）、分组（grouper）、映射（mapper）、流水线（pipeline）和选择器（selector）八大类型。这些算子支持多模态数据的清洗、增强与合成，可无缝集成至 AFlow DataFrame API 中，实现从文本到图像、音频的统一处理流程，显著提升数据工程效率与模型训练质量。

5. 多模数据处理引擎设计：DataFrame API — 算子库兼容（Data-Juicer）

AFlow 通过统一接口实现对 DataJuicer 算子的无缝兼容，支持无参与有参两类 Operator 的灵活调用。用户可将 DataJuicer 算子实例化为 UDF，并直接嵌入 DataFrame 操作链中，实现多模态数据的清洗、增强与推理处理，提升开发效率与系统一致性。

• 无参算子
  对于无参数的 DataJuicer 算子（如 clean_html_mapper），用户可通过 aflow.datajuicer.operator() 构造实例，并将其作为普通 UDF 应用于 DataFrame 的列操作。例如对文本字段进行 HTML 清洗，无需额外配置即可完成自动化预处理流程。

• 有参算子
  针对需参数化的算子（如 text_classifier_mapper），支持传入 GPU 数量、模型 URL 等配置项。通过实例化后调用，可在运行时动态加载远程模型（如 OSS 上的 PyTorch 模型），实现轻量化部署与高并发推理，适用于大模型微调前的数据标注场景。

6. 多模数据处理引擎设计：SQL API

AFlow 提供兼容标准 SQL 语法的 SQL API，支持与 DataFrame API 无缝集成，实现通算与智算逻辑的混合编程。通过内置 UDF 服务对接机制，参考 ppt 右下侧示例用户可将自定义函数（如文本情感分析）直接嵌入 SQL 查询中，提升数据处理的表达能力与开发效率。

• 标准兼容性
  AFlow SQL API 完全兼容 ANSI SQL 标准，支持 SELECT、JOIN、WHERE、GROUP BY 等核心语法，降低用户学习成本。适用于结构化数据清洗、聚合分析等传统 ETL 场景，确保与现有系统平滑迁移。

• 扩展能力
  系统支持对接外部 UDF 服务，允许用户将 Python 编写的 AI 函数注册为 SQL 可调用函数。例如将 analyze_sentiment 注册为 UDF 后，可在 SQL 中直接使用，实现“SQL + 智能模型”的联合查询。

• 互操作性
  SQL 与 DataFrame API 实现双向互通，用户可在同一作业中混合使用两种模式。例如先用 DataFrame 加载和预处理数据，再用 SQL 执行复杂查询，最终结果可写回 Lance 或其他存储，构建灵活的数据流水线。

7. 多模数据处理引擎设计：逻辑层

AFlow 的逻辑层是连接用户 API 与物理执行的核心中间件，采用分层架构实现多模态数据处理的统一表达。通过 DataType、Expression 和 Logical Plan 三层抽象，支持结构化与非结构化数据的融合建模，为后续优化与执行提供标准化基础。

• DataType 层
  定义统一的数据类型体系，涵盖基础类型（如 Int、String）、复杂类型（如 Struct、Array）及扩展的多模态类型（如 Image、Audio、Video）。该设计支持跨模态数据的统一表示与序列化，为异构数据处理提供底层支撑。

• Expression 层
  将计算逻辑抽象为表达式树，包含 Column（列引用）、Literal（常量）、BinaryOp（二元运算）和 Function（函数调用）等节点。支持嵌套表达式与 UDF 调用，实现复杂业务逻辑的精确描述，便于后续优化器分析与重写。

• Logical Plan 层
  所有 DataFrame 与 SQL 操作均被解析为 Logical Plan，由 Project、Filter、Aggregate、Join 等逻辑算子构成。该结构类似于 Spark 的 Catalyst 优化器模型，支持跨 API 统一表示，为执行计划生成与优化提供基础。

8. 多模数据处理引擎设计：优化层

AFlow 的优化层分为 Logical Optimizer 与 Planner 两部分，分别负责逻辑计划的智能优化与物理执行的任务切分。通过 UDF 分离、算子融合等规则提升计算效率，并结合列/行级切分策略实现高并发执行，构建高性能多模态数据处理流水线。

• Logical Optimizer 层
  接收用户提交的 Logical Plan 后，应用通用 SQL 与多模态优化规则。核心能力包括：UDF 表达式从 Project 中分离以支持独立调度；算子融合（如 CPU I/O 或 AI_Filter）减少资源开销；延迟计算将嵌套操作拆分为独立步骤并赋予独立并发资源，提升执行灵活性与性能。

• Planner 层
  基于优化后的逻辑计划，Planner 支持按列与按行两种任务切分策略。按列切分将无血缘依赖的 Node 拆分为独立任务，提升并行度；按行切分则对超大规模数据源进行分片，每个分片生成独立任务，支持横向扩展与容错，确保高吞吐与低延迟执行。

9. 未来规划

我们正推进 AFlow 引擎的未来演进：在 API 层实现多算子库兼容，支持开源与内部业务算子；Optimizer 层引入代价模型，强化 HBO 能力；Engine 层聚焦高性能 Shuffle 优化，基于 RayData 构建外挂 ISS 方案，突破 Object Store 限制，提升性能并计划回馈社区。目标打造高效、灵活的大数据处理底座。

以上就是本次分享的内容，谢谢大家。