一、背景与痛点

业务场景

在实时大数据处理场景中，Flink + ClickHouse 的组合被广泛应用于：

• 日志处理： 海量应用日志实时写入分析库。
• 监控分析： 业务指标、APM 数据的实时聚合。

这些场景的共同特点：

• 数据量大：百万级 TPS，峰值可达千万级。
• 写入延迟敏感： 需要秒级可见。
• 数据准确性要求高：不允许数据丢失。
• 多表写入： 不同数据根据分表策略写入不同的表。

开源 Flink ClickHouse Sink 的痛点

Flink 官方提供的 ClickHouse Sink（flink-connector-jdbc）在生产环境中存在以下严重问题：

痛点一：缺乏基于数据量的攒批机制

问题表现：

// Flink 官方 JDBC Sink 的实现
public class JdbcSink<T> extends RichSinkFunction<T> {
    private final int batchSize;  // 固定批次大小
    @Override
    public void invoke(T value, Context context) {
        bufferedValues.add(value);
        if (bufferedValues.size() >= batchSize) {
            // 只能基于记录数攒批，无法基于数据量
            flush();
        }
    }

带来的问题：

1. 内存占用不可控： 100 条 1KB 的日志和 100 条 10MB 的日志占用内存差距 100 倍。
2. OOM 风险高： 大日志记录（如堆栈转储）会迅速撑爆内存。
3. 写入性能差： 无法根据记录大小动态调整批次，导致小记录批次过大浪费网络开销。

痛点二：无法支持动态表结构

问题表现：

// Flink 官方 Sink 只能写入固定表
public class JdbcSink {
    private final String sql;  // 固定的 INSERT SQL
    public JdbcSink(String jdbcUrl, String sql, ...) {
        this.sql = sql;  // 硬编码的表结构
    }
}

带来的问题：

1. 多应用无法隔离： 所有应用的数据写入同一张表，通过特定分表策略区分。
2. 扩展性差： 新增应用需要手动建表，无法动态路由。
3. 性能瓶颈： 单表数据量过大（百亿级），查询和写入性能急剧下降。

痛点三：分布式表写入性能问题

问题表现：

// 大多数生产实现直接写入分布式表
INSERT INTO distributed_table_all VALUES (...)

ClickHouse 分布式表的工作原理：

带来的问题：

1. 网络开销大： 数据需要经过分布式表层转发，延迟增加。
2. 写入性能差： 分布式表增加了路由和转发逻辑，吞吐量降低。
3. 热点问题： 所有数据先到分布式表节点，再转发，造成单点瓶颈。

生产级方案的核心改进

针对以上痛点，本方案提供了以下核心改进：

改进一：基于数据量的攒批机制

public class ClickHouseSinkCounter {
    private Long metaSize;  // 累计数据量（字节）
    public void add(LogModel value) {
        this.values.add(value);
        this.metaSize += value.getMetaSize();  // 累加数据量
    }
}
// 触发条件
private boolean flushCondition(String application) {
    return checkMetaSize(application)  // metaSize >= 10000 字节
        || checkTime(application);     // 或超时 30 秒
}

优势：

• 内存可控： 根据数据量而非记录数攒批。
• 精确控制： 1KB 的记录攒 10000 条 = 10MB，1MB 的记录攒 10 条 = 10MB。
• 避免OOM： 大日志记录不会撑爆内存。

改进二：动态表结构与分片策略

public abstract class ClickHouseShardStrategy<T> {
    public abstract String getTableName(T data);
}
//日志侧实现为应用级分表
public class LogClickHouseShardStrategy extends ClickHouseShardStrategy<String> {
    @Override
    public String getTableName(String application) {
        // 动态路由：order-service → tb_logs_order_service
        return String.format("tb_logs_%s", application);
    }
}

优势：

• 应用隔离： 日志侧内置应用级分表，每个应用独立分表。
• 动态路由： 根据 application 自动路由到目标表。
• 扩展性强： 新增应用无需手动建表（配合 ClickHouse 自动建表）。

改进三：本地表写入 + 动态节点发现

public class ClickHouseLocalWriter extends ClickHouseWriter {
    // 直接写本地表，避免分布式表转发
    private final ConcurrentMap<String, HikariDataSource> dataSourceMap;
    @Override
    public HikariDataSource getNextDataSource(Set<String> exceptionHosts) {
        // 1. 动态获取集群节点列表
        List<String> healthyHosts = getHealthyHosts(exceptionHosts);
        // 2. 随机选择健康节点
        return dataSourceMap.get(healthyHosts.get(random.nextInt(size)));
    }
}

优势：

• 性能提升： 直接写本地表，避免网络转发。
• 高可用： 动态节点发现 + 故障节点剔除。
• 负载均衡： 随机选择 + Shuffle 初始化。

技术方案概览

基于以上改进，本方案提供了以下核心能力：

1. 本地表/分布式表写入： 性能优化与高可用平衡。
2. 分片策略： 按应用维度路由与隔离。
3. 攒批与内存控制： 双触发机制（数据量 + 超时）。
4. 流量控制与限流： 有界队列 + 连接池。
5. 健壮的重试机制： 递归重试 + 故障节点剔除。
6. Checkpoint 语义保证： At-Least-Once 数据一致性。

二、核心架构设计

架构图

核心组件

核心流程

三、本地表 vs 分布式表写入

ClickHouse 表结构说明

ClickHouse 推荐直接写本地表，原因：

1. 写入性能： 避免分布式表的网络分发。
2. 数据一致性： 直接写入目标节点，减少中间环节故障点，比分布式表写入更安全，利于工程化。
3. 负载均衡： 客户端路由实现负载分散。

-- 本地表（实际存储数据）
CREATE TABLE tb_logs_local ON CLUSTER 'default' (
    application String,
    environment String,
    message String,
    log_time DateTime
) ENGINE = MergeTree()
PARTITION BY toYYYYMM(log_time)
ORDER BY (application, log_time);
-- 分布式表（逻辑视图，不存储数据）
CREATE TABLE tb_logs_all ON CLUSTER 'default' AS tb_logs_local
ENGINE = Distributed('default', dw_log, tb_logs_local, cityHash64(application));

HikariCP 连接池配置

// HikariCP 连接池配置
public class ClickHouseDataSourceUtils {
    private static HikariConfig getHikariConfig(DataSourceImpl dataSource) {
        HikariConfig config = new HikariConfig();
        config.setConnectionTimeout(30000L);    // 连接超时 30s
        config.setMaximumPoolSize(20);          // 最大连接数 20
        config.setMinimumIdle(2);               // 最小空闲 2
        config.setDataSource(dataSource);
        return config;
    }
    private static Properties getClickHouseProperties(ClickHouseSinkCommonParams params) {
        Properties props = new Properties();
        props.setProperty("user", params.getUser());
        props.setProperty("password", params.getPassword());
        props.setProperty("database", params.getDatabase());
        props.setProperty("socket_timeout", "180000");      // Socket 超时 3 分钟
        props.setProperty("socket_keepalive", "true");      // 保持连接
        props.setProperty("http_connection_provider", "APACHE_HTTP_CLIENT");
        return props;
    }
}

配置说明：

• maxPoolSize=20：每个 ClickHouse 节点最多 20 个连接。
• minIdle=2：保持 2 个空闲连接，避免频繁创建。
• socket_timeout=180s：Socket 超时 3 分钟，防止长时间查询阻塞。

ClickHouseLocalWriter：动态节点发现

public class ClickHouseLocalWriter extends ClickHouseWriter {
    // 本地节点缓存，按 IP 维护
    private final ConcurrentMap<String, HikariDataSource> dataSourceMap;
    // 动态获取集群本地表节点
    private final ClusterIpsUtils clusterIpsUtils;
    // IP 变更标志（CAS 锁，避免并发更新）
    private static final AtomicBoolean IP_CHANGING = new AtomicBoolean(false);
    @Override
    public HikariDataSource getNextDataSource(Set<String> exceptionHosts) {
        // 1️⃣ 检测集群节点变化（通过 CAS 避免并发更新）
        if (clusterIpsChanged() && IP_CHANGING.compareAndSet(false, true)) {
            try {
                ipChanged(); // 动态更新 dataSourceMap
            } finally {
                IP_CHANGING.set(false);
            }
        }
        // 2️⃣ 获取异常节点列表（从 Redis + APM 实时查询）
        Set<String> exceptIps = clusterIpsUtils.getExceptIps();
        exceptIps.addAll(exceptionHosts);
        // 3️⃣ 过滤健康节点，随机选择
        List<String> healthyHosts = dataSourceMap.keySet().stream()
            .filter(host -> !exceptIps.contains(host))
            .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(healthyHosts)) {
            throw new RuntimeException("Can't get datasource from local cache");
        }
        return dataSourceMap.get(healthyHosts.get(random.nextInt(healthyHosts.size())));
    }
    private void ipChanged() {
        List<String> clusterIps = clusterIpsUtils.getClusterIps();
        // 新增节点：自动创建连接池
        clusterIps.forEach(ip ->
            dataSourceMap.computeIfAbsent(ip, v ->
                createHikariDataSource(ip, port)
            )
        );
        // 移除下线节点：关闭连接池
        dataSourceMap.forEach((ip, ds) -> {
            if (!clusterIps.contains(ip)) {
                dataSourceMap.remove(ip);
                ds.close();
            }
        });
    }
}

