Flume自定义Sink对接ClickHouse：从零开发到生产实践

做日志采集的人，迟早会遇到这个问题：Flume的数据要进ClickHouse，但Flume自带的Sink里没有ClickHouse的位置。我接到这个需求的那天，第一反应也是先找现成的Sink，翻了一圈，要么是多年不维护的个人项目，要么只支持某一种特定Event格式，根本没法直接扔进生产环境。最后只能老老实实自己写一个。这篇文章就把我开发Flume自定义Sink并对接ClickHouse的全过程整理出来，包括为什么必须自己写、Flume的Sink扩展机制、完整代码骨架、生产参数调优和一堆实测踩坑记录。

1. 需求背景与方案选型：为什么非得自研这个Sink

1.1 原生生态缺的正是ClickHouse这一环

Flume自带的Sink其实不少：写文件的、写HDFS的、写Kafka的、写Hive的、写Elasticsearch的、写HTTP的，基本覆盖了主流数据出口。但ClickHouse是个例外。

ClickHouse是列式存储，写入模型是“大批量、少次数”，这和Flume默认的“流式小批量”模型天然有张力。即使你靠HTTP Sink硬怼，也要自己处理格式化、认证、失败重试、批量攒数据这些逻辑，而这些逻辑放到自定义Sink里才是正道。Flume的自定义Sink开发本质上就是让你接管Channel里的Event，用你自己的方式把数据送出去。

我当时面临的数据链路是这样的：业务服务器上有一堆应用日志，需要实时采集后进ClickHouse，供指标分析平台查询。Kafka在团队里虽然存在，但为了这路日志专门引一条Kafka链路，运维成本和延迟都上来了。最直接的方案就是：Flume Source照常收集日志，Channel做缓冲，Sink直接批量写入ClickHouse。

1.2 三条写入路径的对比，为什么选HTTP

ClickHouse官方写入路径大致有三类：JDBC、Kafka Engine表、HTTP接口。我做了个简单对比：

JDBC直写的方式最容易被想到，但ClickHouse的JDBC驱动偏重，单条INSERT走网络开销大，Batch模式下虽然能攒批，但吞吐和HTTP路径比还是有差距。Kafka中转当然稳，但多一个组件就多一个故障点。

HTTP接口是ClickHouse暴露的一个原生数据入口，配合RowBinary二进制格式能做到“接近Native协议”的写入效率，不需要额外依赖，只要发POST请求把二进制数据塞进Body即可。这也是我最终选择在自定义Sink里用HTTP + RowBinary的原因。

2. 开发前必须吃透的Flume自定义Sink机制

2.1 三个接口的各自职责

一个完整可用的自定义Sink，至少要跟Flume的四个类打交道：AbstractSink、Sink接口、Configurable接口、LifecycleAware接口。

AbstractSink是基类，它帮你把Sink名字、Channel引用这些公共字段管理好了，你只需要实现process()方法。但真正干活的是两个扩展接口：

• Configurable：提供configure(Context context)，Flume启动时会把你配置里的键值对以Context形式传进来，你在这里做参数绑定和校验。
• LifecycleAware：提供start()和stop()，被Flume的SinkRunner在Agent生命周期里调用，你的连接池、HTTP Client、状态清理都要放这里。

Closure顺序是：配置解析 → start()初始化 → process()循环执行 → stop()释放资源。

我见过不少人把HttpClient初始化写在configure()里，这其实不对。configure()在Agent启动和配置热加载时都可能被调用，而start()只在上层Runner准备就绪后才调用。连接资源放configure里，可能出现多次构建、空耗内存的问题。

2.2 process()返回值和Channel事务是核心

process()是Sink的主循环，Flume的SinkRunner会反复调用这个方法，每次调用代表一个处理周期。它有两个返回值：

• Status.READY：表示本次处理完成且有数据，告诉Runner可以立即继续调用。
• Status.BACKOFF：表示本次没有足够数据或处理失败，让Runner稍微休眠再试，避免空转打满CPU。

