简介：bboss-datatran 是由 bboss 社区开源的数据采集与流批一体化处理工具，面向大数据开发工程师、数据平台建设者及需要构建数据湖、数据仓库或实时分析系统的技术团队。它支持 MySQL、Oracle、HBase、MongoDB、Kafka、RabbitMQ、HDFS、FTP 等多种数据源接入，提供数据类型转换、字段筛选、异常值处理、数据拼接等清洗规则，并可将结果写入 Hive、HBase、Elasticsearch、Greenplum 等存储系统，同时以流批一体架构统一处理实时流与批量历史数据，具备水平扩展、容错恢复、可视化监控与 API 集成能力。资源包共 614 个文件，以 453 个 Java 源码为核心，辅以 48 个 Markdown 说明文档、22 个 Gradle 构建脚本及若干配置文件，压缩包约 915KB，目录结构完整，便于二次开发与源码研读。目前已有 332 人学习下载，适合希望深入理解流批一体化数据处理实现细节的开发者参考。

1. bboss 流批一体化采集工具：从数据源到指标计算的一条龙方案

很多团队做数据采集，一开始都是脚本堆出来的：Python 写个定时任务拉接口，Logstash 接文件，再搞个 Flink 做实时。跑上三个月，采集脚本散落在七八台机器上，清洗逻辑改一处要动五个文件，指标口径对不上就开始扯皮。bboss 这套开源工具想解决的就是这个问题——它把数据采集、清洗转换、入库、指标统计计算串成一条流批一体化的链路，用一套配置和 API 覆盖实时和批量两种场景。如果你正在被多套采集链路维护成本折磨，或者需要从零搭一套能同时跑实时和离线的数据管道，这个方向值得花时间摸清楚。它适合有 Java 基础、对数据工程有基本认知的开发者，新手跟着配置走也能跑通最小链路。

2. bboss 采集链路的核心组件与流批一体化的实现逻辑

2.1 采集端：从数据源到清洗转换的管道设计

bboss 的采集端核心思路是「插件化数据源 + 管道式处理」。它内置了多种数据源适配器，关系型数据库、Elasticsearch、HTTP 接口、文件系统、消息队列都能作为采集起点。采集任务定义好之后，数据不是直接扔进目标端，而是先经过一条处理管道，管道上可以挂多个处理器节点，每个节点做一件事：字段映射、格式转换、过滤、脱敏、富化。

这套设计和 Logstash 的 pipeline 概念类似，但 bboss 的处理器是用 Java 写的，意味着你可以在处理器里写任意业务逻辑，不受 DSL 表达能力限制。比如从雪球这类财经数据源采集行情数据时，原始返回的 JSON 字段名可能是current、percent，你需要转成price、change_rate，同时把时间戳从毫秒转成标准格式，再过滤掉停牌股票。这些操作在 bboss 里就是一个转换处理器加一个过滤处理器的事。

流批一体化的关键在于：同一套采集配置和处理器链，既可以以「一次性批量执行」模式跑，也可以以「持续监听」模式跑。批量模式下，采集任务读取全量数据后执行一次管道处理然后结束；流式模式下，采集任务持续监听数据源变化，每来一批数据就触发一次管道处理。两种模式共用同一份处理器代码，不需要为实时和离线分别维护两套逻辑。

2.2 入库端：多目标写入与事务一致性保障

采集和清洗之后就是入库。bboss 支持的目标端包括 Elasticsearch、关系型数据库（MySQL、PostgreSQL、Oracle 等）、Kafka、HBase 等。入库端的设计要点有两个：批量写入的缓冲机制和多目标写入的一致性。

批量写入方面，bboss 维护了一个可配置的缓冲队列。数据从管道出来后先进入缓冲区，攒够一定条数或等待超过一定时间后触发批量提交。这个机制对 Elasticsearch 的_bulk接口和数据库的INSERT INTO ... VALUES批量插入都适用。缓冲区大小和刷新间隔是两个关键参数，设太小写入频繁、吞吐上不去，设太大内存占用高、故障时丢数据多。

多目标写入时，bboss 的做法是每个目标端独立维护自己的缓冲和提交逻辑，但共享同一份上游数据。如果目标 A 写入成功、目标 B 写入失败，bboss 会记录失败批次并支持重试。这里没有强分布式事务，实际使用中需要根据业务容忍度决定是「至少一次」还是「最多一次」语义。

2.3 指标统计计算：流批一体的聚合层怎么搭

指标统计是这条链路的下游环节。bboss 的指标计算支持两种模式：预聚合和即席查询。预聚合模式下，数据入库的同时触发指标计算任务，把结果写入指标表或指标索引；即席查询模式下，指标在查询时动态计算。