核心逻辑：

1. 动态节点发现： 从 system.clusters 查询所有节点。
2. 自动扩缩容： 节点上线自动加入，下线自动剔除。
3. 故障节点剔除： 通过 APM 监控，自动剔除异常节点。
4. 负载均衡： 随机选择健康节点，避免热点。

集群节点动态发现（ClusterIpsUtils）

public class ClusterIpsUtils {
    // 从 system.clusters 查询所有节点
    private static final String QUERY_CLUSTER_IPS =
        "select host_address from system.clusters where cluster = 'default'";
    // LoadingCache：定时刷新节点列表（1 小时）
    private final LoadingCache<String, List<String>> clusterIpsCache =
        CacheBuilder.newBuilder()
            .expireAfterAccess(10, TimeUnit.HOURS)
            .refreshAfterWrite(1, TimeUnit.HOURS)
            .build(CacheLoader.asyncReloading(new CacheLoader<>() {
                @Override
                public List<String> load(String dbName) {
                    return queryClusterIps();  // 定时刷新节点列表
                }
            }));
    // 异常节点缓存（1 分钟刷新）
    private final LoadingCache<String, FlinkExceptIpModel> exceptIpsCache =
        CacheBuilder.newBuilder()
            .refreshAfterWrite(1, TimeUnit.MINUTES)
            .build(CacheLoader.asyncReloading(new CacheLoader<>() {
                @Override
                public FlinkExceptIpModel load(String dbName) {
                    return queryExceptIp();  // 从 Redis + APM 查询异常节点
                }
            }));
}

异常节点监控策略：

• 磁盘使用率 >= 90%： 从 APM 查询 Prometheus 指标，自动加入黑名单。
• HTTP 连接数 >= 50： 连接数过多说明节点压力大，自动加入黑名单。
• 人工配置： 通过 Redis 配置手动剔除节点

数据来源：

1. ClickHouse system.clusters 表： 获取所有集群节点。
2. APM Prometheus 接口： 监控节点健康状态。
3. Redis 缓存： 人工配置的异常节点。

负载均衡优化

public class ClickHouseWriter {
    public <T> ClickHouseWriter(...) {
        // Shuffle：随机打乱数据源顺序
        Collections.shuffle(clickHouseDataSources);
        this.clickHouseDataSources = clickHouseDataSources;
    }
    public HikariDataSource getNextDataSource(Set<String> exceptionHosts) {
        // 轮询 + 随机选择（已 shuffle，避免热点）
        int current = this.currentRandom.getAndIncrement();
        if (current >= clickHouseDataSources.size()) {
            this.currentRandom.set(0);
        }
        return clickHouseDataSources.get(currentRandom.get());
    }
}

优势：

• 初始化时 shuffle，避免所有 writer 同时从第一个节点开始。
• 轮询 + 随机选择，负载分散更均匀。
• 故障节点自动剔除。

四、支持分表策略

分片策略抽象

public abstract class ClickHouseShardStrategy<T> {
    private String tableName;      // 表名模板，如 "tb_log_%s"
    private Integer tableCount;    // 分表数量
    // 根据数据决定目标表名
    public abstract String getTableName(T data);
}

日志分片实现

public class LogClickHouseShardStrategy extends ClickHouseShardStrategy<String> {
    @Override
    public String getTableName(String application) {
        // 表名格式：tb_log_{application}
        // 例如：application = "order-service" -> table = "tb_log_order_service"
        return String.format(
            this.getTableName(),
            application.replace("-", "_").toLowerCase()
        );
    }
}

按表（应用）维度的缓冲区

日志侧维度降级为应用名称维度缓冲区，实则因为按照应用分表，

业务方可使用自身分表策略携带表名元数据，进行表维度缓冲。

public class ClickHouseShardSinkBuffer {
    // 按 application 分组的缓冲区（ConcurrentHashMap 保证并发安全）
    private final ConcurrentHashMap<String, ClickHouseSinkCounter> localValues;
    public void put(LogModel value) {
        String application = value.getApplication();
        // 1️⃣ 检查是否需要 flush
        if (flushCondition(application)) {
            addToQueue(application); // 触发写入
        }
        // 2️⃣ 添加到缓冲区（线程安全的 compute 操作）
        localValues.compute(application, (k, v) -> {
            if (v == null) v = new ClickHouseSinkCounter();
            v.add(value);
            return v;
        });
    }
    private void addToQueue(String application) {
        localValues.computeIfPresent(application, (k, v) -> {
            // 深拷贝并清空（避免并发修改异常）
            List<LogModel> deepCopy = v.copyValuesAndClear();
            // 构造请求 Blank：application + targetTable + values
            String targetTable = shardStrategy.getTableName(application);
            ClickHouseRequestBlank blank = new ClickHouseRequestBlank(deepCopy, application, targetTable);
            // 放入队列
            writer.put(blank);
            return v;
        });
    }
}

核心设计：

• 应用隔离： 每个表（应用）独立的 buffer，互不影响。
• 线程安全： 使用 ConcurrentHashMap.compute()保证并发安全。
• 深拷贝： List.copyOf() 创建不可变副本，避免并发修改。
• 批量清空： 一次性取出所有数据，清空计数器。

五、攒批与内存控制

双触发机制

