让数据管道自己写代码：代码组件，正在成为大模型时代的数据处理主入口

让数据管道自己写代码：代码组件，正在成为大模型时代的数据处理主入口

拖拽组件解决了 80% 的标准化搬运。
真正拖住数据团队的，是剩下 20% 说不清、每次都长得不一样的长尾逻辑。
大模型让这 20%，第一次可以规模化。

一、每个团队都有一张"写不进 SQL"的表

一张普通的订单表：order_id、amount、phone、payload、status、create_time。

同步进数仓前，需求是这样的：

• amount 在源库是分，要转成元，保留两位小数；
• phone 要脱敏，但只遮中间四位；
• payload 里的 items 数组，每个商品都要拆成单独一行；
• status = DELETED 的记录不能进下游；
• CDC 的 UPDATE 事件，还要处理 __before 里的变更前数据。

没有一个是"把 A 字段搬到 B 字段"。它们有判断、循环、数值精度、嵌套展开。

用 SQL 写，能写，但会写成一篇没人愿意 review 的"SQL 散文"；写成 UDF，要发版、要重启；拆成外部服务，链路就断成两截。

问题不在能力，而在位置——管道里缺一个能写代码的组件。

二、代码组件：把"写代码"变成一个节点

在 DataY 的画布上，它就是一个普通转换节点：拖进来、连上上下游、写代码、跑。

不启动独立服务，不引入第二套运行时，调度、重试、监控全部复用管道已有能力。对任务来说，它和 SQL 节点、Join 节点没有区别。

契约只有一个方法

类名必须是 UserScript，不能带 package；方法签名固定为 process(FlowFile, Map)，必须返回 FlowFile；一次处理一批数据，而不是一行调一次。

FlowFile 装着三样东西：

• 数据：JSON 数组 / 对象、CSV、文本、二进制；
• 属性：库名、表名、事件类型等随数据流动的上下文；
• 表元数据：字段名与类型，决定下游能否正确建表、映射。

context 提供 log、componentId、componentName，以及执行上下文。

一个真实的清洗脚本

import com.data.job.FlowFile;
import com.data.job.component.javascript.ScriptContext.LogFunction;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.Map;

public class UserScript {
public FlowFile process(FlowFile flowFile, Map<String, Object> context) {
LogFunction log = (LogFunction) context.get("log");
JSONArray rows = flowFile.getJsonArray();
if (rows == null) return flowFile; // 其它格式原样透传

JSONArray result = new JSONArray();
for (Object item : rows) {
JSONObject row = (JSONObject) item;

if ("DELETED".equals(row.getString("status"))) continue;

Long amount = row.getLong("amount");
if (amount != null) {
row.put("amount_yuan", BigDecimal.valueOf(amount)
.divide(BigDecimal.valueOf(100), 2, RoundingMode.HALF_UP));
}

String phone = row.getString("phone");
if (phone != null && phone.length() == 11) {
row.put("phone", phone.substring(0, 3) + "****" + phone.substring(7));
}

result.add(row);
}

flowFile.setJsonArray(result);
// 新增字段必须同步元数据，否则下游"看不见"它
flowFile.upsertColumnMeta("amount_yuan", "DECIMAL");

log.info("清洗完成，输出 " + result.size() + " 条");
return flowFile;
}
}

没有注解、没有依赖注入、没有配置文件。它只依赖 FlowFile 和 JDK，却能做到 SQL 很难表达、UDF 又要发版才能做到的事。

三、为什么是 Java？

很多平台的代码组件选 Python 或 JavaScript。DataY 选 Java，不是情怀，是工程约束下的最优解。

同进程，零跨语言开销。 数据不用序列化出进程再反序列化回来，也不用维护第二套运行时。对每天几千万行的任务，少一次跨进程拷贝，就是实打实的吞吐差异。

直接复用已有能力。 JSON 用 fastjson2，金额用 BigDecimal，加密、日期用 JDK 自带工具，还能调用企业内部沉淀的 Java SDK。

编译期挡住一半的错。 字段名写错、类型不匹配、方法签名不对，编译阶段就报错，还带行列号。Python 里把 amount 写成 amout，可能跑一个月才在某个空值分支炸掉。数据管道最怕的不是报错，是静默地算错。

部署零新增。 服务器不用装 Python、不用配 Node、不用额外容器。一个 JAR 起来，组件就能用——这也是 DataY "2 核 4G 跑完整链路"的一部分。

