登录/ 注册
 找回密码
 立即注册
关闭

CSDN热搜

程序园 精品问答 技术交流 资源下载
返回列表 发新帖
首页 › 业界区 › 业界 ›从零开始学Flink：数据输出的终极指南

从零开始学Flink：数据输出的终极指南

锷稠 前天 11:21
在实时数据处理的完整链路中，数据输出（Sink）是最后一个关键环节，它负责将处理后的结果传递到外部系统供后续使用。Flink提供了丰富的数据输出连接器，支持将数据写入Kafka、Elasticsearch、文件系统、数据库等各种目标系统。本文将深入探讨Flink数据输出的核心概念、配置方法和最佳实践，并基于Flink 1.20.1构建一个完整的数据输出案例。
一、Flink Sink概述

1. 什么是Sink

Sink（接收器）是Flink数据处理流水线的末端，负责将计算结果输出到外部存储系统或下游处理系统。在Flink的编程模型中，Sink是DataStream API中的一个转换操作，它接收DataStream并将数据写入指定的外部系统。
2. Sink的分类

Flink的Sink连接器可以分为以下几类：

  • 内置Sink：如print()、printToErr()等用于调试的内置输出
  • 文件系统Sink：支持写入本地文件系统、HDFS等
  • 消息队列Sink：如Kafka、RabbitMQ等
  • 数据库Sink：如JDBC、Elasticsearch等
  • 自定义Sink：通过实现SinkFunction接口自定义输出逻辑
3. 输出语义保证

Flink为Sink提供了三种输出语义保证：

  • 最多一次（At-most-once）：数据可能丢失，但不会重复
  • 至少一次（At-least-once）：数据不会丢失，但可能重复
  • 精确一次（Exactly-once）：数据既不会丢失，也不会重复
这些语义保证与Flink的检查点（Checkpoint）机制密切相关，我们将在后面详细讨论。
二、环境准备与依赖配置

1. 版本说明


  • Flink：1.20.1
  • JDK：17+
  • Gradle：8.3+
  • 外部系统：Kafka 3.4.0、Elasticsearch 7.17.0、MySQL 8.0
2. 核心依赖
  1. dependencies {
  2.     // Flink核心依赖
  3.     implementation 'org.apache.flink:flink_core:1.20.1'
  4.     implementation 'org.apache.flink:flink-streaming-java:1.20.1'
  5.     implementation 'org.apache.flink:flink-clients:1.20.1'
  6.    
  7.     // Kafka Connector
  8.     implementation 'org.apache.flink:flink-connector-kafka:3.4.0-1.20'
  9.    
  10.     // Elasticsearch Connector
  11.     implementation 'org.apache.flink:flink-connector-elasticsearch7:3.1.0-1.20'
  12.    
  13.     // JDBC Connector
  14.     implementation 'org.apache.flink:flink-connector-jdbc:3.3.0-1.20'
  15.     implementation 'mysql:mysql-connector-java:8.0.33'
  16.    
  17.     // FileSystem Connector
  18.     implementation 'org.apache.flink:flink-connector-files:1.20.1'
  20. }
复制代码
三、基础Sink操作

1. 内置调试Sink

Flink提供了一些内置的Sink用于开发和调试阶段：
  1. import org.apache.flink.streaming.api.datastream.DataStream;
  2. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  4. public class BasicSinkDemo {
  5.     public static void main(String[] args) throws Exception {
  6.         // 创建执行环境
  7.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  8.         
  9.         // 创建数据源
  10.         DataStream<String> stream = env.fromElements("Hello", "Flink", "Sink");
  11.         
  12.         // 打印到标准输出
  13.         stream.print("StandardOutput");
  14.         
  15.         // 打印到标准错误输出
  16.         stream.printToErr("ErrorOutput");
  17.         
  18.         // 执行作业
  19.         env.execute("Basic Sink Demo");
  20.     }
  21. }
复制代码
2. 文件系统Sink