public class ClickHouseShardSinkBuffer {
    private final int maxFlushBufferSize;  // 最大批次大小（如 10000）
    private final long timeoutMillis;      // 超时时间（如 30s）
    // 触发条件检查（满足任一即触发）
    private boolean flushCondition(String application) {
        return localValues.get(application) != null
            && (checkMetaSize(application) || checkTime(application));
    }
    // 条件1：达到批次大小
    private boolean checkMetaSize(String application) {
        return localValues.get(application).getMetaSize() >= maxFlushBufferSize;
    }
    // 条件2：超时
    private boolean checkTime(String application) {
        long current = System.currentTimeMillis();
        return current - localValues.get(application).getInsertTime() > timeoutMillis;
    }
}

批次大小计算

public class ClickHouseSinkCounter {
    private final List<LogModel> values;
    private Long metaSize; // 累计的 metaSize（字节）
    public void add(LogModel value) {
        this.values.add(value);
        this.metaSize += value.getMetaSize(); // 累加 metaSize
    }
    public List<LogModel> copyValuesAndClear() {
        List<LogModel> logModels = List.copyOf(this.values); // 深拷贝（不可变）
        this.values.clear();
        this.metaSize = 0L;
        this.insertTime = System.currentTimeMillis();
        return logModels;
    }
}

关键点：

• 使用 metaSize（字节数）而非记录数控制批次，内存控制更精确。
• List.copyOf() 创建不可变副本，避免并发修改。
• 清空后重置 insertTime，保证超时触发准确性。

带随机抖动的超时

private final long timeoutMillis;
public ClickHouseShardSinkBuffer(..., int timeoutSec, ...) {
    // 基础超时 + 10% 随机抖动（避免惊群效应）
    this.timeoutMillis = TimeUnit.SECONDS.toMillis(timeoutSec)
                      + new SecureRandom().nextInt((int) (timeoutSec * 0.1 * 1000));
}

目的： 避免多个TM 同时触发 flush，造成写入流量峰值。

配置示例

ClickHouseShardSinkBuffer.Builder
    .aClickHouseSinkBuffer()
    .withTargetTable("single_table")  //单表时，可直接使用指定表名
    .withMaxFlushBufferSize(10000)  // 对应字节数
    .withTimeoutSec(30)              // 30 秒超时
    .withClickHouseShardStrategy(new LogClickHouseShardStrategy("table_prefix_%s", 8))  //分表策略时，使用
    // 分表策略可根据业务实际情况进行扩展
    .build(clickHouseWriter);

六、写入限流与流量控制

有界队列设计

public class ClickHouseWriter {
    // 有界阻塞队列
    private final BlockingQueue<ClickHouseRequestBlank> commonQueue;
    public ClickHouseWriter(ClickHouseSinkCommonParams sinkParams, ...) {
        // 队列最大容量配置（默认 10）
        this.commonQueue = new LinkedBlockingQueue<>(sinkParams.getQueueMaxCapacity());
    }
    public void put(ClickHouseRequestBlank params) {
        unProcessedCounter.incrementAndGet();
        // put() 方法在队列满时会阻塞，实现背压
        commonQueue.put(params);
    }
}

背压传导：

线程池并发控制

public class ClickHouseWriter {
    private final int numWriters; // 写入线程数
    private ExecutorService service;
    private void buildComponents() {
        ThreadFactory threadFactory = ThreadUtil.threadFactory("clickhouse-writer");
        service = Executors.newFixedThreadPool(numWriters, threadFactory);
        // 创建多个 WriterTask 并提交
        for (int i = 0; i < numWriters; i++) {
            WriterTask task = new WriterTask(i, commonQueue, sinkParams, futures, unProcessedCounter);
            service.submit(task);
        }
    }
}

WriterTask 消费逻辑

class WriterTask implements Runnable {
    @Override
    public void run() {
        isWorking = true;
        while (isWorking || !queue.isEmpty()) {
            // poll() 超时返回（100ms），避免无限等待
            ClickHouseRequestBlank blank = queue.poll(100, TimeUnit.MILLISECONDS);
            if (blank != null) {
                // 创建 Future 并设置超时（3 分钟）
                CompletableFuture<Boolean> future = new CompletableFuture<>();
                future.orTimeout(3, TimeUnit.MINUTES);
                futures.add(future);
                try {
                    send(blank, future, new HashSet<>());
                } finally {
                    // final 进行未知异常兜底，防止为捕获异常造成future状态不完成，永久阻塞
                    if (!future.isDone()) {
                        future.completeExceptionally(new RuntimeException("Unknown exception"));
                    }
                    queueCounter.decrementAndGet();
                }
            }
        }
    }
}

配置参数

七、重试机制与超时控制

Future 超时控制

public class ClickHouseWriter {
    private final int numWriters; // 写入线程数
    private ExecutorService service;
    private void buildComponents() {
        ThreadFactory threadFactory = ThreadUtil.threadFactory("clickhouse-writer");
        service = Executors.newFixedThreadPool(numWriters, threadFactory);
        // 创建多个 WriterTask 并提交
        for (int i = 0; i < numWriters; i++) {
            WriterTask task = new WriterTask(i, commonQueue, sinkParams, futures, unProcessedCounter);
            service.submit(task);
        }
    }
}

超时策略：

• Future 超时： 3 分钟（orTimeout）。
• Socket 超时： 3 分钟（socket_timeout=180000）。
• 连接超时： 30 秒（connectionTimeout=30000）。

重试逻辑

class WriterTask implements Runnable {
    @Override
    public void run() {
        isWorking = true;
        while (isWorking || !queue.isEmpty()) {
            // poll() 超时返回（100ms），避免无限等待
            ClickHouseRequestBlank blank = queue.poll(100, TimeUnit.MILLISECONDS);
            if (blank != null) {
                // 创建 Future 并设置超时（3 分钟）
                CompletableFuture<Boolean> future = new CompletableFuture<>();
                future.orTimeout(3, TimeUnit.MINUTES);
                futures.add(future);
                try {
                    send(blank, future, new HashSet<>());
                } finally {
                    // final 进行未知异常兜底，防止为捕获异常造成future状态不完成，永久阻塞
                    if (!future.isDone()) {
                        future.completeExceptionally(new RuntimeException("Unknown exception"));
                    }
                    queueCounter.decrementAndGet();
                }
            }
        }
    }
}

重试控制逻辑

private void handleUnsuccessfulResponse(..., Set<String> exceptHosts) {
    // 检查 Future 是否已完成（避免重复完成）
    if (future.isDone()) {
        return;
    }
    if (attemptCounter >= maxRetries) {
        // 达到最大重试次数，标记失败
        future.completeExceptionally(new RuntimeException("Max retries exceeded"));
    } else {
        // 递归重试
        requestBlank.incrementCounter();
        send(requestBlank, future, exceptHosts); // 递归调用，排除失败节点
    }
}

重试策略：

• 递归重试： 失败后递归调用，直到成功或达到最大次数。
• 异常节点隔离： 每次重试时排除失败的节点（exceptHosts）。
• 超时控制： Future 超时（3 分钟）防止永久阻塞。

为什么递归重试是更好的选择

递归重试（当前实现）

队列重试（假设方案）

保证一致性

  // ClickHouseWriter.java:139-158
  while (!futures.isEmpty() || unProcessedCounter.get() > 0) {
      CompletableFuture<Void> future = FutureUtil.allOf(futures);
      future.get(3, TimeUnit.MINUTES);  // 阻塞直到全部完成
  }

• Checkpoint 时所有数据要么全部成功，要么全部失败。
• 重启后不会有部分数据重复的问题。

简单可靠

• 代码逻辑清晰。
• 对于队列重试且不重复，需要复杂的二阶段提交（这里暂不展开），大幅增加代码复杂度。

性能可接受

