云数仓架构与弹性计算工程 - 第3节

3. 数据源优化

做实时流处理，数据源是第一道关卡。很多团队把精力都花在算子优化上，结果发现瓶颈其实在源头——Kafka 消费端。我见过不少案例，Flink 作业吞吐上不去，一看 Kafka 分区和并行度完全没对齐，白白浪费了集群资源。

说白了，Kafka 分区就是流处理的「水管粗细」。你并行度设得再高，如果分区数不够，大部分 Task 只能闲着。反过来，分区太多但并行度不够，又会造成数据倾斜。这里面的门道，咱们一个一个拆开讲。

分区数与并行度的匹配原则

先记住一个核心公式：Source 并行度 ≤ Kafka 分区数。为什么？因为一个 Kafka 分区最多只能被一个 Source Task 消费。如果你并行度设了 16，但 Topic 只有 8 个分区，那 8 个 Task 永远拿不到数据。

我个人习惯的做法是：

• 分区数 = 并行度 × (1.2 ~ 1.5)。留一点余量，方便后续扩缩容。
• 如果 Topic 已经固定了，那就让并行度等于分区数，或者略小一点。
• 千万别让并行度大于分区数——多出来的 Task 就是空转，浪费资源。

动态分区发现与扩容策略

业务量增长时，Kafka 分区数可能会动态增加。Flink 1.11 之后支持了动态分区发现，但默认是关闭的。你需要显式开启：

// Flink 1.11+ 动态分区发现
Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("flink.partition-discovery.interval-millis", "60000");

DataStream<String> stream = env
    .addSource(new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), props))
    .setParallelism(16);

这里有个坑——分区发现间隔不能设得太短。我曾经见过有人设成 5 秒，结果 Kafka 集群压力暴增，因为每次发现都要拉取元数据。建议至少 30 秒以上，生产环境我一般用 60 秒。

Rebalance 机制与分区分配策略

当分区数变化或 Task 失败重启时，会触发 Rebalance。Flink 默认使用 round-robin 分配策略，但如果你有特殊需求，可以自定义：

// 自定义分区分配器
public class MyPartitioner implements KafkaPartitioner<byte[]> {
    @Override
    public int partition(byte[] key, byte[] value, String topic, int[] partitions) {
        // 按业务 key 哈希，保证相同 key 进入同一分区
        if (key == null) return partitions[0];
        return Math.abs(Arrays.hashCode(key)) % partitions.length;
    }
}

嗯，这里要注意：自定义分配器一定要保证幂等性。我踩过坑——写了一个基于时间戳的分配器，结果时间回拨导致数据全乱了。后来改成基于业务 ID 哈希，稳得很。

Source算子的背压感知

背压（Backpressure）是流处理的核心机制。简单说，就是下游处理不过来时，向上游传递压力，让 Source 放慢速度。但很多人的背压感知是「事后诸葛亮」——等监控报警了才发现。

我个人习惯在 Source 算子中主动感知背压，而不是被动等待。怎么做？看代码：

// 自定义 Source 中感知背压
public class BackpressureAwareSource extends RichSourceFunction<String> {
    private volatile boolean running = true;
    private transient volatile boolean isBackpressured = false;

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (running) {
            // 主动检查背压状态
            if (getRuntimeContext().getMetricGroup()
                .getIOMetricGroup()
                .getBackPressuredTimeMsPerSecond()
                .getCount() > 500) {
                isBackpressured = true;
                // 放慢拉取速度
                Thread.sleep(100);
            } else {
                isBackpressured = false;
                // 正常拉取数据
                String data = pullFromKafka();
                ctx.collect(data);
            }
        }
    }
}

这段代码的核心逻辑是：每秒背压时间超过 500ms 就主动降速。为什么要主动降速？因为 Kafka 消费太快，数据会在 Flink 内部缓冲区堆积，最终导致 OOM。与其让系统崩溃，不如优雅地降速。

你想想看，如果背压时间持续超过 800ms，说明下游已经严重阻塞了。这时候光降速还不够，得考虑扩容下游算子或者优化处理逻辑。

自定义Source的调优技巧