流批一体在指标层的体现是：实时场景下，指标计算任务监听数据变更事件，增量更新指标值；离线场景下，指标计算任务对历史数据做全量重算。两种模式共用同一套指标定义和计算逻辑。比如计算「某只股票近 5 分钟均价」，实时模式下每来一条行情数据就更新滑动窗口，离线模式下对历史行情数据做窗口聚合。bboss 通过统一的指标定义 DSL 来屏蔽两种模式的差异。

3. 用 bboss 跑通一条采集到入库的最小链路

3.1 环境准备与依赖引入

先确认本地环境：JDK 8 或以上，Maven 3.6+，一个可用的 Elasticsearch 实例（单机即可）和一个 MySQL 实例。bboss 的依赖通过 Maven 引入，核心包是bboss-datasource和bboss-datax。

<!-- pom.xml 中引入 bboss 核心依赖 --> <dependency> <groupId>com.bbossgroups</groupId> <artifactId>bboss-datasource</artifactId> <version>5.9.5</version> </dependency> <dependency> <groupId>com.bbossgroups</groupId> <artifactId>bboss-datax</artifactId> <version>5.9.5</version> </dependency>

版本号需要根据实际仓库中可用的最新版本调整，不要照抄。引入依赖后，在resources目录下创建application.properties，配置数据源连接信息。

# 数据源配置：MySQL 作为采集源 datasource.name=mysql_source datasource.driver=com.mysql.cj.jdbc.Driver datasource.url=jdbc:mysql://127.0.0.1:3306/stock_db?useUnicode=true&characterEncoding=utf8 datasource.username=root datasource.password=your_password # Elasticsearch 作为入库目标 elasticsearch.hosts=127.0.0.1:9200 elasticsearch.clusterName=docker-cluster

这里datasource.name是自定义的数据源标识，后续在采集任务中引用。Elasticsearch 的clusterName在 7.x 以上版本可以留空，bboss 会自动协商。

3.2 定义采集任务与清洗转换处理器

采集任务的定义有两种方式：XML 配置和 Java API。XML 方式适合标准化采集场景，Java API 适合需要动态逻辑的场景。先看 XML 方式。

<!-- 采集任务定义：从 MySQL 读取股票行情，清洗后写入 ES --> <properties> <property name="taskName" value="stock_quote_etl"/> <property name="datasource" value="mysql_source"/> <property name="sql" value="SELECT stock_code, current_price, change_percent, trade_time FROM stock_quote WHERE trade_time > ?"/> <property name="sqlParamType" value="timestamp"/> <property name="sqlParamName" value="lastMaxTradeTime"/> <property name="target" value="elasticsearch"/> <property name="targetIndex" value="stock_quote_index"/> <property name="batchSize" value="500"/> <property name="refreshInterval" value="5000"/> </properties>

这段配置的含义：从mysql_source数据源执行 SQL，查询trade_time大于上次最大时间的记录，每批 500 条，每 5 秒刷新一次。lastMaxTradeTime是 bboss 内置的增量采集参数，自动记录上次采集的最大时间戳，避免重复采集。

清洗转换处理器用 Java 实现：

// 自定义转换处理器：字段重命名 + 时间格式转换 + 空值过滤 public class StockQuoteTransform implements DataTransform { @Override public Object transform(Object data) { Map<String, Object> record = (Map<String, Object>) data; // 字段重命名 record.put("price", record.remove("current_price")); record.put("change_rate", record.remove("change_percent")); // 时间格式转换：毫秒时间戳转 ISO8601 Long tradeTime = (Long) record.get("trade_time"); if (tradeTime != null) { record.put("trade_time", new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSZ") .format(new Date(tradeTime))); } // 过滤无效记录 if (record.get("price") == null || ((Number) record.get("price")).doubleValue() <= 0) { return null; // 返回 null 表示丢弃该条记录 } return record; } }

处理器实现DataTransform接口，transform方法接收一条记录，返回处理后的记录或 null。返回 null 时 bboss 会丢弃该条数据，不进入后续管道。这个设计比抛异常更轻量，适合做数据质量过滤。

3.3 配置入库目标与批量写入参数

入库目标的配置决定了写入性能和可靠性。以 Elasticsearch 为例：

# ES 批量写入参数 elasticsearch.bulk.size=1000 elasticsearch.bulk.flushInterval=3000 elasticsearch.bulk.concurrent=4 elasticsearch.bulk.retry=3 elasticsearch.bulk.retryInterval=1000