class WriterTask implements Runnable {
    @Override
    public void run() {
        while (isWorking || !queue.isEmpty()) {
            ClickHouseRequestBlank blank = queue.poll(100, TimeUnit.MILLISECONDS);
            if (blank != null) {
                // 创建 Future 并设置 3 分钟超时
                CompletableFuture<Boolean> future = new CompletableFuture<>();
                future.orTimeout(3, TimeUnit.MINUTES); // 防止永久阻塞
                futures.add(future);
                try {
                    send(blank, future, new HashSet<>());
                } finally {
                    if (!future.isDone()) {
                        future.completeExceptionally(new RuntimeException("Timeout"));
                    }
                    queueCounter.decrementAndGet();
                }
            }
        }
    }
}

• 虽然阻塞，但有超时保护。
• ClickHouse 写入通常很快（秒级）。
• 网络故障时重试也合理。

避开故障节点

  // ClickHouseWriter.java:259-260
  HikariDataSource dataSource = getNextDataSource(exceptHosts);

• 递归时可以传递 exceptHosts。
• 自动避开失败的节点。
• 提高成功率。

异常节点剔除

// 特殊错误码列表（自动加入黑名单）
private final List<Integer> ignoreHostCodes = Arrays.asList(210, 1002);
public HikariDataSource getNextDataSource(Set<String> exceptionHosts) {
    if (CollectionUtils.isNotEmpty(exceptionHosts)) {
        // 过滤异常节点
        List<HikariDataSource> healthyHosts = clickHouseDataSources.stream()
            .filter(ds -> !exceptionHosts.contains(getHostFromUrl(ds)))
            .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(healthyHosts)) {
            return null; // 所有节点都异常
        }
        return healthyHosts.get(random.nextInt(healthyHosts.size()));
    }
    // 正常轮询（已 shuffle，避免热点）
    return clickHouseDataSources.get(currentRandom.getAndIncrement() % size);
}

故障节点剔除策略：

1. 错误码 210（网络异常）： 自动加入黑名单。
2. 错误码 1002（连接池异常）： 自动加入黑名单。
3. APM 监控： 磁盘 >= 90%、HTTP 连接 >= 50 的节点。
4. 手动配置： 通过 Redis 配置剔除。

恢复机制：

• LoadingCache 定时刷新（1 分钟）。
• 节点恢复健康后自动从黑名单移除。

重试流程图

八、异常处理模式

两种 Sink 模式

public Sink buildSink(String targetTable, String targetCount, int maxBufferSize) {
    IClickHouseSinkBuffer buffer = ClickHouseShardSinkBuffer.Builder
        .aClickHouseSinkBuffer()
        .withTargetTable(targetTable)
        .withMaxFlushBufferSize(maxBufferSize)
        .withClickHouseShardStrategy(new LogClickHouseShardStrategy(targetTable, count))
        .build(clickHouseWriter);
    // 根据配置选择模式
    if (ignoringClickHouseSendingExceptionEnabled) {
        return new UnexceptionableSink(buffer);  // 忽略异常
    } else {
        return new ExceptionsThrowableSink(buffer); // 抛出异常
    }
}

UnexceptionableSink（忽略异常 - At-Most-Once）

public class UnexceptionableSink implements Sink<LogModel> {
    private final IClickHouseSinkBuffer<LogModel> buffer;
    @Override
    public void put(LogModel message) {
        buffer.put(message);  // 不检查 Future 状态
    }
    @Override
    public void flush() {
        buffer.flush();
    }
}

适用场景：

• 允许部分数据丢失。
• 不希望因写入异常导致任务失败。
• 对数据准确性要求不高（如日志统计）。

语义保证：At-Most-Once（最多一次）

ExceptionsThrowableSink（抛出异常 - At-Least-Once）

public class ExceptionsThrowableSink implements Sink<LogModel> {
    private final IClickHouseSinkBuffer<LogModel> buffer;
    @Override
    public void put(LogModel message) throws ExecutionException, InterruptedException {
        buffer.put(message);
        // 每次写入都检查 Future 状态
        buffer.assertFuturesNotFailedYet();
    }
    @Override
    public void flush() throws ExecutionException, InterruptedException {
        buffer.flush();
    }
}

Future 状态检查：

public void assertFuturesNotFailedYet() throws ExecutionException, InterruptedException {
    CompletableFuture<Void> future = FutureUtil.allOf(futures);
    // 非阻塞检查
    if (future.isCompletedExceptionally()) {
        logger.error("There is something wrong with the future. exist sink now");
        future.get(); // 抛出异常，导致 Flink 任务失败
    }
}

适用场景：

• 数据准确性要求高。
• 需要保证所有数据写入成功。
• 异常时希望 Flink 任务失败并重启。

语义保证：At-Least-Once（至少一次）

Future 清理策略与并发控制

定时检查器

public class ClickHouseSinkScheduledCheckerAndCleaner {
    private final ScheduledExecutorService scheduledExecutorService;
    private final List<CompletableFuture<Boolean>> futures;
    // ⚠️ volatile 保证多线程可见性（关键并发控制点）
    private volatile boolean isFlushing = false;
    public ClickHouseSinkScheduledCheckerAndCleaner(...) {
        // 单线程定时执行器
        scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(factory);
        // 定时执行清理任务（每隔 checkTimeout 秒，默认 30 秒）
        scheduledExecutorService.scheduleWithFixedDelay(getTask(), ...);
    }
    private Runnable getTask() {
        return () -> {
            synchronized (this) {
                //  关键：检查是否正在 flush，避免并发冲突
                if (isFlushing) {
                    return; // Checkpoint 期间暂停清理
                }
                // 1️⃣ 清理已完成的 Future
                futures.removeIf(filter);
                // 2️⃣ 触发所有 Buffer 的 flush（检查是否需要写入）
                clickHouseSinkBuffers.forEach(IClickHouseSinkBuffer::tryAddToQueue);
            }
        };
    }
    // Checkpoint flush 前调用（暂停 cleaner）
    public synchronized void beforeFlush() {
        isFlushing = true;
    }
    // Checkpoint flush 后调用（恢复 cleaner）
    public synchronized void afterFlush() {
        isFlushing = false;
    }
}

核心设计：

• volatile boolean isFlushing： 标志位，协调 cleaner 与 checkpoint 线程。
• synchronized (this)： 保证原子性，避免并发冲突。
• 单线程执行器： 避免 cleaner 内部并发问题。

并发控制机制

问题场景：

时间轴冲突：
T1: Cleaner 线程正在执行 tryAddToQueue()
T2: Checkpoint 触发，调用 sink.flush()
T3: Cleaner 同时也在执行 tryAddToQueue()
    ├─ 可能导致：数据重复写入
    ├─ 可能导致：Buffer 清空顺序混乱
    └─ 可能导致：Future 状态不一致

解决方案：

// ClickHouseSinkManager.flush()
public void flush() {
    // 1️⃣ 暂停定时清理任务（设置标志）
    clickHouseSinkScheduledCheckerAndCleaner.beforeFlush(); // isFlushing = true
    try {
        // 2️⃣ 执行 flush（此时 cleaner 线程会跳过执行）
        clickHouseWriter.waitUntilAllFuturesDone(false, false);
    } finally {
        // 3️⃣ 恢复定时清理任务
        clickHouseSinkScheduledCheckerAndCleaner.afterFlush(); // isFlushing = false
    }
}

并发控制流程：

关键设计点：

