Flume自定义Sink写入ClickHouse：从原理到生产实践

做日志采集和数据管道这几年，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 集群已经稳定运营很久，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> <groupId>org.apache.flume</groupId> <artifactId>flume-ng-core</artifactId> <version>1.9.0</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.flume</groupId> <artifactId>flume-ng-sdk</artifactId> <version>1.9.0</version> <scope>provided</scope> </dependency> <!-- HTTP 客户端，用于访问 ClickHouse HTTP 接口 --> <dependency> <groupId>org.apache.httpcomponents</groupId> <artifactId>httpclient</artifactId> <version>4.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(); List<Event> 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 接口写入 ClickHouse

ClickHouse 原生支持 HTTP 接口，默认端口 8123。写入方式是在 URL 的 query 参数里带上 INSERT 语句，POST body 里放数据。对于 JSONEachRow 格式，body 就是一行一个 JSON 对象。

private int insertToClickHouse(List<Event> 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=%s&input_format_skip_unknown_fields=1&input_format_defaults_for_omitted_fields=1", 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_fields=1这个参数很关键。业务日志字段变化频繁，偶尔会多出一个新字段，如果没有这个参数，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。一方面，我们的事件在 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 字段不接受 NULL，Nullable(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 到 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，也就是无限等待。节点在恢复副本时负载很高，请求迟迟不返回，导致大量请求堆积。设置 socketTimeout=15000 后，失败的请求快速回滚，等节点恢复后自动重新投递。

故障四：数据重复比预期严重。做了 ReplacingMergeTree 后，重复率最终降了下来。但要注意，ReplacingMergeTree 的合并是后台异步的，刚写入后马上查询可能还会看到重复行。需要在业务侧接受这个最终一致性的时间窗口。

故障五：Flume agent 正常启动，但 ClickHouse 里迟迟没有数据。检查发现是配置里的 tableName 带了引号，被直接拼接进 SQL，ClickHouse 把引号当成了表名的一部分。配置参数越简单越好，不要想着帮用户做转义，保持原样拼接然后快速报错反而更好排查。

5.3 开发过程中积累的核心经验

第一，配置参数的命名要统一前缀，如clickhouse.url、clickhouse.table，这样既避免和 Flume 内置配置冲突，也方便后续做多实例部署时一个 agent 内配置多个 ClickHouse Sink。

第二，务必要为每个批次记录日志。我在每一批 commit 后打一行batch sent: count=500, cost=23ms，压测和线上排障时，这些日志是最直接的数据。

第三，回滚是安全的，但不要过度依赖回滚。如果 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 依然是整个管道里最让我放心的一环。