需要说明的是：脚本定位是纯数据处理。字段加工、结构转换、规则判断可以；直连数据库、访问文件、调用外部 HTTP 不行。需要这些能力时，用对应的输入输出组件，让链路边界保持清晰。

四、它是怎么跑起来的

脚本源码 → 内存 javac → 字节码 Map（主类 + lambda/匿名类辅助类）
→ 自定义 ClassLoader → 反射调用 process() → 下游

三个生产里必须处理的细节：

① Spring Boot 可执行 JAR 的类加载。 打成可执行 JAR 后，依赖都在 BOOT-INF/lib，java.class.path 指不到它们。只靠 -classpath 编译，FlowFile、fastjson2 全会"找不到符号"。所以还要通过运行时类加载器索引包内的类，兼容普通目录、JAR 和 fat JAR。

② 辅助类必须整批加载。 脚本里用了 lambda、匿名内部类，javac 会额外生成 UserScript$1。只加载主类，运行时会直接 NoClassDefFoundError。所以编译产物是一张类名到字节码的 Map。

③ 编译一次，反复执行。 脚本实例会被缓存。第一次执行时编译，之后复用同一实例——同一任务实例内，编译开销只发生一次。

小插曲：组件内部的类名至今还叫 JavaScriptComponent，但执行的是标准 Java，页面展示名是「JAVA脚本」。

五、大模型时代：从"兜底"变成"主入口"

到这里，代码组件还只是一个"能力强、门槛也高"的兜底方案。大模型改变了这个等式。

一个反直觉的判断

大模型进管道，最常见的做法是：每来一行数据，就调一次模型。

做语义理解时这是对的，比如从评论里抽问题类型。但拿它做规则化加工——金额换算、字段拆分、条件过滤——会撞上三个硬问题：

而大模型最擅长的，其实不是"处理数据"，是写处理数据的代码。

让模型写一次代码，让代码跑所有行。

四种做法的分水岭

这张表最关键的一列不是成本，是确定性。

数据管道是生产系统，不是聊天窗口。它可以接受"模型偶尔抽错一个标签"，但不能接受"金额今天按分算，明天按元算"。

代码，就是把一次不确定的智能，固化成一份确定的执行逻辑。

所以三者不是替代，是分工：SQL / 映射处理标准集合运算；大模型组件处理语义理解；代码组件 + 大模型生成处理规则复杂、SQL 难表达、又必须稳定可复现的长尾逻辑。

前两者决定管道的下限，后者决定上限。

六、DataY 已经把它做进了组件

在「JAVA脚本」的配置面板上，除了代码编辑器，还有一个按钮：AI 助手。

点进去，只有一句需求输入框、一排示例，以及"上游调试数据"的状态。但"能用"和"可靠"之间，隔着四道工程保障。

第一道：把契约写进提示词

模型必须知道这个组件的接口规范。提示词里明确约束：

• 不许有 package，必须是 public class UserScript；
• 必须实现 process(FlowFile, Map) 并返回 FlowFile；
• 新增 / 删除 / 重命名字段，必须同步调用对应的元数据方法；
• CDC 场景要处理 __before；
• 只处理认识的数据格式，其它原样透传；
• 运行在 JDK 17，不许用更高版本语法；
• 只做纯数据处理，不访问网络、文件、数据库。

这不是让模型自由发挥，而是先给它一份组件说明书，再让它填实现。

第二道：把真实字段喂给它

大模型写数据代码，最常见的翻车不是语法错，是臆造字段名。你说 amount，它写出 total_amount；你说手机号，它写成 mobile。编译能过，跑起来全是空值。

DataY 会在进入助手时自动拉取上游组件的调试采样——真实字段名、类型、样例值——一并放进提示词，并明确要求：字段必须来自这份元数据，不在其中就提示用户，不许猜。

这一句话，把"猜字段"变成了"查字段"。

第三道：让模型自己先编译一遍

助手挂了一个只读工具 validate_javascript。模型给出最终答案前，可以把代码交给它真实编译一次。

它复用的就是组件运行时的编译器——不是正则，不是模拟，是货真价实的 javac。失败会把行列号和诊断信息返回，模型据此修正再试。

理解需求 → 生成脚本 → 真实编译 → 报错？修正重试 → 编译通过 → 输出

编译器成了模型的"单元测试"。 交付到面前的代码，至少是能编译的。

第四道：抽取、人工确认、再试跑