1. volatile 保证可见性： isFlushing 使用 volatile，确保多线程间的可见性。
2. synchronized 保证原子性： getTask() 整个方法体使用 synchronized (this)。
3. 标志位协调： 通过 isFlushing 标志实现两个线程间的协调。
4. finally 确保恢复： 即使 waitUntilAllFuturesDone() 异常，也会在 finally 中恢复 cleaner。

避免的并发问题：

• 数据重复写入： Cleaner 和 Checkpoint 同时 flush。
• Buffer 状态不一致： 一边清空一边写入。
• Future 清理冲突： 正在使用的 Future 被清理。

性能影响：

• Checkpoint flush 期间，cleaner 暂停执行（通常 1-3 秒）。
• Cleaner 跳过的周期会在下次正常执行时补偿。
• 对整体吞吐影响极小（cleaner 间隔通常 30 秒）。

九、Checkpoint 语义保证

为什么 Checkpoint 时必须 Flush？

不 Flush 的后果

不Flush导致数据永久丢失

正确做法

@Override
public void snapshotState(FunctionSnapshotContext context) throws Exception {
    logger.info("start doing snapshot. flush sink to ck");
    // 1. 先 flush buffer（将内存数据写入 ClickHouse）
    if (sink != null) {
        sink.flush();
    }
    // 2. 等待所有写入完成
    if (sinkManager != null && !sinkManager.isClosed()) {
        sinkManager.flush();
    }
    // 此时 Checkpoint 才能标记为成功
    logger.info("doing snapshot. flush sink to ck");
}

Flush 实现与并发协调

public class ClickHouseSinkManager {
    public void flush() {
        //  步骤1：暂停定时清理任务
        clickHouseSinkScheduledCheckerAndCleaner.beforeFlush(); // isFlushing = true
        try {
            //  步骤2：执行 buffer flush + 等待所有写入完成
            clickHouseWriter.waitUntilAllFuturesDone(false, false);
        } finally {
            //  步骤3：恢复定时清理任务（finally 确保执行）
            clickHouseSinkScheduledCheckerAndCleaner.afterFlush(); // isFlushing = false
        }
    }
}

并发协调详解：

// cleaner 线程执行流程
synchronized (this) {
    if (isFlushing) {
        return; // Checkpoint 期间跳过本次执行
    }
    // 正常执行：清理已完成的 Future + 触发 Buffer flush
    futures.removeIf(filter);
    buffers.forEach(Buffer::tryAddToQueue);
}

关键点：

• volatile 可见性： isFlushing 使用 volatile 确保 cleaner 线程立即看到状态变化。
• synchronized互斥： getTask()方法体使用 synchronized (this) 确保原子性。
• 标志位协调： 通过 beforeFlush() / afterFlush() 管理标志位。
• finally 保证恢复： 即使 flush 异常，也会在 finally 中恢复 cleaner。

等待所有 Future 完成

public synchronized void waitUntilAllFuturesDone(boolean stopWriters, boolean clearFutures) {
    try {
        // 循环等待：直到所有 Future 完成 + 队列清空
        while (!futures.isEmpty() || unProcessedCounter.get() > 0) {
            CompletableFuture<Void> all = FutureUtil.allOf(futures);
            // 最多等待 3 分钟（与 Future 超时一致）
            all.get(3, TimeUnit.MINUTES);
            // 移除已完成的 Future（非异常）
            futures.removeIf(f -> f.isDone() && !f.isCompletedExceptionally());
            // 检查是否有异常 Future
            if (anyFutureFailed()) {
                break; // 有异常则退出
            }
        }
    } finally {
        if (stopWriters) stopWriters();
        if (clearFutures) futures.clear();
    }
}

关键逻辑：

• 循环等待直到所有 Future 完成 + 队列清空。
• 超时 3 分钟（与 Future 超时一致）。
• 移除已完成的非异常 Future。
• 有异常时退出循环。

三种 Flush 触发方式对比

Checkpoint 参数配置

// Checkpoint 配置建议
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 启用 Checkpoint（间隔 1 分钟）
env.enableCheckpointing(60000);
// Checkpoint 超时（必须大于 Future 超时 + 重试时间）
// 建议：CheckpointTimeout > FutureTimeout * MaxRetries
env.getCheckpointConfig().setCheckpointTimeout(600000); // 10 分钟
// 一致性模式
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 最小间隔（避免过于频繁）
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000); // 30 秒
// 最大并发 Checkpoint 数
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

语义保证

推荐配置：

生产环境： 使用 ExceptionsThrowableSink + Checkpoint。

允许部分丢失： 使用 UnexceptionableSink。

十、最佳实践与调优

生产配置

// ========== ClickHouse 连接参数 ==========
clickhouse.sink.target-table = tb_logs_local
clickhouse.sink.max-buffer-size = 104857600        // 批次大小
clickhouse.sink.table-count = 0                // 0 表示不分表
// ========== 写入性能参数 ==========
clickhouse.sink.num-writers = 10               // 写入线程数
clickhouse.sink.queue-max-capacity = 10        // 队列容量
clickhouse.sink.timeout-sec = 30               // flush 超时
clickhouse.sink.retries = 10                   // 最大重试次数
clickhouse.sink.check.timeout-sec = 30         // 定时检查间隔
// ========== 异常处理参数 ==========
clickhouse.sink.ignoring-clickhouse-sending-exception-enabled = false
clickhouse.sink.local-address-enabled = true   // 启用本地表写入
// ========== ClickHouse 集群配置 ==========
clickhouse.access.hosts = 192.168.1.1:8123,192.168.1.2:8123,192.168.1.3:8123
clickhouse.access.user = default
clickhouse.access.password = ***
clickhouse.access.database = dw_xx_xx
clickhouse.access.cluster = default
// ========== HikariCP 连接池配置 ==========
connectionTimeout = 30000                      // 连接超时 30s
maximumPoolSize = 20                           // 最大连接数 20
minimumIdle = 2                                // 最小空闲 2
socket_timeout = 180000                        // Socket 超时 3mi

性能调优

故障排查

十一、总结

本文深入分析了 Flink ClickHouse Sink 的实现方案，核心亮点包括：

技术亮点

• 连接池选型： 使用 HikariCP，性能优异，连接管理可靠。
• Future 超时控制： orTimeout(3min) 防止永久阻塞。
• 显式资源管理： Connection 和 PreparedStatement 显式关闭，防止连接泄漏。
• 负载均衡优化： Shuffle 初始化 + 轮询选择，避免热点。
• 异常处理增强： future.isDone() 检查，避免重复完成。
• 本地表写入： 动态节点发现 + 故障剔除，写入性能提升。
• 分片策略： 按表（应用）维度路由，独立缓冲和隔离。
• 攒批优化： 双触发机制（大小 + 超时）+ 随机抖动。
• 流量控制： 有界队列 + 线程池，实现背压。
• 健壮重试： 递归重试 + 异常节点剔除 + 最大重试限制。

Checkpoint 语义

• At-Least-Once： ExceptionsThrowableSink + Checkpoint。
• At-Most-Once： UnexceptionableSink。
• Exactly-Once： 需要配合 ClickHouse 事务（未实现）。

生产建议

1. 必须： Checkpoint 时 flush，否则会丢数据。
2. 推荐： 使用 HikariCP + 本地表写入。
3. 推荐： 配置合理的超时（Future < Socket < Checkpoint）。
4. 推荐： 监控队列大小、Future 失败率、重试次数。

该方案已在生产环境大规模验证，能够稳定支撑百万级 TPS 的日志写入场景。