Flink支持将数据写入本地文件系统、HDFS等。下面是一个写入本地文件系统的示例：
  1. package com.cn.daimajiangxin.flink.sink;
  3. import org.apache.flink.api.common.serialization.SimpleStringEncoder;
  4. import org.apache.flink.configuration.MemorySize;
  5. import org.apache.flink.connector.file.sink.FileSink;
  6. import org.apache.flink.core.fs.Path;
  7. import org.apache.flink.streaming.api.datastream.DataStream;
  8. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  9. import org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy;
  10. import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
  12. import java.time.Duration;
  14. public class FileSystemSinkDemo {
  15.     public static void main(String[] args) throws Exception {
  16.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  17.         DataStream<Object> stream = env.fromData("Hello", "Flink", "FileSystem", "Sink");
  18.         RollingPolicy<Object, String> rollingPolicy = DefaultRollingPolicy.<Object, String>builder()
  19.                 .withRolloverInterval(Duration.ofMinutes(15))
  20.                 .withInactivityInterval(Duration.ofMinutes(5))
  21.                 .withMaxPartSize(MemorySize.ofMebiBytes(64))
  22.                 .build();
  24.         // 创建文件系统Sink
  25.         FileSink<Object> sink = FileSink
  26.                 .forRowFormat(new Path("file:///tmp/flink-output"), new SimpleStringEncoder<>())
  27.                 .withRollingPolicy(rollingPolicy)
  28.                 .build();
  29.         // 添加Sink
  30.         stream.sinkTo(sink);
  31.         env.execute("File System Sink Demo");
  32.     }
  33. }
复制代码
四、高级Sink连接器

1. Kafka Sink

Kafka是实时数据处理中常用的消息队列，Flink提供了强大的Kafka Sink支持：
  1. import org.apache.flink.api.common.serialization.SimpleStringSchema;
  2. import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
  3. import org.apache.flink.connector.kafka.sink.KafkaSink;
  4. import org.apache.flink.streaming.api.datastream.DataStream;
  5. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import java.util.Properties;
  9. public class KafkaSinkDemo {
  10.     public static void main(String[] args) throws Exception {
  11.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  12.         
  13.         // 开启检查点以支持Exactly-Once语义
  14.         env.enableCheckpointing(5000);
  15.         
  16.         DataStream<String> stream = env.fromElements("Hello Kafka", "Flink to Kafka", "Data Pipeline");
  17.         
  18.         // Kafka配置
  19.         Properties props = new Properties();
  20.         props.setProperty("bootstrap.servers", "localhost:9092");
  21.         
  22.         // 创建Kafka Sink
  23.         KafkaSink<String> sink = KafkaSink.<String>
  24.                 builder()
  25.                 .setKafkaProducerConfig(props)
  26.                 .setRecordSerializer(KafkaRecordSerializationSchema.builder()
  27.                         .setTopic("flink-output-topic")
  28.                         .setValueSerializationSchema(new SimpleStringSchema())
  29.                         .build())
  30.                 .build();
  31.         
  32.         // 添加Sink
  33.         stream.sinkTo(sink);
  34.         
  35.         env.execute("Kafka Sink Demo");
  36.     }
  37. }
复制代码
kafka消息队列消息：
1.png

2. Elasticsearch Sink

Elasticsearch是一个实时的分布式搜索和分析引擎，非常适合存储和查询Flink处理的实时数据：
  1. package com.cn.daimajiangxin.flink.sink;
  3. import com.fasterxml.jackson.databind.ObjectMapper;
  4. import org.apache.flink.connector.elasticsearch.sink.Elasticsearch7SinkBuilder;
  5. import org.apache.flink.connector.elasticsearch.sink.ElasticsearchSink;
  6. import org.apache.flink.streaming.api.datastream.DataStream;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. import org.apache.http.HttpHost;
  9. import org.elasticsearch.action.index.IndexRequest;
  10. import org.elasticsearch.client.Requests;
  12. import java.util.Map;
  14. public class ElasticsearchSinkDemo {
  15.     private static final ObjectMapper objectMapper = new ObjectMapper();
  16.     public static void main(String[] args) throws Exception {
  17.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  18.         env.enableCheckpointing(5000);
  21.         DataStream<String> stream = env.fromData(
  22.                 "{"id":"1","name":"Flink","category":"framework"}",
  23.                 "{"id":"2","name":"Elasticsearch","category":"database"}");
  25.         // 配置Elasticsearch节点
  26.         HttpHost httpHost=new HttpHost("localhost", 9200, "http");
  28.         // 创建Elasticsearch Sink
  29.         ElasticsearchSink<String> sink=new Elasticsearch7SinkBuilder<String>()
  30.                 .setBulkFlushMaxActions(10)        // 批量操作数量
  31.                 .setBulkFlushInterval(5000)          // 批量刷新间隔（毫秒）
  32.                 .setHosts(httpHost)
  33.                 .setConnectionRequestTimeout(60000)  // 连接请求超时时间
  34.                 .setConnectionTimeout(60000)         // 连接超时时间
  35.                 .setSocketTimeout(60000)             // Socket 超时时间
  36.                 .setEmitter((element, context, indexer) -> {
  37.                     try {
  38.                         Map<String, Object> json = objectMapper.readValue(element, Map.class);
  39.                         IndexRequest request = Requests.indexRequest()
  40.                                 .index("flink_documents")
  41.                                 .id((String) json.get("id"))
  42.                                 .source(json);
  43.                         indexer.add(request);
  44.                     } catch (Exception e) {
  45.                         // 处理解析异常
  46.                         System.err.println("Failed to parse JSON: " + element);
  47.                     }
  48.                 })
  49.                 .build();
  51.         // 添加Sink
  52.         stream.sinkTo(sink);
  54.         env.execute("Elasticsearch Sink Demo");
  55.     }
  56. }