模型回复里常夹着解释和 Markdown。前端按优先级抽取代码：```java 围栏 → 任意围栏 → 从 import / public class 截到最后一个右花括号。

抽出来不会自动保存。先看预览、看说明，确认后点"应用到编辑器"，再照常试跑、看日志、看输出。

AI 负责起草，人负责签字。

实际效果

需求：

根据 create_time 算出订单小时和是否周末；金额小于 100 算小额，100 到 1000 算中额，超过 1000 算大额；时间解析失败的记录打 time_invalid 标记，但不要丢。

生成的脚本：

import com.data.job.FlowFile;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import java.math.BigDecimal;
import java.time.DayOfWeek;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Map;

public class UserScript {
private static final DateTimeFormatter FMT =
DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

public FlowFile process(FlowFile flowFile, Map<String, Object> context) {
JSONArray rows = flowFile.getJsonArray();
if (rows == null) return flowFile;

JSONArray result = new JSONArray();
for (Object item : rows) {
JSONObject row = (JSONObject) item;

try {
LocalDateTime time = LocalDateTime.parse(row.getString("create_time"), FMT);
row.put("order_hour", time.getHour());
row.put("is_weekend", time.getDayOfWeek() == DayOfWeek.SATURDAY
|| time.getDayOfWeek() == DayOfWeek.SUNDAY);
} catch (Exception e) {
row.put("time_invalid", true);
}

BigDecimal amount = row.getBigDecimal("amount");
String level = "未知";
if (amount != null) {
if (amount.compareTo(BigDecimal.valueOf(100)) < 0) level = "小额";
else if (amount.compareTo(BigDecimal.valueOf(1000)) <= 0) level = "中额";
else level = "大额";
}
row.put("order_level", level);

result.add(row);
}

flowFile.setJsonArray(result);
flowFile.upsertColumnMeta("order_hour", "INTEGER");
flowFile.upsertColumnMeta("is_weekend", "BOOLEAN");
flowFile.upsertColumnMeta("time_invalid", "BOOLEAN");
flowFile.upsertColumnMeta("order_level", "VARCHAR");
return flowFile;
}
}

这段代码放在两年前，是熟手 Java 工程师半小时的工作量。现在，是一句话加几秒钟。

七、边界与实践

四条建议：

底线只有一条：AI 生成的代码，也要 review。 编译器只能保证它"能跑"，不能保证它"算对"。

八、关于 DataY

前面讲的代码组件，不是演示稿里的截图，而是 DataY 里已经上线的一个普通节点。

DataY 是一个开源、轻量、一体化的数据平台。设计原则只有一句话：

数据平台可以很小，功能不打折。

传统平台动辄十几台机器、上百 G 内存。DataY 的选择是：用一个 JAR，把数据集成、维度建模、任务调度、高性能数仓、智能问数装进去。

• 数据集成：十余种数据源，拖拽或 JSON 编排，全量 / 增量 / CDC，自动建表；
• 维度建模：星型 / 雪花模型，三种维度类型，事实表物化落表；
• 任务调度：Quartz Cron 精准到秒，DAG 编排，master / worker 可分离；
• 高性能数仓：内嵌 DuckDB，列存 + 向量化，可外挂 Doris / ClickHouse；
• 智能问数：自然语言 → 指标 → 维度 → 取数，RAG + 统一指标口径；
• MCP Harness：通过 /mcp/datay，让 AI Agent 直接操作平台对象。

代码组件就在 datay-core 引擎里，和 StreamJdbcInput、StreamSqlUnit、LlmComponent 一样，用注解注册成普通组件。这意味着：

AI 代码生成也不是独立产品，而是长在组件配置面板里的一个按钮：读上游采样、守组件契约、调真实编译器、把结果交回编辑器。用完即走，不改变任务的运行方式。

适合谁？没有专职大数据工程师的中小团队；想用 2 核 4G 快速验证完整链路的团队；需要私有化部署的团队（AI 可指向本地 OpenAI 兼容服务）；以及想读源码、看数据平台怎么造的开发者。

git clone https://github.com/yuyuejia/datay
cd datay
mvn clean package -DskipTests
java -jar datay-web/target/*.jar –DATAY_AI_API_KEY=sk-xxxxxxxxx
# 浏览器打开 http://localhost:8080，默认账号 admin/admin

• 产品首页：http://datay.yuyuejia.com.cn/
• 在线体验：http://datay-demo.yuyuejia.com.cn/
• GitHub：https://github.com/yuyuejia/datay
• Gitee：https://gitee.com/yuyuejia/datay

项目采用 Apache-2.0 协议。如果这篇文章对你有帮助，欢迎点个 Star，或转给正在被"清洗脚本"折磨的同事。

数据平台可以很小，但数据处理的上限，不该被组件的数量决定。