事务边界是这里最容易翻车的地方。process()内部要手动拿Channel事务：

Channel channel = getChannel(); Transaction txn = channel.getTransaction(); txn.begin(); try { // take数据、发送到ClickHouse txn.commit(); return Status.READY; } catch (Exception e) { txn.rollback(); return Status.BACKOFF; } finally { txn.close(); }

重点在于：take()出来的Event，只有在txn.commit()之后才会从Channel里真正移除。如果你在发送到ClickHouse之前就commit，那么发送失败时数据已经丢了；如果你发送成功之后忘记commit，那么这批数据会被重复消费。正确顺序是：发送成功 → commit；发送失败 → rollback，让Event留在Channel中下次重试。

2.3 配置属性的解析规则

Flume的agent配置文件长这样：

agent.sinks.k1.type = com.example.flume.sink.clickhouse.ClickHouseSink agent.sinks.k1.clickhouse.host = 192.0.2.10

自定义Sink里的Context会拿到前缀剥离后的键值对，也就是说k1.clickhouse.host会以“clickhouse.host”为key传给configure()。所以我在代码里用context.getString("clickhouse.host")来取，而不是还带着sink的名字前缀。

一个容易被忽略的坑：所有配置值都是String，你要自己控制类型转换失败的处理。我的习惯是给每个字段提供默认值，能缺省的都缺省，但关键字段比如host和table必须强校验，启动时就报错，好过运行到一半才发现配置错了。

3. 完整开发实录：从空项目到能跑通的Sink

3.1 项目结构及Maven依赖

我用Maven管理，包名是com.example.flume.sink.clickhouse。目录结构就是标准的maven项目，总共只有一个核心类。pom.xml里关键依赖如下：

<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.httpcomponents</groupId> <artifactId>httpclient</artifactId> <version>4.5.13</version> </dependency> <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>1.7.30</version> </dependency> </dependencies>

flume-ng-core一定要用provided，因为最终是扔到Flume的lib目录下由Flume进程加载，避免版本冲突。httpclient需要打进去，因为Flume自带的lib里不一定有。

3.2 核心类骨架和参数绑定

我的ClickHouseSink继承AbstractSink，同时实现Configurable、LifecycleAware。字段设计上，把连接参数、批量参数、格式参数分清楚：

public class ClickHouseSink extends AbstractSink implements Configurable, LifecycleAware { private static final Logger LOG = LoggerFactory.getLogger(ClickHouseSink.class); private String clickhouseHost; private int clickhousePort = 8123; private String clickhouseDatabase = "default"; private String clickhouseTable; private String clickhouseUser; private String clickhousePassword; private int batchSize = 1000; private int connectTimeoutMs = 5000; private int queryTimeoutMs = 30000; private String fieldSeparator = "\t"; private String[] columnNames = new String[0]; private CloseableHttpClient httpClient; @Override public void configure(Context context) { clickhouseHost = context.getString("clickhouse.host"); clickhousePort = context.getInteger("clickhouse.port", 8123); clickhouseDatabase = context.getString("clickhouse.database", "default"); clickhouseTable = context.getString("clickhouse.table"); clickhouseUser = context.getString("clickhouse.user", ""); clickhousePassword = context.getString("clickhouse.password", ""); batchSize = context.getInteger("clickhouse.batchSize", 1000); connectTimeoutMs = context.getInteger("clickhouse.connectTimeout", 5000); queryTimeoutMs = context.getInteger("clickhouse.queryTimeout", 30000); fieldSeparator = context.getString("clickhouse.fieldSeparator", "\t"); String columns = context.getString("clickhouse.columns", ""); if (!columns.isEmpty()) { columnNames = columns.split(","); } Preconditions.checkArgument(StringUtils.isNotBlank(clickhouseHost), "clickhouse.host must be configured"); Preconditions.checkArgument(StringUtils.isNotBlank(clickhouseTable), "clickhouse.table must be configured"); } }

Preconditions.checkArgument来自guava，Flume本身带了guava，直接用就行。我建议关键参数校验在这里做掉，因为configure阶段报错时，Flume启动日志会明确提示哪行配置有问题，省得运行期排查。

3.3 process()实现：批量take和事务处理

这个方法是整个Sink的心脏。我的逻辑不复杂：一次从Channel里最多take batchSize个Event，攒成一批后发给ClickHouse，发送成功才commit。

@Override public Status process() throws EventDeliveryException { Channel channel = getChannel(); Transaction txn = channel.getTransaction(); txn.begin(); try { List<Event> events = new ArrayList<>(batchSize); for (int i = 0; i < batchSize; i++) { Event event = channel.take(); if (event == null) { break; } events.add(event); } if (events.isEmpty()) { txn.commit(); return Status.BACKOFF; } writeToClickHouse(events); txn.commit(); return Status.READY; } catch (Exception e) { LOG.error("Write to ClickHouse failed, will retry", e); txn.rollback(); return Status.BACKOFF; } finally { txn.close(); } }

这里有个细节，take()到null不代表Channel空了，也可能只是当前没有可消费数据，但更多时候确实说明Channel里没有pending数据。返回BACKOFF可以让Runner稍微等等再进下一个循环，不然进程会一直空转。

还有一个容易被忽视的问题：Event的数量达到batchSize之前就break了，这一批是“不满批”的。这在低峰期很正常，但我建议不要为了凑够batchSize无限等下去，因为Channel里可能长时间没有新数据，Sink会卡在这里。Flume的Runner机制保证了process()返回后会立刻再次调用，所以每轮都尝试take一次是最高效的。

3.4 RowBinary报文构造：把Event变成ClickHouse认识的二进制

ClickHouse的HTTP接口支持多种格式，最推荐RowBinary。它是一套紧凑的二进制编码：对字符串类型，先写一个VarUInt表示字节长度，再写原始字节；对整数类型，按小端序写固定字节数。这比JSON格式省带宽，也比CSV格式更少歧义。

每个Event的body我从上游约定好是一个按分隔符分隔的文本行，然后根据配置的columnNames按顺序编码：

private byte[] buildRowBinary(List<Event> events) throws IOException { ByteArrayOutputStream bos = new ByteArrayOutputStream(); for (Event event : events) { String line = new String(event.getBody(), StandardCharsets.UTF_8); String[] fields = line.split(fieldSeparator, -1); if (columnNames.length > 0) { for (int i = 0; i < columnNames.length; i++) { String value = i < fields.length ? fields[i] : ""; writeString(bos, value); } } else { writeString(bos, line); } } return bos.toByteArray(); } private static void writeString(ByteArrayOutputStream bos, String value) throws IOException { byte[] bytes = value.getBytes(StandardCharsets.UTF_8); writeVarUInt(bos, bytes.length); bos.write(bytes); } private static void writeVarUInt(ByteArrayOutputStream bos, long value) { while (value >= 0x80) { bos.write((int) value | 0x80); value >>= 7; } bos.write((int) value); }

RowBinary里的VarUInt是变长编码，每一字节的低7位是数据位，最高位是“是否继续”的标记。上面的writeVarUInt实现的就是这个逻辑。

这里必须强调一个生产级问题：如果表里既有String又有DateTime、Int64这类字段，RowBinary要求每个字段按对应类型编码，比如DateTime是UInt32秒级时间戳，不是字符串。你不能统一按String的方式发，否则ClickHouse会报错或者写进错乱数据。所以生产环境我实际是维护了一个“列名到类型”的映射，在写RowBinary时根据类型分流：

private static void writeDateTime(ByteArrayOutputStream bos, String value) throws Exception { SimpleDateFormat fmt = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); Date date = fmt.parse(value); long epoch = date.getTime() / 1000; for (int i = 0; i < 4; i++) { bos.write((int) (epoch >> (8 * i)) & 0xFF); } }