往期回顾

1.服务拆分之旅：测试过程全揭秘｜得物技术

2.大模型网关：大模型时代的智能交通枢纽｜得物技术

3.从“人治”到“机治”：得物离线数仓发布流水线质量门禁实践

4.AI编程实践：从Claude Code实践到团队协作的优化思考｜得物技术

5.入选AAAI-PerFM｜得物社区推荐之基于大语言模型的新颖性推荐算法

文 /虚白

关注得物技术，每周更新技术干货

要是觉得文章对你有帮助的话，欢迎评论转发点赞～

未经得物技术许可严禁转载，否则依法追究法律责任。

导读：面对万亿级广告数据存量、日均 3 亿行增量及数千个复杂查询模板的挑战，快手广告数据平台如何突破性能瓶颈、实现架构统一与体验跃升？本文系统介绍了快手广告团队从 ClickHouse on ES 混合架构，全面迁移至 Apache Doris 的统一分析实践，最终实现查询性能提升 20～90%，写入吞吐提升 3 倍，存储效率提升 60%。

本文整理自快手高级计算引擎研发工程师 周思闽 在 Doris Summit 2025 中的演讲内容，并以演讲者第一视角进行叙述。

快手是国内日活过亿的短视频平台，其广告投放平台是商业化外部广告主与快手电商商家进行广告投放的主要阵地，支持客户在平台上进行广告物料搭建、物料管理、策略变更、数据查看等操作，这对底层数据系统的存储、计算与查询性能提出了极高要求。

要支撑如此大规模的广告投放与实时分析，底层数据架构面临巨大挑战。当前，快手的广告数据包括：由投放系统产生的物料数据以及用于数据分析的效果数据，这些数据呈现出三个显著特征：

• 数据存量巨大：广告物料累计已达千亿级别，且随业务发展正向万亿规模迈进，存储体量位居公司前列，对架构扩展性提出极高要求。
• 数据增长迅猛：仅 2025 年第一季度，日均新增广告物料数据同比激增 3.5 倍，要求底层引擎具备强大的实时写入与弹性扩展能力。
• 数据模型复杂：整个数据体系涵盖约 700 个核心字段，涉及物料、投放、用户、效果等多个维度；同时，为应对多样化分析场景，沉淀的查询模板已超 4000 个，对查询引擎的兼容性与性能均是严峻考验。

架构演进：从分散存储到统一分析

01 早期架构及挑战

早期存储架构中，物料数据由 MySQL、Elasticsearch 协同存储；效果数据主要存储与 Clickhouse 中。

数据分析时，将分散在 MySQL、Elasticsearch 中的物料数据与 ClickHouse 中的效果数据进行高效关联查询，从而为广告主提供完整、及时的投放效果洞察。

在如上所说的 ClickHouse on ES 架构中，用户提交的查询通常包含 Elasticsearch 外表（a）与 ClickHouse 内表（b）。ClickHouse 会解析查询中外表部分，将其转换为 Elasticsearch 查询语句，通过 HTTP 请求获取数据并封装为 Block，最后在引擎内部完成与内表的关联计算。

然而，随着 Elasticsearch 中数据量持续增长，该架构逐渐暴露诸多问题：

• 查询性能恶化：慢查询率上升至 35%，平均查询耗时达到 1.4 秒；
• 存储瓶颈：Elasticsearch 单分片难以支撑 10 亿级以上数据量，扩容与数据重分布成本高；
• 运维复杂度高：数据链路依赖组件多，运维与监控成本显著上升；
• 问题定位困难：缺少 ClickHouse 与 Elasticsearch 之间的全链路可观测手段，出现查询延迟、数据不一致等问题时，需跨系统排查，耗时较长。

02 选型目标及调研

基于上述问题及挑战，我们为新架构设定了明确目标：

• 慢查询率低于 5%；
• 运维排查耗时降低至分钟级；
• 支持单表万亿级别数据存储；
• 保障数据实时性，延迟低于 5 分钟。

基于以上目标，我们对 Apache Doris、ClickHouse、Elasticsearch 等主流 OLAP 引擎进行了全面的调研与性能压测。测试涵盖了写入吞吐、查询延迟、存储压缩率、全文检索性能等关键维度。

在这过程中，ClickHouse 首先被排除，因其不支持唯一键模型，而广告物料数据存在大量更新场景，要求引擎具备主键更新能力。因此，重点在 Elasticsearch 与 Apache Doris 之间进行对比。

综合测试结果，Apache Doris 在写入性能、查询效率、存储成本及运维复杂度等方面均表现优异，不仅能够满足既定架构目标，还在多个场景下显著优于 Elasticsearch。因此，我们最终选定 Apache Doris 作为下一代广告数据分析引擎。

03 基于 Apache Doris 的统一分析引擎

在实际应用中，我们引入 Apache Doris（计算引擎） 替换了原先架构中的 Elasticsearch、ClickHouse，设计了统一分析引擎 Bleem。通过在外部表模块中引入数据缓存层与元数据服务层，有效提升了跨源查询效率，使数据湖外表的查询性能接近内表水平，实现了关键的性能突破。

具体来看，Bleem 架构自下而上分为 5 层：

• 存储层：数据湖中的 Hive/Hudi 数据存储于 HDFS；存算分离模式下的内表数据存放于对象存储 BlobStore；存算一体模式下的内表数据则存储于本地磁盘。
• 缓存层：将 Hive/Hudi 外部表数据缓存至 Alluxio，保障 I/O 稳定性，提升数据读取效率。
• 计算层：Apache Doris 为核心引擎。不同项目组对应不同的 Doris 集群，以实现计算资源物理隔离，用户可按需申请计算资源。依托于 Doris 湖仓查询能力，可直接对 Doris 内表与外部 Hive/Hudi 数据查询。同时，Doris 也支持存算一体与存算分离两种部署方式，可根据实际需求灵活选择。
• 服务层：元数据缓存服务实时监听 Hive 元数据变更，并同步至缓存中，以提升湖仓外部表的查询效率。
• 接入层：将 OneSQL 作为统一查询接入网关，提供集群路由、查询改写、物化改写、查询鉴权、限流与阻断等功能。

依托 Doris 强大的 OLAP 计算与湖仓一体能力，将此前分散的数据湖分析、实时 OLAP 查询、在线报表及全文检索等多种场景，统一整合至同一套引擎架构中，实现了技术栈的收敛与提效。该架构在实际落地中已带来显著收益：

• 性能大幅提升：慢查询率低于 5%，整体查询性能提升了 20%～90%；
• 存储扩展高效：支持万亿级别数据存储，水平扩容效率较 Elasticsearch 提升 10 倍以上；
• 运维大幅简化：一套引擎覆盖全部查询场景，系统依赖组件少，运维复杂度显著降低；
• 可观测性全面加强：Doris 支持全链路追踪与全面监控，平均问题排查时间降低 80%。

迁移实践及调优经验

整个迁移过程分为三个阶段，稳步推进以确保业务平稳过渡：

• 第一阶段（试点验证）：选取关键词推广场景进行试点，跑通全量与增量数据导入流程，搭建双链路并行验证数据一致性与查询正确性。
• 第二阶段（主体迁移）：迁移原 ClickHouse on ES 查询链路，将 Elasticsearch 中全量物料数据导入 Doris，完成业务切换后下线 Elasticsearch 集群。
• 第三阶段（收尾统一）：迁移剩余纯 ClickHouse 场景，将无需关联 Elasticsearch 的查询任务及其数据全部迁移至 Doris，完成整体架构统一。

