简介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 groupIdcom.bbossgroups/groupId artifactIdbboss-datasource/artifactId version5.9.5/version /dependency dependency groupIdcom.bbossgroups/groupId artifactIdbboss-datax/artifactId version5.9.5/version /dependency版本号需要根据实际仓库中可用的最新版本调整不要照抄。引入依赖后在resources目录下创建application.properties配置数据源连接信息。# 数据源配置MySQL 作为采集源 datasource.namemysql_source datasource.drivercom.mysql.cj.jdbc.Driver datasource.urljdbc:mysql://127.0.0.1:3306/stock_db?useUnicodetruecharacterEncodingutf8 datasource.usernameroot datasource.passwordyour_password # Elasticsearch 作为入库目标 elasticsearch.hosts127.0.0.1:9200 elasticsearch.clusterNamedocker-cluster这里datasource.name是自定义的数据源标识后续在采集任务中引用。Elasticsearch 的clusterName在 7.x 以上版本可以留空bboss 会自动协商。3.2 定义采集任务与清洗转换处理器采集任务的定义有两种方式XML 配置和 Java API。XML 方式适合标准化采集场景Java API 适合需要动态逻辑的场景。先看 XML 方式。!-- 采集任务定义从 MySQL 读取股票行情清洗后写入 ES -- properties property nametaskName valuestock_quote_etl/ property namedatasource valuemysql_source/ property namesql valueSELECT stock_code, current_price, change_percent, trade_time FROM stock_quote WHERE trade_time ?/ property namesqlParamType valuetimestamp/ property namesqlParamName valuelastMaxTradeTime/ property nametarget valueelasticsearch/ property nametargetIndex valuestock_quote_index/ property namebatchSize value500/ property namerefreshInterval value5000/ /properties这段配置的含义从mysql_source数据源执行 SQL查询trade_time大于上次最大时间的记录每批 500 条每 5 秒刷新一次。lastMaxTradeTime是 bboss 内置的增量采集参数自动记录上次采集的最大时间戳避免重复采集。清洗转换处理器用 Java 实现// 自定义转换处理器字段重命名 时间格式转换 空值过滤 public class StockQuoteTransform implements DataTransform { Override public Object transform(Object data) { MapString, Object record (MapString, 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-ddTHH: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.size1000 elasticsearch.bulk.flushInterval3000 elasticsearch.bulk.concurrent4 elasticsearch.bulk.retry3 elasticsearch.bulk.retryInterval1000bulk.size是单批提交的文档数flushInterval是强制刷新间隔毫秒即使没攒够bulk.size也会在超时后提交。concurrent是并发写入线程数根据 ES 集群的写入能力调整。retry和retryInterval控制失败重试策略。数据库入库的配置类似但参数名不同# 数据库批量写入参数 database.batch.size500 database.batch.flushInterval5000 database.batch.transactiontruetransactiontrue表示批量写入在一个事务中提交失败时整批回滚。如果业务允许部分成功设为 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?size1 | 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 nametargetIdField valuestock_code,trade_time/4.2 批量写入缓冲区溢出导致 OOM现象任务运行一段时间后 JVM 抛OutOfMemoryError堆内存持续增长。原因bulk.size设得太大比如 10000或者flushInterval设得太长数据在缓冲区里堆积。如果上游采集速度大于下游写入速度缓冲区会无限增长。解决把bulk.size降到 5001000flushInterval设为 20005000 毫秒。同时给缓冲区设置上限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-ddTHH:mm:ss.SSSZ} } } }4.5 流式模式下任务假死但无报错现象任务状态显示运行中但目标端长时间没有新数据写入。原因流式采集依赖数据源的事件通知机制。如果数据源是数据库bboss 通过轮询方式检测新数据轮询间隔设得太长比如 60 秒会让人误以为任务挂了。如果数据源是消息队列可能是消费者组 offset 提交失败导致重复消费或停止消费。解决先确认采集端的轮询间隔配置数据库采集的refreshInterval建议设为 30005000 毫秒。消息队列采集检查 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 统一管理最大的体会是采集和清洗的逻辑一定要收在一处不要为了「实时快」就单独写一套。流批一体化的价值不在技术多先进而在维护成本低。指标计算能预聚合就别即席查询能离线重算就别手动修数据。希望帮到你。本文还有配套的精品资源点击获取