如果是Nullable类型，还要先写一个字节标记是否为NULL。基础版本可以先全部按String字段处理，但上线前务必逐一核对表结构，特别是那些有默认值、有空值的列。

3.5 HTTP发送与结果校验

数据包构造好后，通过HttpClient POST到ClickHouse。URL格式是：

http://host:port/?query=INSERT INTO db.table FORMAT RowBinary

需要把query参数做URL编码，然后整个二进制包作为POST body发送：

private void writeToClickHouse(List<Event> events) throws IOException { String insertQuery = "INSERT INTO " + clickhouseDatabase + "." + clickhouseTable + " FORMAT RowBinary"; String url = "http://" + clickhouseHost + ":" + clickhousePort + "/?query=" + URLEncoder.encode(insertQuery, "UTF-8") + (StringUtils.isNotBlank(clickhouseUser) ? "&user=" + URLEncoder.encode(clickhouseUser, "UTF-8") : "") + (StringUtils.isNotBlank(clickhousePassword) ? "&password=" + URLEncoder.encode(clickhousePassword, "UTF-8") : ""); byte[] payload = buildRowBinary(events); HttpPost post = new HttpPost(url); post.setHeader("Content-Type", "application/octet-stream"); post.setEntity(new ByteArrayEntity(payload)); try (CloseableHttpResponse resp = httpClient.execute(post)) { int code = resp.getStatusLine().getStatusCode(); if (code != 200) { String body = EntityUtils.toString(resp.getEntity(), StandardCharsets.UTF_8); throw new IOException("ClickHouse write failed, code=" + code + ", body=" + body); } } }