在架构升级及迁移过程中，我们收获了许多实践及优化经验，在此逐一分享。

01 解决极端场景下数据一致性问题

在数据导入层面，我们基于 SeaTunnel 实现流式数据同步，该方式支持批处理场景下的 Overwrite 语义，所有导入均采用两阶段提交机制，以确保数据同步的最终一致性。

而在基于 SeaTunnel 和 Spark 的数据同步过程中，我们遇到了极端场景下的数据重复问题。主要有两种情况：

• Spark 推测执行时，两个 Task 同时写入同一份数据并均完成 Doris 两阶段提交，尽管 Driver 只认定一个 Task 成功，但数据已重复。
• Spark Task 完成 Doris 提交后，在向 Driver 汇报前因抢占或异常退出，Driver 重启 Task 并重新写入数据。

为解决该问题，我们在 Doris 的两阶段事务提交环节引入了 ZooKeeper 分布式锁机制，通过记录并校验事务状态来保证批同步的一致性。具体流程如下：

• 准备提交阶段，先获取 ZooKeeper 临时锁，确保同一时间只有一个事务进入提交流程；
• 获取锁后，将 Prepare 状态写入 ZooKeeper 临时节点，并记录当前事务 ID；

• 查询上一个事务的状态：

  • 若不存在，直接提交当前事务；
  • 若上一事务处于 Prepare 状态，则先回滚上一事务，再提交当前事务；
  • 若上一事务已 Commit，则直接回滚当前事务；
• 最终将 Commit 状态写入 ZooKeeper 持久节点，完成本次提交。

02 Stream Load 机制优化

为应对高并发数据导入，我们对 Apache Doris 的 Stream Load 机制进行了调优。通过合理配置任务优先级与合并（Compaction）参数，显著提升了写入吞吐与稳定性。Doris 内部通过 Load Channel 进行任务调度，以区分高优与普通优先级通道。

调优的核心在于合理配置相关参数，例如当 Stream Load 任务指定的 timeout 时间小于 300 秒时，系统会将其判定为高优任务并分配至高优通道。参数优化如下：

load_task_high_priority_threshold_second=300
compaction_task_num_per_fast_disk=16
max_base_compaction_threads=8
max_cumu_compaction_threads=8

03 差异化的建表策略

OLAP 引擎的查询性能很大程度上取决于表结构设计。因此，我们针对不同业务场景制定了差异化的建表策略：

物料表（高频更新与大规模检索）：该表数据量极大且需支持实时更新。业务查询主要基于 account_id 进行过滤，而非原 MySQL 的自增 ID。为充分发挥 Doris 前缀索引与排序键的优势，在保证业务逻辑等价的前提下，我们将 account_id 与 id 组合为联合主键，并将account_id 设为首个排序键及分桶字段，大幅提升查询过滤效率。同时配置倒排索引以支持多维检索，并选用 ZSTD 压缩算法平衡存储与 IO 性能。

-- 建表语句参考
CREATE TABLE ad_core_winfo
(account_id BIGINT NOT NULL,
id BIGINT NOT NULL, 
word STRING,
INDEX idx_word (`word`) USING INVERTED...) 
UNIQUE KEY(account_id,id) 
DISTRIBUTED BY HASH(account_id) BUCKETS 1000;

效果表（多维聚合分析）： 相较于物料表，效果表侧重于数仓指标的累加与聚合。因此，我们直接采用聚合模型，并按照“天”或“小时”粒度设置分区。

-- 建表语句参考
CREATE TABLE ad_dsp_report
(__time DATETIME, 
account_id BIGINT, ...
`ad_dsp_cost` BIGINT SUM,
...) 
AGG KEY(__time,account_id,...) 
AUTO PARTITION BY RANGE(date_trunc(`__time`,'hour'))()
DISTRIBUTED BY HASH(account_id) BUCKETS 2;

04 大账户数据倾斜治理

在数据压测中，我们发现不同 Account ID 对应的数据量差异极大，小至个位数、大至百万级别，导致 BE 节点 CPU 负载严重不均。通过 SHOW DATA SKEW 命令进一步确认，Tablet 存储分布明显倾斜：大 Tablet 占用空间达 3–4 GB，小 Tablet 仅 100-200 MB，且大账户查询延迟较高。为此，我们实施了以下两点优化：

A：按账户范围进行分区

经分析，Account ID 为 5–8 位数字，且未来不会超过 10 位。因此使用 FROM_UNIXTIME 函数将 Account ID 转换为 Datetime 类型，按月对历史数据进行分区，共划分出 33 个历史分区。每个分区可容纳 2,592,000 个 Account ID，后续每新增约 200 多万个 Account ID 才会新增一个月份分区。同时，针对历史分区，根据数据存量进行手动分桶，新分区则默认设置为 256 个分桶。

该方案通过分区裁剪有效过滤了大量无关数据，同时为未来数据膨胀预留了扩展空间（物料表日均增量约 3 亿），显著降低分区增长对查询性能的影响。

B：对 Account ID 进行二次哈希

为缓解单个 Account ID 数据量差异过大导致的分布不均，我们选取与 Account ID 无关的 ID 字段，通过 ID MOD 7 计算得到一个取值在 0～6 之间的 mod 字段。将原本仅基于 account_id 的哈希分桶键调整为 (account_id, mod) 联合键，从而将同一 Account ID 的数据分散到 7 个 BE 节点上。

优化后，各 Tablet 大小基本均衡稳定在 1GB 左右，数据存储与查询负载得以在多个 BE 间均匀分布，有效解决了 此前 CPU 负载不均的问题。

05 万级分区下的查询优化

当分区数量达到万级别时，简单点查 SQL 的耗时达到 250 毫秒，远超 100 毫秒的预期。通过分析，耗时主要集中在 Plan 阶段，原因是 Doris（2.1 版本）在分区裁剪时，会遍历所有分区进行匹配，万级分区的顺序遍历开销巨大。

为此，我们将顺序遍历改为二分查找：对万级分区先进行排序，再利用二分查找快速定位目标分区，将时间复杂度从 O(n) 降至 O(log n)。优化后，该查询耗时从 250 毫秒降至 12 毫秒，性能提升超过 20 倍。目前，二分查找已在 Doris 3.1 版本中实现。

06 并发调优

在查询优化过程中，我们发现：多数查询经过条件过滤后，实际命中的数据量并不大，即便在大账户场景下，命中数据量也仅在百万级别。然而，Profile 显示这类查询的 Total Instance 数高达 800 个，其默认并发数为 32，存在明显的过度并发。

为此，我们调整以下参数降低并发开销：

set global parallel_exchange_instance_num=5;
set global parallel_pipeline_task_num=2;

调整后，同一查询的 Total Instance 数量降至 17 个，查询耗时也显著缩短。这说明在小数据量点查场景下，适当降低并发可有效减少 RPC 开销，从而降低延迟（220ms 降至 147ms）。同时，这一优化也提升了系统的整体 QPS 承载能力。

收益及规划

经过上述架构迁移与深度优化，我们在三个核心维度取得了显著收益：

• 查询性能大幅提升：关键词推广页平均查询延迟下降 64%，创意推广页延迟下降超过 90%，整体查询体验实现跨越式提升。
• 写入能力显著增强：单节点写入承载能力提升 3 倍以上，单表实时导入峰值突破 300 万行/秒。
• 存储效率优化明显：通过分区策略与 ZSTD 压缩算法，存储效率较 Elasticsearch 提升约 60%，并可轻松支撑万亿级数据存储。