复制代码
使用post工具查看数据
2.png

3. JDBC Sink

使用JDBC Sink可以将数据写入各种关系型数据库：
  1. package com.cn.daimajiangxin.flink.sink;
  3. import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
  4. import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
  5. import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
  6. import org.apache.flink.connector.jdbc.core.datastream.Jdbc;
  7. import org.apache.flink.connector.jdbc.core.datastream.sink.JdbcSink;
  8. import org.apache.flink.connector.jdbc.datasource.statements.SimpleJdbcQueryStatement;
  9. import org.apache.flink.streaming.api.datastream.DataStream;
  10. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  12. import java.util.Arrays;
  13. import java.util.List;
  15. public class JdbcSinkDemo {
  16.     public static void main(String[] args) throws Exception {
  17.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  18.         env.enableCheckpointing(5000);
  19.         List<User> userList = Arrays.asList(     new User(1, "Alice", 25,"alice"),
  20.                 new User(2, "Bob", 30,"bob"),
  21.                 new User(3, "Charlie", 35,"charlie"));
  22.         // 模拟用户数据
  23.         DataStream<User> userStream = env.fromData(userList);
  25.         JdbcExecutionOptions jdbcExecutionOptions = JdbcExecutionOptions.builder()
  26.                 .withBatchSize(1000)
  27.                 .withBatchIntervalMs(200)
  28.                 .withMaxRetries(5)
  29.                 .build();
  30.         JdbcConnectionOptions connectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
  31.                 .withUrl("jdbc:mysql://localhost:3306/test")
  32.                 .withDriverName("com.mysql.cj.jdbc.Driver")
  33.                 .withUsername("username")
  34.                 .withPassword("password")
  35.                 .build();
  36.         String insertSql = "INSERT INTO user (id, name, age, user_name) VALUES (?, ?, ?, ?)";
  37.         JdbcStatementBuilder<User> statementBuilder = (statement, user) -> {
  38.             statement.setInt(1, user.getId());
  39.             statement.setString(2, user.getName());
  40.             statement.setInt(3, user.getAge());
  41.             statement.setString(4, user.getUserName());
  42.         };
  43.         // 创建JDBC Sink
  45.         JdbcSink<User> jdbcSink = new Jdbc().<User>sinkBuilder()
  46.                 .withQueryStatement( new SimpleJdbcQueryStatement<User>(insertSql,statementBuilder))
  47.                 .withExecutionOptions(jdbcExecutionOptions)
  48.                 .buildAtLeastOnce(connectionOptions);
  49.         // 添加Sink
  50.         userStream.sinkTo(jdbcSink);
  51.         env.execute("JDBC Sink Demo");
  52.     }
  54.     // 用户实体类
  55.     public static class User {
  56.         private int id;
  57.         private String name;
  58.         private String userName;
  59.         private int age;
  61.         public User(int id, String name, int age,String userName) {
  62.             this.id = id;
  63.             this.name = name;
  64.             this.age = age;
  65.             this.userName=userName;
  66.         }
  68.         public int getId() {
  69.             return id;
  70.         }
  72.         public String getName() {
  73.             return name;
  74.         }
  76.         public int getAge() {
  77.             return age;
  78.         }
  80.         public String getUserName() {
  81.             return userName;
  82.         }
  83.     }
  84. }