为什么一定要检查HTTP状态码？因为ClickHouse的HTTP接口在没有开启异常输出时，可能返回200但body是空的，这算成功；但大多数错误会通过非200状态码返回，body里带错误信息。你发的这批数据如果格式不对，ClickHouse会拒绝整个请求，不会只跳过坏行，所以失败就整批回滚，让数据留在Channel里重试。

3.6 配置文件样例和部署验证流程

写好的Sink要接入Flume Agent。下面是我在测试环境用的配置：

agent.sources = src1 agent.channels = ch1 agent.sinks = sink1 agent.sources.src1.type = spooldir agent.sources.src1.spoolDir = /data/logs/app agent.channels.ch1.type = memory agent.channels.ch1.capacity = 100000 agent.channels.ch1.transactionCapacity = 10000 agent.sinks.sink1.type = com.example.flume.sink.clickhouse.ClickHouseSink agent.sinks.sink1.clickhouse.host = 192.0.2.10 agent.sinks.sink1.clickhouse.port = 8123 agent.sinks.sink1.clickhouse.database = default agent.sinks.sink1.clickhouse.table = app_log agent.sinks.sink1.clickhouse.user = default agent.sinks.sink1.clickhouse.password = secret agent.sinks.sink1.clickhouse.batchSize = 2000 agent.sinks.sink1.clickhouse.columns = ts,level,msg agent.sinks.sink1.clickhouse.fieldSeparator = \t agent.sinks.sink1.channel = ch1

部署步骤就三步：

1. 把项目打成jar，丢到Flume安装目录的lib下。
2. 重启Flume，日志里应该能看到“ClickHouse Sink started”以及你打印的连接参数。
3. 往spoolDir里丢测试文件，然后在ClickHouse客户端执行select count(*)看数据有没有进来。

我习惯先起一个本地单机环境验证，确认二进制格式没问题后，再改真正的接入配置。

4. 可靠性、性能与生产落地经验

4.1 batchSize和流量的数学账

batchSize直接决定吞吐。RowBinary是小批量多次HTTP请求，还是大批量少次请求，差别非常大。我从经验里给一个参考区间：单批数据量控制在几MB以内，行数在500到5000之间比较稳。

假设一行Event平均1KB，batchSize=2000就是一次约2MB的请求，ClickHouse处理这种量级的POST基本几十毫秒内完成，加上网络RTT，整体延迟可控。如果把batchSize拉到50000，一次请求50MB，ClickHouse端的merge压力和内存占用都会明显抬升，而且一旦失败要回滚重试，代价很大。