bulk.size是单批提交的文档数，flushInterval是强制刷新间隔（毫秒），即使没攒够bulk.size也会在超时后提交。concurrent是并发写入线程数，根据 ES 集群的写入能力调整。retry和retryInterval控制失败重试策略。

数据库入库的配置类似，但参数名不同：

# 数据库批量写入参数 database.batch.size=500 database.batch.flushInterval=5000 database.batch.transaction=true

transaction=true表示批量写入在一个事务中提交，失败时整批回滚。如果业务允许部分成功，设为 false 可以提升吞吐但需要自己处理失败记录。

3.4 启动任务并验证数据链路

启动采集任务的代码：

// 启动采集任务 public class EtlBootstrap { public static void main(String[] args) throws Exception { // 初始化 bboss 上下文 BBossStartup.start(); // 注册转换处理器 DataTransformRegistry.register("stockQuoteTransform", new StockQuoteTransform()); // 加载采集任务配置 DataTaskConfig config = DataTaskConfig.load("stock_quote_etl.xml"); // 启动任务 DataTask task = DataTaskManager.startTask(config); // 注册任务状态监听 task.addListener(new TaskListener() { @Override public void onStatusChange(TaskStatus status) { System.out.println("Task status: " + status); } }); // 保持进程运行 Thread.currentThread().join(); } }

启动后验证数据链路是否通畅，分三步：第一步查 MySQL 源表有没有数据；第二步查 ES 目标索引的文档数是否与源表匹配；第三步抽查几条文档的字段格式是否正确。

# 查看 ES 索引文档数 curl -s "http://127.0.0.1:9200/stock_quote_index/_count" | python -m json.tool # 抽查一条文档 curl -s "http://127.0.0.1:9200/stock_quote_index/_search?size=1" | python -m json.tool

如果文档数为 0，先检查采集任务的日志级别是否开到 DEBUG，看 SQL 是否执行、返回了多少行。如果文档数对不上，检查转换处理器是否有过滤逻辑误杀了数据。

4. 采集链路跑起来后最容易翻车的几个地方

4.1 增量采集的时间戳回拨导致数据重复

现象：ES 里出现大量重复文档，trade_time相同的记录有多条。

原因：增量采集依赖lastMaxTradeTime参数，如果数据源的时间戳出现回拨（比如上游系统时间校准、时区切换），bboss 会认为有新的数据需要采集，导致重复拉取。

解决：在 SQL 里加去重条件，或者用stock_code + trade_time作为 ES 的文档_id，利用 ES 的幂等写入特性覆盖重复文档。bboss 支持在配置中指定_id字段：

<property name="targetIdField" value="stock_code,trade_time"/>

4.2 批量写入缓冲区溢出导致 OOM

现象：任务运行一段时间后 JVM 抛OutOfMemoryError，堆内存持续增长。

原因：bulk.size设得太大（比如 10000），或者flushInterval设得太长，数据在缓冲区里堆积。如果上游采集速度大于下游写入速度，缓冲区会无限增长。

解决：把bulk.size降到 500～1000，flushInterval设为 2000～5000 毫秒。同时给缓冲区设置上限，bboss 支持buffer.maxSize参数，超过上限时阻塞采集端而不是继续堆积。

4.3 转换处理器里的类型转换异常被静默吞掉

现象：部分记录没有进入目标端，但日志里没有明显的错误信息。

原因：转换处理器里做了强制类型转换，比如把 String 转 Long，遇到格式不对的数据时抛ClassCastException。bboss 默认对处理器异常做捕获并跳过该条记录，日志级别不够时看不到。

解决：在转换处理器里加 try-catch，把异常记录写到单独的日志文件或死信队列。同时把 bboss 的日志级别调到 DEBUG，确认异常是否被吞掉。