复制代码
登录mysql客户端查看数据
3.png

五、Sink的可靠性保证机制

1. 检查点与保存点

Flink的检查点（Checkpoint）机制是实现精确一次语义的基础。当开启检查点后，Flink会定期将作业的状态保存到持久化存储中。如果作业失败，Flink可以从最近的检查点恢复，确保数据不会丢失。
  1. // 配置检查点
  2. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  4. // 启用检查点，间隔5000ms
  5. env.enableCheckpointing(5000);
  7. // 配置检查点模式为EXACTLY_ONCE（默认）
  8. env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
  10. // 设置检查点超时时间
  11. env.getCheckpointConfig().setCheckpointTimeout(60000);
  13. // 设置最大并行检查点数量
  14. env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
  16. // 开启外部化检查点，作业失败时保留检查点
  17. env.getCheckpointConfig().enableExternalizedCheckpoints(
  18.     CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
复制代码
2. 事务与二阶段提交

对于支持事务的外部系统，Flink使用二阶段提交（Two-Phase Commit）协议来实现精确一次语义：

  • 第一阶段（预提交）：Flink将数据写入外部系统的预提交区域，但不提交
  • 第二阶段（提交）：所有算子完成预提交后，Flink通知外部系统提交数据
这种机制确保了即使在作业失败或恢复的情况下，数据也不会被重复写入或丢失。
3. 不同Sink的语义保证级别

不同的Sink连接器支持不同级别的语义保证：

  • 支持精确一次（Exactly-once）：Kafka、Elasticsearch（版本支持）、文件系统（预写日志模式）
  • 支持至少一次（At-least-once）：JDBC、Redis、RabbitMQ
  • 最多一次（At-most-once）：简单的无状态输出
六、自定义Sink实现

当Flink内置的Sink连接器不能满足需求时，我们可以通过实现SinkFunction接口来自定义Sink：
  1. package com.cn.daimajiangxin.flink.sink;
  3. import org.apache.flink.api.common.functions.RuntimeContext;
  4. import org.apache.flink.api.connector.sink2.Sink;
  5. import org.apache.flink.api.connector.sink2.SinkWriter;
  6. import org.apache.flink.streaming.api.datastream.DataStream;
  7. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  8. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
  10. import java.io.IOException;
  12. public class CustomSinkDemo {
  13.     public static void main(String[] args) throws Exception {
  14.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  16.         DataStream<String> stream = env.fromElements("Custom", "Sink", "Example");
  18.         // 使用自定义Sink
  19.         stream.sinkTo(new CustomSink());
  21.         env.execute("Custom Sink Demo");
  22.     }
  24.     // 自定义Sink实现 - 使用新API
  25.     public static class CustomSink implements Sink<String> {
  27.         @Override
  28.         public SinkWriter<String> createWriter(InitContext context) {
  29.             return new CustomSinkWriter();
  30.         }
  32.         // SinkWriter负责实际的数据写入逻辑
  33.         private static class CustomSinkWriter implements SinkWriter<String> {
  35.             // 初始化资源
  36.             public CustomSinkWriter() {
  37.                 // 初始化连接、客户端等资源
  38.                 System.out.println("CustomSink initialized");
  39.             }
  41.             // 处理每个元素
  42.             @Override
  43.             public void write(String value, Context context)  throws IOException, InterruptedException {
  44.                 // 实际的写入逻辑
  45.                 System.out.println("Writing to custom sink: " + value);
  46.             }
  48.             // 刷新缓冲区
  49.             @Override
  50.             public void flush(boolean endOfInput) {
  51.                 // 刷新逻辑（如果需要）
  52.             }
  54.             // 清理资源
  55.             @Override
  56.             public void close() throws Exception {
  57.                 // 关闭连接、客户端等资源
  58.                 System.out.println("CustomSink closed");
  59.             }
  60.         }
  61.     }
  63. }
复制代码
4.png

七、实战案例：实时数据处理流水线

下面我们将构建一个完整的实时数据处理流水线，从Kafka读取数据，进行转换处理，然后输出到多个目标系统：
1. 系统架构
  1. Kafka Source -> Flink Processing -> Multiple Sinks
  2.                                |-> Kafka Sink
  3.                                |-> Elasticsearch Sink
  4.                                |-> JDBC Sink
复制代码
2. 数据模型

我们将使用日志数据模型，定义一个LogEntry类来表示日志条目：
  1. package com.cn.daimajiangxin.flink.sink;
  3. public class LogEntry {
  4.     private String timestamp;
  5.     private String logLevel;
  6.     private String source;
  7.     private String message;
  9.     public String getTimestamp() {
  10.         return timestamp;
  11.     }
  13.     public void setTimestamp(String timestamp) {
  14.         this.timestamp = timestamp;
  15.     }
  17.     public String getLogLevel() {
  18.         return logLevel;
  19.     }
  21.     public void setLogLevel(String logLevel) {
  22.         this.logLevel = logLevel;
  23.     }
  25.     public String getSource() {
  26.         return source;
  27.     }
  29.     public void setSource(String source) {
  30.         this.source = source;
  31.     }
  33.     public String getMessage() {
  34.         return message;
  35.     }
  37.     public void setMessage(String message) {
  38.         this.message = message;
  39.     }
  41.     @Override
  42.     public String toString() {
  43.         return String.format("LogEntry{timestamp='%s', logLevel='%s', source='%s', message='%s'}",
  44.                 timestamp, logLevel, source, message);
  45.     }
  46. }
复制代码
定义一个日志统计实体类LogStats，用于表示每个源的日志统计信息：
  1. package com.cn.daimajiangxin.flink.sink;
  3. public class LogStats {
  4.     private String source;
  5.     private long count;
  7.     public LogStats() {
  8.     }
  10.     public LogStats(String source, long count) {
  11.         this.source = source;
  12.         this.count = count;
  13.     }
  15.     public String getSource() {
  16.         return source;
  17.     }
  19.     public void setSource(String source) {
  20.         this.source = source;
  21.     }
  23.     public long getCount() {
  24.         return count;
  25.     }
  27.     public void setCount(long count) {
  28.         this.count = count;
  29.     }
  31.     @Override
  32.     public String toString() {
  33.         return String.format("LogStats{source='%s', count=%d}", source, count);
  34.     }
  35. }
复制代码
3. 完整实现代码
  1. package com.cn.daimajiangxin.flink.sink;
  3. import org.apache.flink.api.common.eventtime.WatermarkStrategy;
  4. import org.apache.flink.api.common.serialization.SimpleStringSchema;
  5. import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
  6. import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
  7. import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
  8. import org.apache.flink.connector.jdbc.core.datastream.Jdbc;
  9. import org.apache.flink.connector.jdbc.core.datastream.sink.JdbcSink;
  10. import org.apache.flink.connector.jdbc.datasource.statements.SimpleJdbcQueryStatement;
  11. import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
  12. import org.apache.flink.connector.kafka.sink.KafkaSink;
  13. import org.apache.flink.connector.kafka.source.KafkaSource;
  14. import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
  15. import org.apache.flink.streaming.api.datastream.DataStream;
  16. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  17. import org.apache.flink.connector.elasticsearch.sink.Elasticsearch7SinkBuilder;
  18. import org.apache.flink.connector.elasticsearch.sink.ElasticsearchSink;
  19. import org.apache.http.HttpHost;
  20. import org.elasticsearch.action.index.IndexRequest;
  21. import org.elasticsearch.client.Requests;
  23. import java.sql.PreparedStatement;
  24. import java.time.LocalDateTime;
  25. import java.util.ArrayList;
  26. import java.util.HashMap;
  27. import java.util.List;
  28. import java.util.Map;
  29. import java.util.Properties;
  31. public class MultiSinkPipeline {
  32.     public static void main(String[] args) throws Exception {
  33.         // 1. 创建执行环境并配置检查点
  34.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  35.         env.enableCheckpointing(5000);
  37.         // 2. 创建Kafka Source
  38.         KafkaSource<String> source = KafkaSource.<String>
  39.                         builder()
  40.                 .setBootstrapServers("localhost:9092")
  41.                 .setTopics("logs-input-topic")
  42.                 .setGroupId("flink-group")
  43.                 .setStartingOffsets(OffsetsInitializer.earliest())
  44.                 .setValueOnlyDeserializer(new SimpleStringSchema())
  45.                 .build();
  47.         // 3. 读取数据并解析
  48.         DataStream<String> kafkaStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
  50.         // 解析日志数据
  51.         DataStream<LogEntry> logStream = kafkaStream
  52.                 .map(line -> {
  53.                     String[] parts = line.split("\\|");
  54.                     return new LogEntry(parts[0], parts[1], parts[2], parts[3]);
  55.                 })
  56.                 .name("Log Parser");
  58.         // 4. 过滤错误日志
  59.         DataStream<LogEntry> errorLogStream = logStream
  60.                 .filter(log -> "ERROR".equals(log.getLogLevel()))
  61.                 .name("Error Log Filter");
  63.         // 5. 配置并添加Kafka Sink - 输出错误日志
  64.         // Kafka配置
  65.         Properties props = new Properties();
  66.         props.setProperty("bootstrap.servers", "localhost:9092");
  68.         // 创建Kafka Sink
  69.         KafkaSink<LogEntry> kafkaSink = KafkaSink.<LogEntry>builder()
  70.                 .setKafkaProducerConfig(props)
  71.                 .setRecordSerializer(KafkaRecordSerializationSchema.<LogEntry>builder()
  72.                         .setTopic("error-logs-topic")
  73.                         .setValueSerializationSchema(element -> element.toString().getBytes())
  74.                         .build())
  75.                 .build();
  77.         errorLogStream.sinkTo(kafkaSink).name("Error Logs Kafka Sink");
  79.         // 6. 配置并添加Elasticsearch Sink - 存储所有日志
  80.         // 配置Elasticsearch节点
  81.         HttpHost httpHost=new HttpHost("localhost", 9200, "http");
  83.         ElasticsearchSink<LogEntry> esSink = new Elasticsearch7SinkBuilder<LogEntry>()
  84.                 .setBulkFlushMaxActions(10)        // 批量操作数量
  85.                 .setBulkFlushInterval(5000)          // 批量刷新间隔（毫秒）
  86.                 .setHosts(httpHost)
  87.                 .setConnectionRequestTimeout(60000)  // 连接请求超时时间
  88.                 .setConnectionTimeout(60000)         // 连接超时时间
  89.                 .setSocketTimeout(60000)             // Socket 超时时间
  90.                 .setEmitter((element, context, indexer) -> {
  91.                     Map<String, Object> json = new HashMap<>();
  92.                     json.put("timestamp", element.getTimestamp());
  93.                     json.put("logLevel", element.getLogLevel());
  94.                     json.put("source", element.getSource());
  95.                     json.put("message", element.getMessage());
  96.                     IndexRequest request = Requests.indexRequest()
  97.                             .index("logs_index")
  98.                             .source(json);
  99.                     indexer.add(request);
  100.                 })
  101.                 .build();
  103.         logStream.sinkTo(esSink).name("Elasticsearch Sink");
  105.         // 7. 配置并添加JDBC Sink - 存储错误日志统计
  106.         // 先进行统计
  107.         DataStream<LogStats> statsStream = errorLogStream
  108.                 .map(log -> new LogStats(log.getSource(), 1))
  109.                 .keyBy(LogStats::getSource)
  110.                 .sum("count")
  111.                 .name("Error Log Stats");
  112.         JdbcExecutionOptions jdbcExecutionOptions = JdbcExecutionOptions.builder()
  113.                 .withBatchSize(1000)
  114.                 .withBatchIntervalMs(200)
  115.                 .withMaxRetries(5)
  116.                 .build();
  117.         JdbcConnectionOptions connectionOptions = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
  118.                 .withUrl("jdbc:mysql://localhost:3306/test")
  119.                 .withDriverName("com.mysql.cj.jdbc.Driver")
  120.                 .withUsername("mysql用户名")
  121.                 .withPassword("mysql密码")
  122.                 .build();
  123.         String insertSql = "INSERT INTO error_log_stats (source, count, last_updated) VALUES (?, ?, ?) " +
  124.                "ON DUPLICATE KEY UPDATE count = count + VALUES(count), last_updated = VALUES(last_updated)";
  125.         JdbcStatementBuilder<LogStats> statementBuilder = (statement, stats) -> {
  126.             statement.setString(1, stats.getSource());
  127.             statement.setLong(2, stats.getCount());
  128.             statement.setTimestamp(3,  java.sql.Timestamp.valueOf(LocalDateTime.now()));
  129.         };
  130.         // 创建JDBC Sink
  131.         JdbcSink<LogStats> jdbcSink = new Jdbc().<LogStats>sinkBuilder()
  132.                 .withQueryStatement( new SimpleJdbcQueryStatement<LogStats>(insertSql,statementBuilder))
  133.                 .withExecutionOptions(jdbcExecutionOptions)
  134.                 .buildAtLeastOnce(connectionOptions);
  135.         statsStream.sinkTo(jdbcSink).name("JDBC Sink");
  136.         // 8. 执行作业
  137.         env.execute("Multi-Sink Data Pipeline");
  138.     }
  140. }
复制代码
4. 测试与验证

要测试这个完整的流水线，我们需要：

  • 启动Kafka并创建必要的主题：
    1. # 创建输入主题
    2. kafka-topics.sh --create --topic logs-input-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
    4. # 创建错误日志输出主题
    5. kafka-topics.sh --create --topic error-logs-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
    复制代码
  • 启动Elasticsearch并确保服务正常运行
  • 在MySQL中创建必要的表：
    1. CREATE DATABASE test;
    2. USE test;
    4. CREATE TABLE error_log_stats (
    5.   source VARCHAR(100) PRIMARY KEY,
    6.   count BIGINT NOT NULL,
    7.   last_updated TIMESTAMP NOT NULL
    8. );
    复制代码
  • 向Kafka发送测试数据：
    1. kafka-console-producer.sh --topic logs-input-topic --bootstrap-server localhost:9092
    3. # 输入以下测试数据
    4. 2025-09-29 12:00:00|INFO|application-service|Application started successfully
    5. 2025-09-29 12:01:30|ERROR|database-service|Failed to connect to database
    6. 2025-09-29 12:02:15|WARN|cache-service|Cache eviction threshold reached
    7. 2025-09-29 12:03:00|ERROR|authentication-service|Invalid credentials detected
    复制代码
  • 运行Flink作业并观察数据流向各个目标系统
查看Kafka Sink中的数据：
5.png

查看MySQL中的数据：
6.png

查看Elasticsearch中的数据：
7.png

八、性能优化与最佳实践

1. 并行度配置

合理设置Sink的并行度可以显著提高吞吐量：
  1. // 为特定Sink设置并行度
  2. stream.addSink(sink).setParallelism(4);
  4. // 或为整个作业设置默认并行度
  5. env.setParallelism(4);
复制代码
2. 批处理配置

对于支持批处理的Sink，合理配置批处理参数可以减少网络开销：
  1. // JDBC批处理示例
  2. JdbcExecutionOptions.builder()
  3.     .withBatchSize(1000)  // 每批次处理的记录数
  4.     .withBatchIntervalMs(200)  // 批处理间隔
  5.     .withMaxRetries(3)  // 最大重试次数
  6.     .build();
复制代码
3. 背压处理

当Sink无法处理上游数据时，会产生背压。Flink提供了背压监控和处理机制：

  • 使用Flink Web UI监控背压情况
  • 考虑使用缓冲机制或调整并行度
  • 对于关键路径，实现自定义的背压处理逻辑
4. 资源管理

合理管理连接和资源是保证Sink稳定运行的关键：

  • 使用连接池管理数据库连接
  • 在RichSinkFunction的open()方法中初始化资源
  • 在close()方法中正确释放资源
5. 错误处理策略

为Sink配置适当的错误处理策略：
  1. // 重试策略配置
  2. env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
  3.     3,  // 最大重试次数
  4.     Time.of(10, TimeUnit.SECONDS)  // 重试间隔
  5. ));
复制代码
九、总结与展望

本文深入探讨了Flink数据输出（Sink）的核心概念、各种连接器的使用方法以及可靠性保证机制。我们学习了如何配置和使用内置Sink、文件系统Sink、Kafka Sink、Elasticsearch Sink和JDBC Sink，并通过自定义Sink扩展了Flink的输出能力。最后，我们构建了一个完整的实时数据处理流水线，将处理后的数据输出到多个目标系统。
在Flink的数据处理生态中，Sink是连接计算结果与外部世界的桥梁。通过选择合适的Sink连接器并配置正确的参数，我们可以构建高效、可靠的数据处理系统。
源文来自：http://blog.daimajiangxin.com.cn
源码地址：https://gitee.com/daimajiangxin/flink-learning

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
开始Flink数据输出终极

使用道具 举报

相关推荐
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册
发帖
锷稠
前天 11:21

0

粉丝关注

16

主题发布

板块介绍填写区域，请于后台编辑
微信扫一扫