做日志采集和数据管道这几年Flume 一直是我手里最稳的组合件之一但真正迎来挑战的是某个需求要求把数据直接写入 ClickHouse。官方 Sink 只有 HDFS、Hive、Kafka 这些常规出口翻遍全网也很难找到一个维护活跃、支持批量、能扛住生产流量的 Flume ClickHouse Sink。没办法我只能自己动手从源码到上线折腾了一轮。这篇文章把整个开发过程、代码骨架、参数设计、事务边界和排查实录全部整理出来给同样被这个需求卡住的朋友做参考。1. 项目背景与方案选型1.1 数据管道里的最后一公里当时的需求并不复杂业务日志通过 Filebeat 采集到 Kafka下游有一个传统的 Flume 消费者集群负责把不同业务线的数据分发到各类存储。新的分析平台选定 ClickHouse 作为明细查询引擎于是需要把日志明细近实时写入 ClickHouse允许分钟级延迟但要求吞吐稳定、数据不丢。摆在面前的问题非常直接Flume 生态里没有官方 ClickHouse Sink。我们不可能为了让数据进 ClickHouse 就把整个 Flume 集群换成其他采集框架那样改造成本太大而且会破坏现有的 Channel、监控和运维体系。开发一个自定义 Sink 变成了最务实的路径。相比在 Kafka 下游再接一套独立的写入程序直接在 Flume 里做 Sink 有几个天然优势复用已有的 Channel 缓冲机制天然获得背压控制和故障恢复接入现有的 Flume 监控体系不用额外搭一套采集链路部署方式简单把 jar 丢到 Flume 插件目录改几行配置就能灰度。1.2 开源社区里为什么没有合适的现成方案不是说完全没有GitHub 上能找到一些 flume-clickhouse-sink 仓库但大多停留在 demo 阶段。有的只支持单条插入吞吐上不去有的硬编码了数据库和表名扩展性几乎是零还有的异常处理非常粗暴失败就直接抛异常导致 sink 线程退出连最基本的 rollback 都做不好。更麻烦的是版本兼容问题。ClickHouse 的 HTTP 接口这些年变化不小老代码里用的字段名、URL 参数、驱动坐标放到新版本上经常不兼容。与其花时间修别人的半成品不如自己写一个代码量也就几百行还能按我们的业务场景做深度定制。1.3 候选技术方案对比设计初期我列过三个可选路线这里放出来对比能让后续开发少走弯路。方案开发量数据一致性维护成本备注在 Flume 中开发自定义 Sink中走 Channel 事务at-least-once低可完全复用现有 Flume 体系Kafka Connect ClickHouse 连接器大依赖连接器自身实现高需要额外部署 Connect 集群单独写消费程序从 Kafka 写入 ClickHouse中需自己控制 offset高相当于重复造 Flume 的轮子最终选择第一条。理由很简单团队的 Flume 集群已经稳定运营很久Channel 缓冲、故障转移、负载均衡都有现成方案插入 ClickHouse 只是整个链路的最后一跳不应该为了这一跳引入新的技术组件。2. Sink 开发底层原理先搞懂 Flume 的事务模型2.1 Sink 生命周期四步走写自定义 Sink 前必须清楚 Flume 框架对 Sink 的调用约定。一个标准的 Sink 组件要经历四个阶段configure、start、process、stop。框架首先从配置文件读取 Sink 的全类名反射创建实例然后调用 configure(Context context) 方法。这个 Context 就是配置文件中agent.sinks.sinkName.之后所有键值对的集合简单说就是你的参数入口。之后框架调用 start()。这个方法只调用一次用来创建连接池、初始化状态、开启后台线程。configure 和 start 的区别要特别注意configure 在组件初始化和配置刷新时都可能被调用而 start 只执行一次所以不要混用。然后是最重要的 process() 方法。SinkRunner 线程会不断循环调用它整个 Flume 的数据消费循环全部围绕这个方法展开。停止时框架调用 stop() 释放资源。2.2 Channel 事务与 at-least-once 语义Flume 的 Channel 事务和数据库事务不是一回事。数据库事务保证的是数据的 ACID而 Flume 的 Channel 事务只保证操作的原子性一批 take 或 put 要么全部生效要么全部不生效。自定义 Sink 的标准操作模式如下Channel ch getChannel(); Transaction txn ch.getTransaction(); txn.begin(); try { // 从 Channel 中 take 事件 Event event ch.take(); // 发送到下游 // 只有发送成功才提交事务 txn.commit(); return Status.READY; } catch (Exception e) { txn.rollback(); throw new EventDeliveryException(e); } finally { txn.close(); }关键点在于发送动作发生在 Channel 事务之内。只有下游确认写入成功后事务才提交事件才会从 Channel 中真正移除。如果发送失败事务回滚这批事件会回到 Channel 头部下次 process 调用时重新取出来投递。这正是 Flume 数据管道至少一次语义的根源。它保证不丢数据但也不承诺不重复。理解这一点对后面设计去重策略非常重要。2.3 process() 的返回值与背压控制process() 有两个返回值看到很多新手写 sink 时忽略了它们的含义。Status.READY 表示这次调用已经成功消费了数据框架会立即再次调用 process() 形成持续消费循环。Status.BACKOFF 表示当前没有数据可消费或者下游暂时不可用希望框架稍作退避再调用。如果 Channel 中没有事件channel.take() 返回 null这时应该 rollback 并返回 BACKOFF。不少 demo 代码在数据为空时直接返回 READY结果就是 Flume 线程进入空转CPU 飙高。正确做法if (rows.isEmpty()) { txn.rollback(); return Status.BACKOFF; }BACKOFF 的退避时长由 SinkRunner 内部策略控制会从较短时间开始逐步加大到上限所以就算 ClickHouse 长时间不可用也不会把你的 CPU 打满。3. 自定义 ClickHouse Sink 的完整实现3.1 工程骨架与 Maven 依赖项目用标准 Maven 结构关键是依赖声明。Flume 提供的接口在主工程里编译时需要但运行时 Flume 集群已经有这些 jar所以 scope 用 provided。ClickHouse 驱动则要打包进最终交付的 jar。dependencies dependency groupIdorg.apache.flume/groupId artifactIdflume-ng-core/artifactId version1.9.0/version scopeprovided/scope /dependency dependency groupIdorg.apache.flume/groupId artifactIdflume-ng-sdk/artifactId version1.9.0/version scopeprovided/scope /dependency !-- HTTP 客户端用于访问 ClickHouse HTTP 接口 -- dependency groupIdorg.apache.httpcomponents/groupId artifactIdhttpclient/artifactId version4.5.13/version /dependency /dependencies我最终选择用 HTTP 接口直接写入 ClickHouse 而不是 JDBC 驱动。原因后面会展开先说结论HTTP 接口依赖更少出错信息更直观对 Flume Sink 这种攒批场景非常合适。3.2 配置参数解析与默认值设计Sink 的配置通过 configure(Context) 解析。为了灵活适配不同场景我把如下参数全部做成可配置而不是硬编码在代码里。public class ClickHouseSink extends AbstractSink implements Configurable { private String clickHouseUrl; private String database; private String tableName; private String user; private String password; private int batchSize 500; private long flushIntervalMillis 3000; private int connectTimeout 5000; private int socketTimeout 15000; Override public void configure(Context context) { clickHouseUrl context.getString(clickhouse.url); database context.getString(clickhouse.database, default); tableName context.getString(clickhouse.table); user context.getString(clickhouse.user, default); password context.getString(clickhouse.password, ); batchSize context.getInteger(clickhouse.batchSize, 500); flushIntervalMillis context.getLong(clickhouse.flushIntervalMillis, 3000); connectTimeout context.getInteger(clickhouse.connectTimeout, 5000); socketTimeout context.getInteger(clickhouse.socketTimeout, 15000); if (clickHouseUrl null || tableName null) { throw new IllegalArgumentException(clickhouse.url and clickhouse.table must be configured); } } }一个很小的细节batchSize 设为 500 是实测下来的经验值。太小每个批次都要发起一次 HTTP 请求吞吐拉不上去太大单批次数据量过大会导致 ClickHouse 内存压力上升并且一旦失败回滚的数据量也大恢复时间更长。生产环境里 500 到 1000 是比较稳妥的范围。3.3 process() 核心逻辑与线程模型process() 是 Sink 的心脏。我采用每个调用批次累积到 batchSize 就发送一次的方式理论上不需要额外 sleep因为 SinkRunner 在返回 READY 后会立刻调用下一次 process()。Override public Status process() throws EventDeliveryException { Channel ch getChannel(); Transaction txn ch.getTransaction(); ListEvent events new ArrayList(batchSize); txn.begin(); try { for (int i 0; i batchSize; i) { Event event ch.take(); if (event null) { break; } events.add(event); } if (events.isEmpty()) { txn.rollback(); return Status.BACKOFF; } int sentCount insertToClickHouse(events); if (sentCount events.size()) { throw new IOException(partial insert failed: sentCount / events.size() accepted); } txn.commit(); return Status.READY; } catch (Exception e) { txn.rollback(); LOG.error(ClickHouseSink process failed, events will be re-delivered, e); return Status.BACKOFF; } finally { txn.close(); } }这里有一个非常重要的语义需要强调只要 insertToClickHouse 抛异常整个事务回滚全部事件回到 Channel 重新投递。这样设计保证了不丢数据但代价是可能重复。ClickHouse 侧如果没有去重机制会出现重复明细。关于怎么在 ClickHouse 侧做幂等第 4 章专门讲。我遇到过团队里的人把部分失败理解成把成功的那几行留下失败的重新投递然后试图在 Sink 里做部分提交。这个思路非常危险。一旦部分提交Channel 里的数据和你实际写入 ClickHouse 的数据就无法对齐后续恢复、监控、幂等全部失控。保持全有或全无非线性的简单逻辑这是 Sink 设计的底线。3.4 通过 HTTP 接口写入 ClickHouseClickHouse 原生支持 HTTP 接口默认端口 8123。写入方式是在 URL 的 query 参数里带上 INSERT 语句POST body 里放数据。对于 JSONEachRow 格式body 就是一行一个 JSON 对象。private int insertToClickHouse(ListEvent events) throws IOException { StringBuilder sb new StringBuilder(); for (Event event : events) { String body new String(event.getBody(), StandardCharsets.UTF_8); if (body.trim().isEmpty()) { continue; } sb.append(body).append(\n); } if (sb.length() 0) { return events.size(); } String insertQuery String.format( INSERT INTO %s.%s FORMAT JSONEachRow, database, tableName ); String url String.format( %s/?query%sinput_format_skip_unknown_fields1input_format_defaults_for_omitted_fields1, clickHouseUrl, URLEncoder.encode(insertQuery, StandardCharsets.UTF_8) ); HttpPost post new HttpPost(url); post.setEntity(new StringEntity(sb.toString(), ContentType.APPLICATION_JSON)); String auth user : password; String encodedAuth Base64.getEncoder().encodeToString(auth.getBytes(StandardCharsets.UTF_8)); post.setHeader(Authorization, Basic encodedAuth); try (CloseableHttpResponse response httpClient.execute(post)) { int code response.getStatusLine().getStatusCode(); if (code ! 200) { String errBody EntityUtils.toString(response.getEntity()); throw new IOException(ClickHouse write failed, code code , response errBody); } } return events.size(); }几个踩过的细节第一query 参数必须做 URL 编码。Insert 语句里有空格、点号直接拼到 URL 里的话ClickHouse 可能返回 400。我用 URLEncoder.encode 把整个 SQL 编码后拼进去。第二POST body 只放数据不要放 SQL。有些人习惯把整个 INSERT 语句放 body那会触发 ClickHouse 的多语句解析逻辑容易出幺蛾子。规范用法是 SQL 在 URL query 里数据在 body 里。第三input_format_skip_unknown_fields1这个参数很关键。业务日志字段变化频繁偶尔会多出一个新字段如果没有这个参数ClickHouse 对 JSONEachRow 里目标表不存在的字段直接报错整个批次失败。加上这个参数后未知字段会被跳过大大增强容错性。3.5 连接池与性能优化每个 process() 都 new 一个 HttpClient 是典型的性能杀手。我使用连接池管理 HTTP 连接Sink 启动时初始化一次。private HttpClient httpClient; Override public void start() { PoolingHttpClientConnectionManager cm new PoolingHttpClientConnectionManager(); cm.setMaxTotal(4); cm.setDefaultMaxPerRoute(2); httpClient HttpClients.custom() .setConnectionManager(cm) .setDefaultRequestConfig(RequestConfig.custom() .setConnectTimeout(connectTimeout) .setSocketTimeout(socketTimeout) .build()) .build(); }socketTimeout 是重点。ClickHouse 写入过程中如果恰好碰上 merge 或磁盘抖动响应可能很慢。不设置 socketTimeout请求会一直挂着process() 一直阻塞Channel 数据不断累积最后整个 Flume agent 卡死。设置合理的超时让失败快速暴露反而比慢吞吞的成功更安全。4. 与 ClickHouse 交互容易被坑的细节4.1 三种数据格式选型对比ClickHouse HTTP 接口支持很多数据格式对 Flume Sink 场景我认真比较了三种。格式优点缺点适合场景JSONEachRow字段名自动映射容错性好体积略大解析略慢字段经常变化的日志数据TabSeparated体积最小写入最快字段顺序必须严格匹配表结构固定的批数据CSV通用性好转义麻烦类型容易混淆对接外部系统最终选 JSONEachRow。一方面我们的事件在 Kafka 里本来就是 JSON 格式直接用 JSONEachRow 写入中间不需要做字段顺序编排另一方面业务字段持续新增JSONEachRow 天然支持字段名映射配合 3.4 节里的 skip_unknown_fields 参数可以保证新字段出现时不至于让整个管道挂掉。4.2 数据重复与幂等设计前面说过Flume 的事务语义是 at-least-once重复是常态而不是意外。网络超时是最典型的场景ClickHouse 实际写入成功了但 Sink 没有收到 200 响应超时抛异常事务回滚下个批次又投递一遍同样的事件。ClickHouse 本身不提供行级唯一键去重。如果业务对重复数据敏感需要在表设计上做一些功课。最实用的方案是使用 ReplacingMergeTree配合一个版本字段。CREATE TABLE events ( event_id String, ts DateTime, biz_tag String, payload String, _version UInt64 ) ENGINE ReplacingMergeTree(_version) PARTITION BY toYYYYMM(ts) ORDER BY event_id;写入的时候_version 用当前时间戳比如INSERT INTO events FORMAT JSONEachRow {event_id:abc-123,ts:2025-06-11 10:00:00,biz_tag:order,payload:{...},_version:1780000000}ReplacingMergeTree 在后台合并时对相同 ORDER BY 字段的数据保留 _version 最大的一条。这样即使 Flume 重复投递了同一条事件最终查询时也只有一条。注意去重依赖后台 merge不是实时的但通常分钟级内就能完成。对分析场景来说这个延迟完全可以接受。4.3 时区、NULL 与字段类型陷阱ClickHouse 的 DateTime 字段会按服务器时区解析字符串。如果你的日志时间戳是 UTC而 ClickHouse 服务器是东八区写入的 ts 就会偏移。最稳妥的方案是统一在 Sink 侧处理从 event 的 header 或 body 中解析出时间字段后转成目标时区的字符串再发送。或者干脆全部用 DateTime64 存 UTC 时间查询时再转换显示。不管选哪种核心原则只有一个全链路时区必须对齐否则数据查出来是乱的。JSONEachRow 的 NULL 处理也有讲究。ClickHouse 里普通 String 字段不接受 NULLNullable(String) 才接受。如果日志数据里某个字段是null写入普通 String 字段会报错。我们的处理方式是在 Flume 上游处理日志格式时就把 null 统一改成空字符串避免在 Sink 里做复杂的字段预处理。整数类型的解析也值得注意。JSONEachRow 对数字字段要求比较严格如果表结构是 UInt32而 JSON 里给了一个字符串会导致解析失败。因此表结构和日志字段类型必须严格对应这种问题只有在压测才能暴露出来。5. 压测与线上问题排查实录5.1 性能瓶颈定位Sink 写完第一版后我用一个三节点 ClickHouse 集群和单节点 Flume agent 做压测。数据源是 Kafka 里固定速率的日志流每个 Event body 约 1KB。通过 Flume 自带的 HTTP 监控接口默认端口 41414可以实时看到 Channel 的当前容量、Sink 处理成功和失败的批次数。我重点观测两个指标Channel 剩余容量是否持续下降Sink 的 ProcessAll 成功事件数增长速度。实测数据如下batchSize单批耗时ms吞吐事件/秒CPU 占用1008680025%500221580022%1000351870021%2000681820023%可以看到batchSize 到 1000 之后吞吐不再线性增长说明瓶颈已经从 Sink 的攒批逻辑转移到了 HTTP 请求序列化、网络传输和 ClickHouse 解析速度。最终我把生产环境的 batchSize 定为 1000兼顾吞吐和单批失败的回滚成本。还有一个秘密如果单条日志很小比如几百字节可以把多个事件合并成一个 HTTP 请求体。ClickHouse 对 HTTP 单次请求体的上限比较宽容10MB 以内的请求都能稳定处理。5.2 五个高频故障与应对记录故障一HTTP 400响应信息是 empty query。原因是 URL 里的 SQL 没有做 URL 编码空格和点号被浏览器或 HttpClient 解析掉了。改为 URLEncoder.encode 后解决。故障二字段数量不一致导致 JSONEachRow 解析失败。日志偶尔会多出嵌套对象字段而表结构是扁平 String。解决办法是在 ClickHouse 表里把不确定的字段全部定义成 String提前将复杂对象 JSON 序列化后再发送。故障三ClickHouse 节点重启后Flume 的 HTTP 请求全部超时。排查发现 socketTimeout 默认是 0也就是无限等待。节点在恢复副本时负载很高请求迟迟不返回导致大量请求堆积。设置 socketTimeout15000 后失败的请求快速回滚等节点恢复后自动重新投递。故障四数据重复比预期严重。做了 ReplacingMergeTree 后重复率最终降了下来。但要注意ReplacingMergeTree 的合并是后台异步的刚写入后马上查询可能还会看到重复行。需要在业务侧接受这个最终一致性的时间窗口。故障五Flume agent 正常启动但 ClickHouse 里迟迟没有数据。检查发现是配置里的 tableName 带了引号被直接拼接进 SQLClickHouse 把引号当成了表名的一部分。配置参数越简单越好不要想着帮用户做转义保持原样拼接然后快速报错反而更好排查。5.3 开发过程中积累的核心经验第一配置参数的命名要统一前缀如clickhouse.url、clickhouse.table这样既避免和 Flume 内置配置冲突也方便后续做多实例部署时一个 agent 内配置多个 ClickHouse Sink。第二务必要为每个批次记录日志。我在每一批 commit 后打一行batch sent: count500, cost23ms压测和线上排障时这些日志是最直接的数据。第三回滚是安全的但不要过度依赖回滚。如果 ClickHouse 持续不可用Channel 会快速积压最终触发 Channel 容量上限上游开始背压这其实是正常的保护机制。不要为了提高一次性成功率去扩大 Channel 容量那样只会延迟问题的爆发。第四如果将来要支持动态分表比如按天自动切换表名可以在 process() 里根据当前时间拼出 table 名。但要注意在一批数据里不要混入不同分表的数据极端情况会引发意想不到的问题。我采用的做法是在 URL 里使用模板参数当 channel 取到的第一条事件时间戳确定后生成固定的表名整批次都写入这张表避免跨表写同一批数据。6. 回望与可复用建议这次自定义 Sink 的开发周期并不长从动手写第一个类到压测通过大约用了三天。真正的难点不在写代码而在理解 Flume 的语义、ClickHouse 的行为和两者之间的边界。搞懂了事务边界Sink 的骨架就清晰了搞懂了 ClickHouse HTTP 接口的黑话写入逻辑其实只是一个封装 HTTP 请求的过程。如果你也要做类似的事情建议从一开始就把源码和配置分离Sink 里只做通用逻辑表名、字段映射、时区这些都是配置项不要为了图省事硬编码。后续业务扩展、表结构调整的时候你会感谢当初这个决定。还有一个我自己很受益的小习惯为这个 Sink 单独写一个单元测试用一个本地启动的 ClickHouse Docker 实例模拟真实环境。每次改完代码先跑测试再打包发版比依赖集群环境做集成测试高效太多。这套方案现在已经在生产环境稳定运行了大半年Day-1 上线后的故障率远低于预期。如果哪一天 ClickHouse 官方或社区出了维护良好的 Flume Sink我们当然可以考虑迁移但在那之前自己写的这个 Sink 依然是整个管道里最让我放心的一环。