一、背景与需求分析

在蓝12（泛指高并发、低延迟的实时数据处理场景）中，企业需要构建从数据源到分析引擎的高效管道。以电商场景为例，用户行为数据（如点击、加购、支付）需实时写入ClickHouse进行OLAP分析，同时保证Flink作业的稳定性和吞吐量。传统方案中，直接使用JDBC Sink可能面临连接池耗尽、批量提交策略不合理等问题，而自定义Source/Sink能更好地适配业务需求。

二、Flink与ClickHouse集成架构设计

1. 核心组件选型

Flink版本：推荐1.16+（支持更完善的DataStream API和状态管理）
ClickHouse客户端：使用clickhouse-jdbc（0.3.x版本兼容性更佳）
连接池：HikariCP（配置maxPoolSize=20, minimumIdle=5）

2. 数据流向设计

Kafka Topic（用户行为原始数据）
   ↓
Flink Source（反序列化+字段映射）
   ↓
Flink Processing（窗口聚合/过滤）
   ↓
ClickHouse Sink（批量写入+异常重试）
   ↓
ClickHouse集群（分布式表引擎）

三、自定义ClickHouse Source实现

1. 基于Kafka的SourceFunction开发

public class ClickHouseKafkaSource extends SourceFunction<UserBehavior> {
    private volatile boolean isRunning = true;
    private final KafkaConsumer<String, String> consumer;
    public ClickHouseKafkaSource(String bootstrapServers, String topic) {
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("group.id", "flink-clickhouse-group");
        this.consumer = new KafkaConsumer<>(props, new StringDeserializer(), new StringDeserializer());
        consumer.subscribe(Collections.singletonList(topic));
    }
    @Override
    public void run(SourceContext<UserBehavior> ctx) {
        try {
            while (isRunning) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    UserBehavior behavior = JSON.parseObject(record.value(), UserBehavior.class);
                    ctx.collectWithTimestamp(behavior, behavior.getTimestamp());
                    ctx.emitWatermark(new Watermark(behavior.getTimestamp() - 1));
                }
            }
        } finally {
            consumer.close();
        }
    }
    @Override
    public void cancel() {
        isRunning = false;
    }
}

关键点：

水印生成策略需与事件时间对齐
反序列化异常需捕获并记录（避免作业失败）
偏移量提交由Flink Checkpoint机制管理

2. 性能优化实践

并行度调整：根据Kafka分区数设置Source并行度（建议1:1映射）
反序列化优化：使用FastJSON/Gson的线程安全实例
背压处理：通过env.setBufferTimeout(100)控制缓冲区大小

四、自定义ClickHouse Sink实现

1. 批量写入SinkFunction开发

public class ClickHouseBatchSink extends RichSinkFunction<UserBehavior> {
    private transient Connection connection;
    private transient PreparedStatement insertStmt;
    private final int batchSize;
    private final List<UserBehavior> buffer = new ArrayList<>();
    public ClickHouseBatchSink(String jdbcUrl, int batchSize) {
        this.batchSize = batchSize;
    }
    @Override
    public void open(Configuration parameters) {
        try {
            connection = DriverManager.getConnection(
                jdbcUrl + "&socket_timeout=300000", 
                "default", 
                ""
            );
            insertStmt = connection.prepareStatement(
                "INSERT INTO user_behavior VALUES (?, ?, ?, ?)"
            );
        } catch (SQLException e) {
            throw new RuntimeException("Failed to initialize ClickHouse connection", e);
        }
    }
    @Override
    public void invoke(UserBehavior behavior, Context context) {
        try {
            insertStmt.setLong(1, behavior.getUserId());
            insertStmt.setString(2, behavior.getAction());
            insertStmt.setTimestamp(3, new Timestamp(behavior.getTimestamp()));
            insertStmt.setString(4, behavior.getDevice());
            insertStmt.addBatch();
            buffer.add(behavior);
            if (buffer.size() >= batchSize) {
                flush();
            }
        } catch (SQLException e) {
            throw new RuntimeException("Batch insert failed", e);
        }
    }
    private void flush() throws SQLException {
        insertStmt.executeBatch();
        connection.commit();
        buffer.clear();
    }
    @Override
    public void close() {
        try {
            if (!buffer.isEmpty()) {
                flush();
            }
            insertStmt.close();
            connection.close();
        } catch (SQLException e) {
            log.error("Close resources failed", e);
        }
    }
}

2. 高级特性实现

异步写入优化

// 使用CompletableFuture实现异步写入
private ExecutorService executor = Executors.newFixedThreadPool(4);
public void asyncInvoke(UserBehavior behavior) {
    executor.submit(() -> {
        try (Connection conn = getConnection()) {
            // 执行写入逻辑
        } catch (Exception e) {
            // 异常处理
        }
    });
}

故障恢复机制

重试策略：指数退避重试（最大3次）
死信队列：将失败数据写入Kafka备用Topic
监控告警：通过Prometheus暴露写入延迟指标

五、蓝12场景下的专项优化

1. 低延迟配置

Flink参数：

taskmanager.numberOfTaskSlots: 4
parallelism.default: 8
execution.checkpointing.interval: 10s

ClickHouse参数：

<!-- config.xml -->
<max_connections>1000</max_connections>
<background_pool_size>32</background_pool_size>

2. 高吞吐优化

批量大小调整：通过压测确定最佳batchSize（通常500-2000条/批）
压缩传输：启用compress=true参数
并行写入：对ClickHouse分布式表使用ON CLUSTER语法

六、完整作业示例

public class FlinkClickHousePipeline {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.setParallelism(8);
        // 1. 创建Source
        DataStream<UserBehavior> source = env.addSource(
            new ClickHouseKafkaSource("kafka:9092", "user_behavior")
        ).name("Kafka Source");
        // 2. 数据处理
        SingleOutputStreamOperator<UserBehavior> processed = source
            .filter(b -> b.getAction().equals("purchase"))
            .keyBy(UserBehavior::getUserId)
            .window(TumblingEventTimeWindows.of(Time.minutes(5)))
            .aggregate(new PurchaseCountAggregator());
        // 3. 创建Sink
        processed.addSink(new ClickHouseBatchSink(
            "jdbc:clickhouse://ch-server:8123/default", 
            1000
        )).name("ClickHouse Sink");
        env.execute("Flink to ClickHouse Pipeline");
    }
}

七、常见问题解决方案

1. 连接泄漏问题

现象：Too many connections错误
解决：
- 在Sink中实现try-with-resources
- 设置连接池最大生命周期（maxLifetime=1800000）

2. 数据倾斜处理

方案：
- 对keyBy字段添加盐值（userId % 10）
- 使用rebalance()算子重新分配

3. 版本兼容性

ClickHouse JDBC：
- 21.x版本使用ru.yandex.clickhouse:clickhouse-jdbc
- 22.x+版本使用com.clickhouse:clickhouse-jdbc

八、最佳实践总结

监控体系：
- Flink Metrics（NumRecordsIn/Out）
- ClickHouse System Tables（metrics, queries）
压测方法：
- 使用flink-benchmark工具模拟生产流量
- 逐步增加负载至TPS稳定
升级策略：
- 先升级Sink端ClickHouse集群
- 再升级Flink作业（避免二进制不兼容）

通过上述实践，在蓝12场景下可实现：

端到端延迟<500ms（P99）
吞吐量达20万条/秒（3节点ClickHouse集群）
作业恢复时间<1分钟（基于Checkpoints）

蓝12场景下Flink与ClickHouse的Source-Sink集成实践