还有一个很容易忽略的点：batchSize要跟Channel的transactionCapacity配合。transactionCapacity是每次事务能take/put的最大Event数。如果你Channel的transactionCapacity只有1000，而Sink的batchSize配了5000，那一次process里最多也就能take1000个，剩下4000的batchSize是无效配置。原则上要保持：

sink batchSize <= channel transactionCapacity

这样才能让batchSize真正生效。

4.2 连接池和超时参数怎么定

我在start()里初始化HttpClient时，特意用了PoolingHttpClientConnectionManager。原因很简单：Flume的Sink是单线程循环调process，但如果你的Agent里配了多个Sink共用一个HttpClient（比如多个Sink线程指向不同表），连接池就能复用TCP连接，避免频繁建立/断开连接。

Start阶段初始化逻辑：

@Override public synchronized void start() { RequestConfig requestConfig = RequestConfig.custom() .setConnectTimeout(connectTimeoutMs) .setSocketTimeout(queryTimeoutMs) .setConnectionRequestTimeout(connectTimeoutMs) .build(); PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager(); cm.setMaxTotal(50); cm.setDefaultMaxPerRoute(30); httpClient = HttpClients.custom() .setDefaultRequestConfig(requestConfig) .setConnectionManager(cm) .build(); LOG.info("ClickHouse Sink started, host={}, db={}, table={}, batchSize={}", clickhouseHost, clickhouseDatabase, clickhouseTable, batchSize); }

超时参数我一般这样定：connectTimeout设5秒，queryTimeout设30秒。ClickHouse在高峰时可能出现排队，30秒能覆盖大部分慢查询，但如果超过30秒还没响应，基本可以判定这批写入超时了，回滚重试更合理。

4.3 幂等与重复消费的权衡

这是我做这个Sink时踩得最深的一个坑。

如果写入失败回滚，Channel里的Event会重新被Sink take到，然后重新发给ClickHouse。如果第一次请求其实已经到达ClickHouse并执行成功，只是响应超时了，那么重试就会产生重复数据。Flume不可能像数据库一样提供精确一次的语义，它只能保证至少一次。所以业务侧必须接受“可能重复”这个事实，最好的办法是ClickHouse表用ReplacingMergeTree或者去重引擎，以日志主键做去重。

我在生产上将表设计成ReplacingMergeTree，以业务唯一ID作为排序键。ClickHouse在后台merge时会自动去重，查询结果不重复。如果你的日志天生没有唯一ID，那就需要在上游生成一个，否则重复数据会一直堆积。

4.4 多Sink并行和压缩传输

单个Agent的Sink处理能力是有上限的。单Sink单线程，即使每批2000条，日志量大时CPU和网络也可能成为瓶颈。Flume支持一个Channel挂多个Sink，但默认的SinkProcessor是DefaultSinkProcessor，只允许一个Sink。要让多个Sink并行消费同一Channel，需要改成LoadBalancingSinkProcessor：

agent.sinks.sink1.type = com.example.flume.sink.clickhouse.ClickHouseSink agent.sinks.sink2.type = com.example.flume.sink.clickhouse.ClickHouseSink agent.sinks.sink-group.sinks = sink1 sink2 agent.sinks.sink-group.processor.type = load_balance agent.sinks.sink-group.processor.backoff = true

这样两个Sink会并行take数据，整体吞吐接近翻倍。但要注意：如果多个Sink把数据打到同一个ClickHouse节点，ClickHouse端的写入压力也要看能不能扛住。

带宽紧张时，还可以在HTTP请求上启用压缩。ClickHouse支持gzip压缩的POST body，只要在URL上带enable_http_compression=1并且请求头带上Content-Encoding: gzip。这在高流量日志场景下能把网络流量砍掉一大截，代价是Sink端多一次压缩的CPU开销。