try { Long tradeTime = Long.parseLong(record.get("trade_time").toString()); record.put("trade_time", tradeTime); } catch (NumberFormatException e) { // 记录异常数据到死信队列 DeadLetterQueue.add(record, e.getMessage()); return null; }

4.4 ES 索引 mapping 冲突导致写入失败

现象：ES 返回mapper_parsing_exception，提示字段类型不匹配。

原因：第一次写入时 ES 自动推断字段类型，比如price推断为float，后续数据里price是字符串格式，写入就失败了。

解决：提前创建索引 mapping，明确每个字段的类型。bboss 支持在任务配置中指定 mapping 文件，任务启动时自动创建索引。

{ "mappings": { "properties": { "stock_code": {"type": "keyword"}, "price": {"type": "scaled_float", "scaling_factor": 100}, "change_rate": {"type": "float"}, "trade_time": {"type": "date", "format": "yyyy-MM-dd'T'HH:mm:ss.SSSZ"} } } }

4.5 流式模式下任务假死但无报错

现象：任务状态显示运行中，但目标端长时间没有新数据写入。

原因：流式采集依赖数据源的事件通知机制。如果数据源是数据库，bboss 通过轮询方式检测新数据，轮询间隔设得太长（比如 60 秒）会让人误以为任务挂了。如果数据源是消息队列，可能是消费者组 offset 提交失败导致重复消费或停止消费。

解决：先确认采集端的轮询间隔配置，数据库采集的refreshInterval建议设为 3000～5000 毫秒。消息队列采集检查 consumer 的session.timeout.ms和max.poll.interval.ms是否合理。另外给任务加心跳日志，每处理一批数据打一条 INFO 日志，方便判断任务是否在正常工作。

5. 指标计算层的进阶用法：滑动窗口聚合与离线重算

指标计算是这条链路里最容易做复杂的地方。我一般会先用 bboss 的预聚合模式跑通一个最小指标，再根据业务需求决定要不要上滑动窗口。

最小指标示例：计算每只股票的「最新价格」和「当日累计涨跌幅」。这个指标不需要窗口，每次数据更新时覆盖旧值即可。bboss 的指标定义如下：

// 定义指标：最新价格 + 当日累计涨跌幅 MetricDefinition latestPrice = MetricDefinition.builder() .name("latest_price") .groupBy("stock_code") .aggregate(AggregateFunction.LAST, "price") .build(); MetricDefinition dailyChange = MetricDefinition.builder() .name("daily_change_rate") .groupBy("stock_code") .aggregate(AggregateFunction.SUM, "change_rate") .filter("trade_time >= todayStart()") .build();

AggregateFunction.LAST表示取最新值，SUM表示累加。filter条件限定只计算当日数据。这两个指标在实时模式下每来一条数据就更新，在离线模式下对历史数据做全量重算，结果一致。

滑动窗口聚合稍微复杂一些。比如计算「近 5 分钟均价」，需要维护一个时间窗口内的数据集合。bboss 支持基于事件时间的滑动窗口：

// 滑动窗口指标：近 5 分钟均价 MetricDefinition ma5min = MetricDefinition.builder() .name("ma_5min") .groupBy("stock_code") .window(WindowType.SLIDING, Duration.ofMinutes(5), Duration.ofSeconds(30)) .aggregate(AggregateFunction.AVG, "price") .build();

WindowType.SLIDING表示滑动窗口，窗口大小 5 分钟，滑动步长 30 秒。这意味着每 30 秒重新计算一次过去 5 分钟的均价。实时模式下，窗口随数据到达滚动推进；离线模式下，窗口按事件时间对齐后批量计算。

这里有个血泪经验：滑动窗口的步长不要设得太小。步长 1 秒、窗口 5 分钟意味着要维护 300 个窗口状态，内存和计算开销都很大。实际业务里 30 秒到 1 分钟的步长通常够用。

离线重算的场景是：指标定义改了，或者发现历史数据有误，需要对过去某段时间的数据重新计算指标。bboss 的做法是提交一个离线重算任务，指定时间范围，任务会读取该范围内的原始数据，按当前指标定义重新计算，然后覆盖指标存储中的旧值。

// 离线重算任务：重算 2024-01-01 到 2024-01-31 的指标 RecalcTask recalcTask = RecalcTask.builder() .metricNames("latest_price", "daily_change_rate", "ma_5min") .startTime("2024-01-01T00:00:00") .endTime("2024-01-31T23:59:59") .parallelism(4) .build(); RecalcTaskManager.submit(recalcTask);

parallelism控制并行度，根据数据量和集群资源调整。重算任务执行期间，实时指标计算不受影响，两者写入不同的指标版本，重算完成后切换版本。

验证指标计算结果是否正确，我通常用「对账」的方式：从原始数据里用 SQL 手动算一遍，和 bboss 算出来的指标值对比。比如近 5 分钟均价，用 SQL 的AVG(price) OVER (PARTITION BY stock_code ORDER BY trade_time RANGE BETWEEN INTERVAL '5' MINUTE PRECEDING AND CURRENT ROW)算出来，和 bboss 的ma_5min指标逐条比对。差异超过阈值就排查窗口边界和数据乱序问题。

这套链路我从最早用脚本拼凑，到后来换成 bboss 统一管理，最大的体会是：采集和清洗的逻辑一定要收在一处，不要为了「实时快」就单独写一套。流批一体化的价值不在技术多先进，而在维护成本低。指标计算能预聚合就别即席查询，能离线重算就别手动修数据。希望帮到你。

本文还有配套的精品资源，点击获取