不知道你有没有遇到过这种场景日志数据已经在 Kafka 里躺好了下游分析团队那边的业务人员天天催说想直接用 SQL 做多维统计最好能秒级出数。你评估了一圈发现 ClickHouse 是最合适的存储引擎但数据链路在 Flume 这一环卡住了——Flume 自带的 Sink 无非是 HDFS、Kafka、Logger 这几类根本没有现成的 ClickHouse Sink 可以用。是的你没看错官方仓库里至今没有提供 ClickHouse 的 Sink 实现网上能搜到的第三方方案要么版本太老要么依赖了一批你根本不想引进来的包。我当时的做法和很多人一样自己写一个。这篇内容就是完整记录我从零开发 Flume 自定义 Sink 并写入 ClickHouse 的整个过程包含接口设计、生命周期处理、批量写入策略、参数调优以及几个让我花了一整天才解决的坑。对准备做 Flume 二次开发、或者正在为 ClickHouse 寻找稳定写入通道的开发者来说这份经验可以直接抄作业。1. 确认需求边界什么时候该自研 Flume Sink什么时候不该1.1 先说 Flume 的 Sink 生态现状Flume 的定位是日志收集和搬运它的核心抽象就是 Source、Channel、Sink 三段式管道。Source 负责对接数据来源Channel 作为中间缓冲Sink 负责把数据写到目的地。官方自带的 Sink 覆盖面其实很广HDFS Sink、Hive Sink、Kafka Sink、HTTP Sink、File Roll Sink 等等功能上够用但就是没有 ClickHouse Sink。HBase Sink 倒是原生支持但是 ClickHouse 这种列式分析库的写入协议和 HBase 完全不一样不能拿过来硬改。我记得当时项目里有两个选择一个是在 Flume 的 Sink 端接 Kafka再由独立消费者程序把 Kafka 数据转写到 ClickHouse另一个就是直接在 Flume 里写一个自定义 Sink让数据从 Kafka Source 到 Channel 之后直接落入 ClickHouse。第一种方案多了一跳链路MOM 消息中间件上多了一个消费者组运维复杂度上升而且 Kafka 落地的时效性完全取决于那个消费者程序的调度频率。第二种方案链路最短从 Kafka 消费到 ClickHouse 写入中间只有一层 Channel 缓冲延迟可以做到秒级。最终我选了第二种。1.2 自研前必须想清楚的四件事不要一上来就写代码先花半天把需求边界画清楚。我当时给自己列了四个问题第一数据量级是多少。这决定了 Sink 是逐条写还是批量写。ClickHouse 对单条插入的兼容性很差虽然它支持 VALUES 语法但逐条插入的吞吐量低到没法看。如果你每秒只有几十条数据那确实可以偷懒逐条写但只要有每秒几百条以上就必须走批量。第二数据格式是什么。Flume 的 Event 结构是 headers 加 bodybody 是字节数组。你的数据是 JSON、CSV、纯文本还是自带分隔符ClickHouse 接收数据的方式很多样可以走 JDBC 的 PreparedStatement 占位符也可以用 CSV 格式直接灌甚至是 RowBinary 二进制协议。格式决定了解析逻辑的复杂度。第三容错要求。ClickHouse 本身不支持跨机器的事务Flume Sink 的 process 方法返回 BACKOFF 后Channel 中的数据会回滚或者保留这里的重复消费风险你要提前想清楚。ClickHouse 不是 Kafka没有完善的幂等机制重复写一遍数据就是实打实的重复数据。第四最大可接受延迟。ClickHouse 建议每次积攒一批数据再写入批次越大性能越好但延迟会升高。如果业务要求秒级可见那批大小就不能设置得太夸张。这四个问题其实是在帮你提前设计方案哪怕最后不写代码只是想在 Flume 生态里找一个写 ClickHouse 的现成插件这四件事也是你评估插件是否能用的标准。2. 掌握 Flume Sink 的开发骨架接口、生命周期与返回语义2.1 Sink 的核心接口与三类返回状态Flume 的 Sink 接口定义非常精简核心方法就几个public interface Sink extends LifecycleAware, NamedComponent { Status process() throws EventDeliveryException; }LifecycleAware 规定了 start() 和 stop() 两个生命周期方法NamedComponent 加了一个 setName()。真正让你每次写逻辑的地方就是 process()而这个方法的返回值只有两个Status.READY 和 Status.BACKOFF。初次开发的人很容易忽略这两个返回值的语义。READY 表示当前 Sink 成功处理了一批事件可以继续立即调用 process() 来处理下一批BACKOFF 表示当前没有数据可处理或者暂时无法处理数据Flume 的 SinkRunner 会指数退避一段时间再重新调用 process()避免空转打满 CPU。我之前见过有人在 Sink 内部自己写 while 循环去拉数据这是对 Flume 模型的误解。正确姿势是process() 每次都只处理一批处理完返回 READY没数据返回 BACKOFF剩下的交给 SinkRunner 调度。2.2 生命周期方法configure、start、process、stop 怎么配合Flume 自定义 Sink 的完整生命周期有四个阶段configure(Context context) 在 Sink 构建时调用一次负责解析 flume.conf 里配置的 Agent 参数。这里有一个容易被坑的点configure 只调用一次不是每次 process 都调用。如果你想做动态刷新配置需要在 start() 里启动独立线程监控配置不要想着在 configure 里做周期任务。start() 在 Sink 启动后调用适合初始化连接、创建线程池、启动后台线程。ClickHouse 的 JDBC 连接和缓冲队列都可以放在这里初始化和预热。process() 是核心方法每次由 SinkRunner 调用在里面实现批次读取和写入。stop() 在 Agent 停止时调用负责优雅关闭连接、清空缓冲队列、把还没写完的数据尽力 flush然后释放资源。这四步的顺序 Flume 会管理好你不需要关心调用方但要在实现 Sink 时严格遵守不要在 process() 里重置连接或重复加载配置这部分开销很浪费也不要跳过 stop() 里的资源释放尤其是你用了线程池和连接池时不释放会导致 Agent 停止后仍然有后台线程存活JVM 进程无法退出。2.3 一个可以复用的 BaseSink 模板我写了很多次自定义 Flume Sink后来干脆沉淀了一个 BaseSink 抽象类把生命周期和公共参数都抽出来。下面是一个极简模板public abstract class BaseClickHouseSink extends AbstractSink implements Configurable { protected String host; protected int port; protected String database; protected String table; protected String user; protected String password; protected int batchSize; protected long maxBackoffSleep; Override public void configure(Context context) { host context.getString(host, localhost); port context.getInteger(port, 8123); database context.getString(database, default); table context.getString(table, default_table); user context.getString(username, ); password context.getString(password, ); batchSize context.getInteger(batchSize, 5000); maxBackoffSleep context.getLong(maxBackoffSleep, 1000L); } Override public synchronized void start() { super.start(); // 初始化连接和缓冲子类实现 } Override public synchronized void stop() { // 关闭连接和缓冲子类实现 super.stop(); } Override public Status process() throws EventDeliveryException { // 子类实现核心写入逻辑 return Status.READY; } }AbstractSink 是 Flume 提供的基础抽象类已经实现了 setName 和 getName也会帮你维护一个 Channel 引用通过 getChannel() 获取。任何自定义 Sink 都应该继承它而不必直接实现最原始的 Sink 接口。3. 配置解析与 ClickHouse 客户端选型先解决“参数从哪来”和“怎么连”3.1 Context 解析从 flume.conf 到 Java 对象Flume 的 Agent 配置是一个 properties 风格文件开发者通过 Sink 的 configure(Context context) 拿到参数。参数名在 flume.conf 里怎么命名代码里就怎么取。比如 flume.conf 这样写agent1.sinks.chSink.type com.examples.clickhouse.ClickHouseSink agent1.sinks.chSink.host clickhouse-server agent1.sinks.chSink.port 8123 agent1.sinks.chSink.database analytics agent1.sinks.chSink.table event_log agent1.sinks.chSink.username default agent1.sinks.chSink.password agent1.sinks.chSink.batchSize 5000 agent1.sinks.chSink.maxBackoffSleep 1000注意 type 一定要写成你自定义 Sink 类的全限定类名否则 Flume 的组件工厂找不到对应的 Class。代码里的 context.getString(host, localhost) 就是读取配置。这里要特别提醒ClickHouse 有两个常用端口8123 是 HTTP 端口9000 是原生 TCP 端口。如果你想用 JDBC 连需要确认客户端驱动用的是哪个协议。我在第一次开发时犯过一个低级错误配置里写了 9000 端口但 JDBC 驱动走的是 HTTP 协议结果连接超时。后来换了 URL 模板才搞定。3.2 ClickHouse 连接选型JDBC 方案与行二进制方案怎么选ClickHouse 的客户端接入方式有几种最常用的是 clickhouse-jdbc 和 clickhouse-client。clickhouse-jdbc 是官方提供的 JDBC 驱动支持标准的 JDBC APIPreparedStatement、ResultSet 这些都支持。缺点是比较重且对于高吞吐大规模写入来说性能弱于原生客户端但胜在简单、好调试、生态兼容好。如果你团队里都是写 Java 的用 JDBC 最省事。clickhouse-client 是官方更新的客户端库性能更优支持 RowBinary 格式写入但 API 设计和 JDBC 差异很大学习成本高一些。RowBinary 是 ClickHouse 的二进制行格式写入效率很优秀适合日志类大批量数据。我当时选的是 clickhouse-jdbc。原因很直接项目里已经有大量 JDBC 代码团队成员对它熟能少踩一个坑。而且后面我把同一个 Sink 扩展去写 PostgreSQL 时代码几乎可以复用。Java 侧引入的 Maven 依赖大致如下dependency groupIdcom.clickhouse/groupId artifactIdclickhouse-jdbc/artifactId version0.4.6/version /dependency这个版本主要看你的 ClickHouse 服务端版本新版本的服务端搭配新版本驱动老版本用老驱动更稳。升级驱动前记得先看服务端兼容性说明。3.3 连接池与 HTTP 接口的取舍clickhouse-jdbc 底层的连接默认是 HTTP 方式走 8123 端口它本质上是 JDBC over HTTP每一次请求都要在 HTTP 协议上包装。如果你每次 process 都新建连接网络握手开销会很高还会引起大量 TIME_WAIT 连接。所以我加了一个简单的连接池而不是每次新建。但是连接池不要自己造轮子直接用现成的连接池框架比如 HikariCP 或者 Apache Commons DBCP。HikariCP 对 ClickHouse-JDBC 的兼容性是不错的配置也比较轻HikariConfig hikariConfig new HikariConfig(); hikariConfig.setJdbcUrl(jdbc:clickhouse:// host : port / database); hikariConfig.setUsername(user); hikariConfig.setPassword(password); hikariConfig.setMaximumPoolSize(3); hikariConfig.setMinimumIdle(1); hikariConfig.setConnectionTimeout(10000); dataSource new HikariDataSource(hikariConfig);这里有个细节ClickHouse 的 JDBC URL 里一般不需要加额外参数但如果你的服务端启用了重定向或者你希望启用 HTTP 连接复用的 keep-alive可以去看驱动的连接参数文档。HikariCP 的 maximumPoolSize 不用设太大因为 ClickHouse 写入瓶颈通常不在连接数而是批次大小。4. 写入策略的核心设计批量、缓冲与表分区怎么配合4.1 为什么逐条插入一定不行这是新手最容易犯的错误把从 Channel 取到的 Event 一条一条循环 insert。在 ClickHouse 里每次 INSERT 都是一次完整的数据块写入服务端需要做校验、压缩、写入分区。单条插入意味着每个块只有一行数据压缩比低分区写入次数爆炸性能会非常难看。我们来算一笔账假设单条数据 1KB一条一条插入ClickHouse 处理一百个分区文件碎片可能只要几秒钟但如果每秒来几百条服务端的 merge 压力会持续累积很多 MergeTree 表的 merge 任务会堆积查询性能被拖垮。所以正确做法是攒批。所谓攒批就是把一批数据放进同一条 INSERT 语句里或者用 PreparedStatement.addBatch() executeBatch()。用 JDBC 的方式核心代码如下PreparedStatement preparedStatement connection.prepareStatement( INSERT INTO database . table (event_time, event_type, content) VALUES (?, ?, ?) ); for (Event event : batchEvents) { String[] fields parseEvent(event); preparedStatement.setString(1, fields[0]); preparedStatement.setString(2, fields[1]); preparedStatement.setString(3, fields[2]); preparedStatement.addBatch(); } preparedStatement.executeBatch();executeBatch() 会把一批数据作为一个整体发送给服务端。批的大小取决于你的 batchSize 参数我建议从 2000 到 10000 之间调整先站在 5000 起步跑一段时间看服务端负载。4.2 用缓冲队列把 process() 调用与批量提交解耦Flume 的 process() 是同步方法它每被调用一次逻辑上就要处理一批数据。如果直接在里面从 Channel 拿事件再写 ClickHouse当 ClickHouse 响应变慢时Channel 的事务会长时间挂着进而影响 Source 端的消费效率。更好的设计是引入一个内存缓冲队列。process() 只负责把事件从 Channel 取出并放入队列然后立即返回 READYSink 内部一个独立的写入线程从队列里批量取数据写 ClickHouse。这样 Channel 不会被 Sink 拖垮而且缓冲队列天然把攒批的活干了。队列我推荐用 ArrayBlockingQueue有界且线程安全大小可以根据最大堆内存估算。假设 batchSize 是 5000每条事件平均 1KB那么队列容量设到 50000 也只是消耗约 50MB 内存堆足够的情况下不影响。写入线程用简单 while 循环ExecutorService executorService Executors.newSingleThreadExecutor(); executorService.submit(() - { while (!stopFlag.get()) { try { ListEvent batch new ArrayList(batchSize); queue.drainTo(batch, batchSize); if (batch.isEmpty()) { Thread.sleep(100); continue; } writeBatch(batch); } catch (Exception e) { // 记录错误继续循环 } } });好处是Flume 的 process() 调用频率可以非常高只要队列没满它处理得就很快不会因为一次 ClickHouse 慢查询影响整个数据链路。代价是内存消耗增加而且停机时需要在 stop() 里做一次队列 flush把缓存的数据尽量写出去。4.3 时间分区写入按事件时间决定目标表ClickHouse 高效查询的根源在于分片和分区裁剪。如果所有数据都写进一张大表分区字段选业务时间那理论上 ClickHouse 能自动按分区裁剪。但很多团队的实际做法是按天建表比如 event_log_20250207、event_log_20250208分区物理隔离更彻底也方便清理过期数据。在这种非动态表名模式下Sink 要根据事件的时间戳去拼目标表名。事件体里如果没有时间戳就用系统当前时间。目标表名解析放在写入线程里每发现日期切换就重建 PreparedStatementString targetTable event_log_ daySuffix; if (!targetTable.equals(currentTable)) { if (preparedStatement ! null) preparedStatement.close(); preparedStatement connection.prepareStatement( INSERT INTO database . targetTable (...) ); currentTable targetTable; }这里有一个必须写的逻辑跨天切换瞬间如果队列里还残留前一天的数据你需要根据事件自身时间而不是当前系统时间来判断它属于哪一天。也就是说攒批时数据可能跨两个日期最好在写入前按日期分组否则你拼错了表名数据就落在错误的日期分区后面排查起来极其痛苦。5. 稳定性与所谓“事务”重试、幂等与 ClickHouse 的天然限制5.1 process() 返回状态与重试的关系Flume 的多级重试机制建立在 Channel 事务上。Sink 的 process() 里你会拿到一个 Channel 事务Channel channel getChannel(); Transaction transaction channel.getTransaction(); try { transaction.begin(); // 从 channel 取出 Event写入缓冲队列 transaction.commit(); return Status.READY; } catch (Exception e) { transaction.rollback(); return Status.BACKOFF; } finally { transaction.close(); }这里的关键是如果你在 process() 内部同步写 ClickHouse写失败了 rollback事件会回到 Channel 等下一轮重试但如果你把事件放进了内存队列commit 之后就返回 READY理论上 Channel 认为事件已经成功分发。真正的 ClickHouse 写入在异步线程里失败后队列中事件确实丢了幂等性保护就丢失了。我前几个版本用的是同步写性能一般但语义清晰。后来切换为异步队列后我额外加了一层持久化兜底写入失败的事件事务回滚到重试队列重试达到最大次数就抛给单独的死信表。如果你的场景要求数据一条都不能丢异步方案务必要考虑这层兜底否则所谓“不丢数据”就是自欺欺人。5.2 ClickHouse 幂等的困境与 ReplacingMergeTree 兜底ClickHouse 不是天然事务型系统对同一条数据的重复 INSERT 不会像 MySQL 唯一键冲突那样报错它会静默接受两条完全相同的数据。所以 Sink 层如果要保证严格的恰好一次语义只能在写入前做去重或者在表引擎上做兜底。如果业务允许轻微延迟可以建 ReplacingMergeTree 表用业务主键字段做去重标识。ClickHouse 会在后台分区 merge 时根据排序键去重选择保留最新版本的数据。这个引擎已经能解决大部分重复场景。但如果业务需要严格去重且不能等后台 merge那至少要在 Sink 里维护最近写入的批次 ID 集合对重复 Event 做过滤。我见过有人引入 Redis 存最近一小时的事件指纹做去重效果不错但架构上又重了。对大多数日志分析场景而言ReplacingMergeTree 允许低频重复已经足够。5.3 可观测性把自定义指标暴露给 Flume 监控Flume 本身有基于 Dropwizard Metrics 的监控体系但自定义 Sink 默认不会被纳入。你不加监控指标生产环境里出了故障很难分清是 Flume 的问题还是 ClickHouse 的问题。我建议在 Sink 里加几个简单计数器通过 context 里的 metrics 上报private final AtomicLong processedCount new AtomicLong(0); private final AtomicLong failedCount new AtomicLong(0); private final AtomicLong lastBatchWaitTime new AtomicLong(0);然后在每次写入结束后更新计数。配合 Grafana 去观察 Flume JVM 的 GC、堆内存、Channel 容量和 ClickHouse 的 insert 吞吐整个链路状态一目了然。不要小看这一步批大小调优和故障定位全靠这些指标。6. 实测调参与避坑记录从 Kafka 到 ClickHouse 的完整链路6.1 参数推荐与效果参考在某个日志项目里我最终跑通的配置参考如下。这些数值以中等规格集群为背景具体还要结合你的数据量进行微调。参数推荐值说明batchSize5000每批事件条数过低吞吐上不去过高延迟升高queueCapacity50000缓冲队列容量受堆内存约束maxBackoffSleep1000 msprocess 无数据时的退避时间写入线程数1ClickHouse 写入瓶颈在网络和 Merge单线程足够HikariCP maximumPoolSize3连接池上限避免过多 HTTP 连接内存队列 flush 时间100 ms空队列轮询间隔实测效果5000 条一批、每条约 800 字节时单 Agent 写入 ClickHouse 的吞吐在每秒 3-6 万条左右延迟约 1-2 秒服务端 CPU 占用平稳。如果你需要更高吞吐优先加大 batchSize而不是增加 Sink 线程数。6.2 三个让我印象深刻的坑第一个坑是 ClickHouse JDBC 驱动的类加载冲突。项目里既有 Hive 相关依赖又引了 clickhouse-jdbc结果启动时出现 NoClassDefFoundError。原因是 Hive 的某些库把旧版本 httpclient 打进去了和 ClickHouse 驱动依赖的 httpclient 版本不一致。最后靠 Maven 排除冲突依赖解决。遇到 Flume Agent 无法启动先检查依赖树。第二个坑是动态分区拼表名时的 SQL 注入风险。表名和数据库名不应该依赖外部输入但如果你确实需要从事件头里抽取库名或表名必须做严格的白名单校验只允许字母、数字、下划线。否则恶意事件头构造的表名可能导致服务端 SQL 异常。第三个坑是 batchSize 设置过大导致内存溢出。我一开始因为追求高吞吐把 batchSize 设到 50000结果事件积压在队列里时堆内存直接 OOM。后来把队列容量和批次条数一起调小再配合 JVM heap 加大问题消失。记住一个原则队列容量 × 平均事件大小 × 2 必须小于堆内存空余量否则迟早爆掉。6.3 性能收益与后续扩展从 Kafka 到 ClickHouse 的整条链路里自定义 Sink 写数据在性能上已经不比 Flume 原生的 Kafka Sink 差。整个链路选型是Kafka Source 消费 - Memory Channel 缓冲 - ClickHouse Sink 批量写入。相比起中间接一个 Kafka 消费者程序这一套链路少了一跳REST 和运维都简单很多。后来我还给这个 Sink 扩展了几个功能支持多目标表轮询写、支持按事件类型选择表、支持写入前做简单的字段校验和清洗。整套代码沉淀下来后团队新同事在上手时基本只要看 BaseSink 的注释和 flume.conf 的配置示例就能快速接入新表。我从这个项目里得到的最大体会是别急着把需求做成“完美通用”的框架先把一条链路的延迟、吞吐和容错做到可控后面再考虑扩展。每个新业务表接入时你需要关心的就是建表语句和一行 Sink 配置剩下的交给已跑通的逻辑。最后分享一个非常实用的小技巧如果你在本地调试自定义 Sink可以把 flume.conf 里的 channel 配成 null channel 吗做不到Flume 没有 null channel。但你可以用 Memory Channel 配合一个简单的 tail Source 去模拟数据源在 debug 模式下打断点看 process() 里的事件流。这比把整个 Flume 打包丢到测试集群里反复重启要快得多。我第一次调试 ClickHouse Sink 时就是这么干的整个开发周期缩短了一半以上。