有时候内置的 Kafka Source 满足不了需求，比如要对接 RocketMQ、Pulsar，或者要做一些特殊的数据预处理。这时候就得自己写 Source。我总结了几条实战经验：

1. 批量拉取，减少网络开销

单条拉取效率极低。我见过有人一条一条地 poll，结果网络开销占了 70% 的 CPU。正确的做法是批量拉取：

// 批量拉取示例
public class BatchSource extends RichSourceFunction<List<String>> {
    private static final int BATCH_SIZE = 1000;

    @Override
    public void run(SourceContext<List<String>> ctx) throws Exception {
        List<String> buffer = new ArrayList<>(BATCH_SIZE);
        while (running) {
            String data = pullFromExternal();
            buffer.add(data);
            if (buffer.size() >= BATCH_SIZE) {
                ctx.collect(new ArrayList<>(buffer));
                buffer.clear();
            }
        }
        // 最后一批也要发出去
        if (!buffer.isEmpty()) {
            ctx.collect(buffer);
        }
    }
}

批量大小建议设置在 500~2000 之间。太小了效果不明显，太大了延迟会变高。我一般用 1000，延迟和吞吐的平衡点。

2. 异步 IO + 连接池

如果 Source 需要调用外部服务（比如 Redis、MySQL），一定要用异步 IO。同步调用会阻塞整个 Task，吞吐直接腰斩。

// 异步 Source 示例
public class AsyncSource extends RichAsyncFunction<String, String> {
    private transient ExecutorService executor;

    @Override
    public void open(Configuration parameters) {
        // 创建连接池，大小等于并行度
        executor = Executors.newFixedThreadPool(
            getRuntimeContext().getNumberOfParallelSubtasks()
        );
    }

    @Override
    public void asyncInvoke(String input, ResultFuture<String> resultFuture) {
        executor.submit(() -> {
            String result = callExternalService(input);
            resultFuture.complete(Collections.singleton(result));
        });
    }
}

这里有个细节：连接池大小不要超过并行度。我曾经设成 100，结果把下游数据库打挂了。后来改成和并行度一致，既够用又不会造成压力。

3. 水位线（Watermark）的正确生成

自定义 Source 中，水位线生成是个容易出错的地方。很多人直接在 Source 里生成水位线，但忽略了乱序数据的影响。

// 正确的水位线生成方式
public class WatermarkSource extends RichSourceFunction<Event> {
    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        // 使用周期性水位线生成器
        ctx.collectWithTimestamp(event, event.getTimestamp());
        ctx.emitWatermark(new Watermark(event.getTimestamp() - 5000));
    }
}

为什么要减 5000？这是允许 5 秒的乱序。如果业务对延迟要求高，可以减 1000 甚至 500。但别设成 0——现实世界的数据没有完全有序的。

4. 状态管理与 Checkpoint 集成

自定义 Source 一定要实现 CheckpointedFunction，否则故障恢复时会重复消费或丢失数据。

// Checkpoint 集成示例
public class CheckpointedSource extends RichSourceFunction<String>
    implements CheckpointedFunction {

    private transient ListState<Long> offsetState;
    private long currentOffset = 0L;

    @Override
    public void initializeState(FunctionInitializationContext context) {
        offsetState = context.getOperatorStateStore()
            .getListState(new ListStateDescriptor<>("offset", Long.class));
        // 恢复偏移量
        if (context.isRestored()) {
            for (Long offset : offsetState.get()) {
                currentOffset = offset;
            }
        }
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) {
        offsetState.clear();
        offsetState.add(currentOffset);
    }
}

这里要注意：状态不要存太大。我见过有人把整个数据批次都存到状态里，Checkpoint 直接超时。只存偏移量或者元数据就够了。

知识体系总览

下面这张图把本章的核心逻辑串起来了。你可以看到，数据源优化其实是一个从「分区对齐」到「背压感知」再到「自定义调优」的递进过程。

数据源优化是流处理性能调优的「第一公里」。分区没对齐，后面再怎么优化都是白搭。背压感知让你能提前发现问题，而不是等报警了才手忙脚乱。自定义 Source 的调优技巧，则是应对复杂业务场景的「杀手锏」。

嗯，这一章的内容就到这里。记住一句话：源头稳了，下游才能跑得快。