java

7.2 连接器：JDBC / Redis / Elasticsearch

Flink 侧输出连接器实践：JdbcSink 批量写入与 SQL 注入防范、用 lettuce 自建 Redis RichSinkFunction（Flink 无官方 Redis 连接器）、Elasticsearch Sink 的批量索引与退避重试，以及 RichSinkFunction 生命周期管理。

Leeting Yan 2026-11-01 4 分钟阅读 1518 字

JDBC Sink

Flink 内置 JdbcSink（在 flink-connector-jdbc 包中）：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc</artifactId>
    <version>3.3.0-1.20</version>
</dependency>

import org.apache.flink.connector.jdbc.JdbcSink;

stream.addSink(JdbcSink.sink(
    "INSERT INTO metrics (window_start, symbol, count, total) VALUES (?, ?, ?, ?)",
    (ps, metric) -> {
        ps.setTimestamp(1, new Timestamp(metric.windowStart()));
        ps.setString(2, metric.symbol());
        ps.setLong(3, metric.count());
        ps.setDouble(4, metric.total());
    },
    JdbcExecutionOptions.builder()
        .withBatchSize(1000)
        .withBatchIntervalMs(200)
        .withMaxRetries(3)
        .build(),
    new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:postgresql://db:5432/eventd")
        .withDriverName("org.postgresql.Driver")
        .withUsername("eventd")
        .withPassword("eventd")
        .build()
));

参数	说明
`batchSize`	批量写入条数
`batchIntervalMs`	批量写入间隔
`maxRetries`	失败重试次数

两点版本提醒：

org.apache.flink.connector.jdbc.JdbcSink 这个类在 connector 3.x 已标记废弃，新代码用 JdbcSink.builder()（buildAtLeastOnce / buildExactlyOnce），返回 Sink V2 对象，配 sinkTo()。上面的写法仍能编译，但属于旧接口。
DataStream.addSink() 在 1.20 也已废弃，Sink V2 统一用 sinkTo()；只有 SinkFunction（如自定义 Redis Sink）还需要 addSink。

Redis Sink（自定义）

使用 RichSinkFunction + lettuce：

import io.lettuce.core.RedisClient;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;

public class RedisMetricSink extends RichSinkFunction<Metric> {
    private transient RedisClient client;
    private transient StatefulRedisConnection<String, String> connection;
    private transient RedisAsyncCommands<String, String> async;

    @Override
    public void open(OpenContext parameters) {
        client = RedisClient.create("redis://localhost:6379");
        connection = client.connect();
        async = connection.async();
    }

    @Override
    public void invoke(Metric metric, Context context) {
        String key = "metrics:" + metric.symbol() + ":" + metric.windowStart();
        String value = metric.count() + "," + metric.total();
        async.setex(key, 300, value);  // TTL 5 分钟
    }

    @Override
    public void close() {
        if (connection != null) connection.close();
        if (client != null) client.shutdown();
    }
}

stream.addSink(new RedisMetricSink());

async.setex(...) 返回 RedisFuture，上面的代码没有等它完成——这是故意的高吞吐写法，代价是进程崩溃时在途命令可能丢失。若要求「不丢」，至少做一件事：让类额外 implements CheckpointedFunction（RichSinkFunction 本身不带快照能力），然后在 snapshotState 里调用 connection.flushCommands().toCompletableFuture().join()；或者直接退回 connection.sync()。选哪种取决于你能接受多大的重复窗口，不能既不等结果又声称 exactly-once。

Elasticsearch Sink

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-elasticsearch7</artifactId>
    <version>3.1.0-1.20</version>
</dependency>

import org.apache.flink.connector.elasticsearch.sink.ElasticsearchSink;
import org.apache.flink.connector.elasticsearch.sink.Elasticsearch7SinkBuilder;