未来，我们将深度探索 Apache Doris ，重点围绕两方面展开：

• 增强全文检索与分词能力：引入社区在 Doris 4.0 版本中推出的 BM25 打分功能，以及 IK 分词器等更多分词组件，实现按业务场景灵活选用最优分词方案。
• 增强向量索引：基于 Doris 4.0 版本，在内表和数据湖外表场景下对向量检索的性能和边界能力做验证与优化。

本文完。您还可以阅读来自快手另一篇实践案以及中通快递、小米集团、顺丰科技用户故事来了解湖仓分析。

很多团队在业务发展到一定阶段后，都会认真评估一次：
用户行为分析系统，是继续用现成产品，还是自己搭一套？

实际上，当企业需要埋点分析时，往往已经没有太多时间成本可投入。
业务方希望尽快看到数据结果，管理层关注投入产出比，而完全从零自建埋点系统，周期长、风险高、不可控。
因此，基于成熟开源方案快速上线，再按需求自己二开，是目前更常见、也更可控的一种选择。

这篇文章不讨论“埋点的重要性”，只做一件事：
以自建埋点分析系统为参照，给出一个成本参考，并对比基于 ClkLog 开源方案的实际投入。

完全自建一套埋点分析系统成本通常在几十万，且建设周期长、不可控因素多。
基于ClkLog开源方案搭建首期成本可控制在几万，最快一周完成部署集成，可以快速交付使用，并具备持续扩展的能力。

一、自建埋点分析系统，通常需要哪些模块？
很多团队低估了“自建”的工作量，下面只列最基础、不可回避的部分：
1.数据采集层（SDK + 埋点规范）
这个阶段往往被低估，但实际上 SDK 会长期伴随业务演进，需要持续维护。
2.数据接入与处理层
核心目标是稳定接住数据：常见技术栈包括接入服务 + Kafka / MQ。
3.数据存储层
通常会选择 ClickHouse / Doris / Druid 这类分析型数据库，同时需要设计分区、冷热数据策略。
4.复杂分析计算
这是自建中最耗精力的部分，很多团队会发现：统计不难，难的是保证分析口径正确且性能可用。
5.管理后台与可视化
这部分前端和交互成本往往被严重低估。
6.运维与长期维护
系统上线只是开始，后续还包括各项调优、异常排查等运维工作。

二、为什么很多团队不会选择「完全自建」？
问题不在“能不能做”，而在是否划算。
●早期业务验证阶段，数据系统很难直接创造业务价值
●自建系统容错成本高，试错周期长
因此，越来越多团队会选择：
在成熟的开源埋点分析系统基础上建设，而不是从零开始。

三、ClkLog开源方案能解决什么问题？
ClkLog提供了一套可直接落地的开源埋点分析方案，不依赖第三方SaaS服务。全面覆盖了埋点系统中最重、最复杂的核心能力：

1.数据采集层
支持神策SDK与自研鸿蒙SDK
2.数据接受层
进行日志数据接收与存储
3.数据处理层
进行数据处理、归档等服务
4.数据存储层
使用clickhouse进行大量数据查询
5.数据可视化
内置多种成熟分析模型，开箱即用

企业无需从零搭建底层能力，只需要围绕自身业务场景完成部署、运维和少量定制，即可形成一套可用的自有埋点分析系统。

四、基于 ClkLog，企业实际需要投入哪些成本？
1. 基础运行环境（参考）
以 ClkLog 社区版为例，在 1万日活应用规模下，采用Docker方式部署，单台服务器即可满足基础使用需求。
推荐配置参考：8核CPU；32GB内存
在常见的云厂商环境中，约1-2万/年，即可覆盖服务器、云盘、流量、备份等成本。

2. 软件部署与集成
●获取ClkLog代码（Github/Gitee）
●自行部署ClkLog服务（docker部署最快10分钟完成）
●接入埋点SDK（兼容web/小程序/iOS/安卓等）
●常规运维数据库和服务
整体实施周期短，最快一天即可完成部署并交付使用。

3. 业务层面的工作
●埋点规范梳理
●事件与指标定义
●少量业务定制
ClkLog已经内置十几种行业标准分析模型，可供业务直接开箱使用。若还有更多定制业务需要分析，可以通过自定义事件或二次开发来实现，与完全自建相比，省去的是部门团队沟通、大量底层系统设计与长期维护成本。

五、写在最后
对于大多数团队来说，先把系统跑起来、用起来、产生价值，比一开始追求完美更重要。
如果团队希望：
●完全掌控数据
●又不想长期投入基础设施研发
●把精力更多放在业务分析而不是系统本身
那么，基于成熟开源方案搭建自己的埋点分析系统，是一个性价比较高、风险更可控的选择。

做量化交易系统的后端开发，最头疼的不是策略算法（那是 Quants 的事），而是数据管道（Data Pipeline）的健壮性。

特别是处理历史 Tick 数据时，我们面临的是一个典型的“高并发写入+高精度时序”场景。在早期的架构设计中，我经常因为低估了 Tick 数据的体量和复杂性，导致系统在回放时出现“幽灵交易”——即数据到达顺序与交易所撮合顺序不一致。

工程上的三个拦路虎

时间戳的绝对真理： 在分钟线级别，这一秒和下一秒区别不大。但在 Tick 级别，毫秒级的乱序就是灾难。工程上必须严格依赖 Exchange Timestamp 而不是本地接收时间。

分页与流量控制： 也就是 Pagination。一次请求拉取全天 Tick 是不现实的，HTTP 响应体过大会导致超时或内存溢出。

异构数据源： 历史归档数据通常是冷存储结构，而实时流是 WebSocket 热数据，如何用一套代码兼容这两种接口？

高效的解决方案

为了解决这些 IO 密集型任务，我的思路是：将数据获取层（Ingestion Layer）完全解耦。

不要尝试自己在应用层去清洗原始报文。目前比较成熟的做法是直接对接第三方聚合 API。以我目前使用的 AllTick API 为例，它在服务端已经做好了清洗和标准化。这就相当于把复杂的 ETL 过程外包了出去，我们只需要通过简单的 HTTP 请求拿到 JSON 格式的结构化数据。

这样，我们的工程重心就可以从“怎么抓数据”转移到“怎么用数据”上。

代码实现：构建数据拉取器

下面是一个基于 Python requests 库构建的简单拉取器原型。注意看参数中的 limit 和时间窗口设置，这是处理大流量数据的关键：

import requests

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://api.alltick.co/v1/market/tick/history"

params = {
    "symbol": "AAPL.US",
    "market": "US",
    "start_time": "2024-01-02 09:30:00",
    "end_time": "2024-01-02 09:31:00",
    "limit": 1000
}

headers = {
    "Authorization": f"Bearer {API_KEY}"
}

resp = requests.get(BASE_URL, params=params, headers=headers)
data = resp.json()

for tick in data.get("data", []):
    ts = tick["timestamp"]
    price = tick["price"]
    volume = tick["volume"]
    print(ts, price, volume)

架构师视角的补充

在实际生产环境中，这段代码拉下来的数据，我不建议直接进 Pandas 分析，而是应该先进入消息队列（如 Kafka）或者写入 ClickHouse 这样的列式数据库。

为什么？因为历史 Tick 的价值在于高保真回放。标准化的接口解决了“源头”问题，而合理的存储架构解决了“流转”问题。这才是构建低延迟交易系统的正确姿势。