4.5 生产参数参考表

5. 常见问题与排查实录

5.1 问题速查表

5.2 高频问题逐条拆解

问题1：RowBinary报“Invalid number of columns”

这个错误几乎都是clickhouse.columns配置和表结构对不上。比如表有5列，但columns只配了3个字段名，那么RowBinary只写了3个字段的值，ClickHouse就认为客户端提交的列数不匹配。解决方法是把表结构完整列出来，包括默认列，或者在INSERT语句里显式指定列名，比如：

INSERT INTO app_log (ts, level, msg) FORMAT RowBinary

这样ClickHouse只会按这3列解析，其他列走默认值。

问题2：写入成功但查不到数据

我遇到过几次，问题出在数据库和表名的大小写上，比如ClickHouse是区分大小写的，配置里写App_Log而表实际是小写app_log，写入请求可能被重定向到另一个库。另外检查一下是不是写到了别的节点上。如果是分布式表，写入请求会重定向到对应分片，查询时如果查的是本地表而不是分布式表，数据在另一个分片里当然查不到。

问题3：大量重复数据

这个问题我在4.3里说过，本质是“至少一次”语义导致的。如果业务对重复零容忍，除了用ReplacingMergeTree，还可以在Sink里维护一个“最近发送成功的请求指纹”缓存，比如对一批事件的body做hash，如果短时间内收到相同hash的批次就丢弃。但这个做法有副作用，会牺牲一部分可靠性，我不建议在公共场景下用，除非业务完全不能接受重复。

问题4：Flume进程OOM

Flume是Java进程，堆内存是固定的。Channel配置的capacity是个数组，理论上会占内存，但这个占用量不高。真正的内存黑洞是你把batchSize调得过大，比如一次take几万条Event到List里，每条Event的body都还在堆上，瞬间就是几百MB。生产环境我见过把batchSize配到100000然后进程直接挂掉的案例。把batchSize控制在5000以内，单批内存就在可控范围。同时留意Channel的capacity和transactionCapacity，这两个值太大也会拖累GC。

问题5：写ClickHouse时频繁报超时

ClickHouse在后台做merge时，如果正在写入的表分区很多，或者磁盘性能差，查询会排队。这时候你的queryTimeout设得再大也没用。建议优先控制写入频率和包大小，让ClickHouse能及时消化。如果写入量确实很大，考虑把表按天分区，减少单个分区的merge压力。另外确保ClickHouse的max_part_per_partition参数没有被频繁触顶，否则写入会被迫等待。

5.3 上线前值得做的三个压测项目

自定义Sink这种代码，不做压测直接上生产就是给自己埋雷。我每次交付前至少会跑三轮针对性测试：

第一轮是格式压测。用真实结构的Event数据灌进去，在ClickHouse端查每列的值、类型、NULL情况，重点验证DateTime、Nullable这些特殊字段的编码对不对。

第二轮是吞吐压测。用不同的batchSize跑一遍，记下每次请求耗时和CPU占用，画出“batchSize vs 吞吐”的曲线，选择拐点作为生产配置。注意同时观察Flume进程的堆内存和GC频率。

第三轮是故障演练。把ClickHouse节点停掉，观察Flume的Channel是否堆积、数据有没有丢、重启ClickHouse后Sink能否自动恢复。这一步能验证事务回滚机制是否真的可靠。

写在最后的一些经验

这套自定义Sink写完后，我把同样的思路复制到了另一个离线的数据同步任务里，只是把HTTP发送端换成另一个目标系统，Sink的骨架基本没动。Flume自定义Sink的开发门槛其实不高，真正决定质量的是事务边界、批量参数和故障处理这三件事。

最后分享一个我个人的习惯：上线前一定先用测试Channel和模拟数据灌一遍，观察内存和ClickHouse侧的merge停顿，再决定batchSize最终定多少。我见过太多人照搬网上的配置直接上生产，结果高峰一来就是一堆504和线程阻塞。参数永远是调出来的，不是抄出来的。