ElasticsearchSink<Event> esSink = new Elasticsearch7SinkBuilder<Event>()
    .setHosts(new HttpHost("es", 9200))
    .setBulkFlushMaxActions(1000)
    .setBulkFlushInterval(5000)
    .setEmitter((event, context, indexer) -> {
        indexer.add(new IndexRequest("events")
            .id(event.getEventId())   // 第 7 章仍用 4.1 的 POJO Event，不是 record
            .source(JsonUtils.toJson(event), XContentType.JSON));
    })
    .build();

stream.sinkTo(esSink);

版本号同样是「连接器版本-Flink 版本」：3.1.0-1.20 表示 connector 3.1.0 适配 Flink 1.20。Elasticsearch 连接器在 Apache Flink 仓库里仍按 Flink 版本发布，但更新节奏比 Kafka/JDBC 慢，升级前先确认目标版本是否存在。

RichSinkFunction 生命周期

public class CustomSink extends RichSinkFunction<MyEvent> {
    @Override
    public void open(OpenContext openContext) {  // 1.19 起替代 open(Configuration)
        // 初始化资源（连接池、线程池）
        // 获取 RuntimeContext（当前并行度、子任务索引）
        int subtask = getRuntimeContext().getIndexOfThisSubtask();
    }

    @Override
    public void invoke(MyEvent value, Context context) {
        // 处理每条数据
    }

    @Override
    public void close() {
        // 关闭资源，必须做！
    }
}

幂等写入

Sink 可能重复写入（AT_LEAST_ONCE），下游需要幂等：

// Redis：SETEX 覆盖旧值（幂等）
async.setex(key, ttl, value);

// JDBC：INSERT ON CONFLICT UPDATE（Upsert）
"INSERT INTO metrics (id, count) VALUES (?, ?) ON CONFLICT (id) DO UPDATE SET count = EXCLUDED.count"

异步 Sink

Flink 的异步写入能力在 Sink V2 里，由 flink-connector-base 的 AsyncSinkBase / AsyncSinkWriter 提供（1.20 实际签名）：

public abstract class AsyncSinkBase<InputT, RequestEntryT extends Serializable>
        implements StatefulSink<InputT, BufferedRequestState<RequestEntryT>> {
    // 构造函数接收：ElementConverter、最大批量、最大在途请求数、
    // 最大缓冲请求数、字节上限、缓冲时间等
}

public abstract class AsyncSinkWriter<InputT, RequestEntryT extends Serializable>
        implements StatefulSink.StatefulSinkWriter<InputT, BufferedRequestState<RequestEntryT>> {
    protected void submitRequestEntries(List<RequestEntryT> entries,
                                        Consumer<List<RequestEntryT>> requestResult) { ... }
}

注意它不是一个可以简单继承的 Function：应用代码通常不直接实现它，而是使用已经封装好的连接器。如果确实要自己写异步 Redis Sink，有三个约束不能省：

在途请求数要有上限（maxInFlightRequests）。否则 Checkpoint 期间积压的 Future 会撑爆内存。
flush(true) 必须等到所有在途请求完成。否则 Checkpoint 结束后数据还可能只在内存里，恢复时对不上。
失败要让 Sink 抛错，由作业重启 + 上游重放兜底；只打日志会把「丢数据」变成静默故障。

完整的 Sink V2 实现见 7.3 自定义连接器。

eventd v3 的输出

eventd v3 的 Flink 作业输出到 Redis（面板缓存）和可选的 Kafka（下游消费）：

Flink DataStream
    ├── 自定义 RedisMetricSink（RichSinkFunction）→ 实时指标缓存（TTL 5 分钟）
    └── KafkaSink → eventd-metrics Topic（下游归档）

小结

Sink	包	特点
JDBC	`flink-connector-jdbc`	批量异步，SQL parameterized
Redis	自定义 / lettuce	异步管道，TTL 控制
Elasticsearch	`flink-connector-elasticsearch7`	批量索引，按 Flink 版本发布
自定义	`RichSinkFunction`	完全控制生命周期

下一节 7.3 连接器：自定义 Source 与 Sink 。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页