怎么自学做网站网址没封的来一个

怎么自学做网站,网址没封的来一个,网店营销策略,长春企业网站制作优化实时数据处理的完整链路中#xff0c;数据输出#xff08;Sink#xff09;是最后一个关键环节#xff0c;它负责将处理后的结果传递到外部系统供后续使用。Flink提供了丰富的数据输出连接器#xff0c;支持将数据写入Kafka、Elasticsearch、文件系统、数据库等各种目标系统…实时数据处理的完整链路中数据输出Sink是最后一个关键环节它负责将处理后的结果传递到外部系统供后续使用。Flink提供了丰富的数据输出连接器支持将数据写入Kafka、Elasticsearch、文件系统、数据库等各种目标系统。本文将深入探讨Flink数据输出的核心概念、配置方法和最佳实践并基于Flink 1.20.1构建一个完整的数据输出案例。一、Flink Sink概述1. 什么是SinkSink接收器是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. 版本说明Flink1.20.1JDK17Gradle8.3外部系统Kafka 3.4.0、Elasticsearch 7.17.0、MySQL 8.02. 核心依赖dependencies {// Flink核心依赖implementation org.apache.flink:flink_core:1.20.1implementation org.apache.flink:flink-streaming-java:1.20.1implementation org.apache.flink:flink-clients:1.20.1// Kafka Connectorimplementation org.apache.flink:flink-connector-kafka:3.4.0-1.20// Elasticsearch Connectorimplementation org.apache.flink:flink-connector-elasticsearch7:3.1.0-1.20// JDBC Connectorimplementation org.apache.flink:flink-connector-jdbc:3.3.0-1.20implementation mysql:mysql-connector-java:8.0.33// FileSystem Connectorimplementation org.apache.flink:flink-connector-files:1.20.1}三、基础Sink操作1. 内置调试SinkFlink提供了一些内置的Sink用于开发和调试阶段import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class BasicSinkDemo {public static void main(String[] args) throws Exception {// 创建执行环境StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();// 创建数据源DataStreamString stream env.fromElements(Hello, Flink, Sink);// 打印到标准输出stream.print(StandardOutput);// 打印到标准错误输出stream.printToErr(ErrorOutput);// 执行作业env.execute(Basic Sink Demo);}}2. 文件系统SinkFlink支持将数据写入本地文件系统、HDFS等。下面是一个写入本地文件系统的示例package com.cn.daimajiangxin.flink.sink;import org.apache.flink.api.common.serialization.SimpleStringEncoder;import org.apache.flink.configuration.MemorySize;import org.apache.flink.connector.file.sink.FileSink;import org.apache.flink.core.fs.Path;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy;import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;import java.time.Duration;public class FileSystemSinkDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();DataStreamObject stream env.fromData(Hello, Flink, FileSystem, Sink);RollingPolicyObject, String rollingPolicy DefaultRollingPolicy.Object, Stringbuilder().withRolloverInterval(Duration.ofMinutes(15)).withInactivityInterval(Duration.ofMinutes(5)).withMaxPartSize(MemorySize.ofMebiBytes(64)).build();// 创建文件系统SinkFileSinkObject sink FileSink.forRowFormat(new Path(file:///tmp/flink-output), new SimpleStringEncoder()).withRollingPolicy(rollingPolicy).build();// 添加Sinkstream.sinkTo(sink);env.execute(File System Sink Demo);}}四、高级Sink连接器1. Kafka SinkKafka是实时数据处理中常用的消息队列Flink提供了强大的Kafka Sink支持import org.apache.flink.api.common.serialization.SimpleStringSchema;import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;import org.apache.flink.connector.kafka.sink.KafkaSink;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.Properties;public class KafkaSinkDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();// 开启检查点以支持Exactly-Once语义env.enableCheckpointing(5000);DataStreamString stream env.fromElements(Hello Kafka, Flink to Kafka, Data Pipeline);// Kafka配置Properties props new Properties();props.setProperty(bootstrap.servers, localhost:9092);// 创建Kafka SinkKafkaSinkString sink KafkaSink.Stringbuilder().setKafkaProducerConfig(props).setRecordSerializer(KafkaRecordSerializationSchema.builder().setTopic(flink-output-topic).setValueSerializationSchema(new SimpleStringSchema()).build()).build();// 添加Sinkstream.sinkTo(sink);env.execute(Kafka Sink Demo);}}kafka消息队列消息202509291047492. Elasticsearch SinkElasticsearch是一个实时的分布式搜索和分析引擎非常适合存储和查询Flink处理的实时数据package com.cn.daimajiangxin.flink.sink;import com.fasterxml.jackson.databind.ObjectMapper;import org.apache.flink.connector.elasticsearch.sink.Elasticsearch7SinkBuilder;import org.apache.flink.connector.elasticsearch.sink.ElasticsearchSink;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.http.HttpHost;import org.elasticsearch.action.index.IndexRequest;import org.elasticsearch.client.Requests;import java.util.Map;public class ElasticsearchSinkDemo {private static final ObjectMapper objectMapper new ObjectMapper();public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.enableCheckpointing(5000);DataStreamString stream env.fromData({\id\:\1\,\name\:\Flink\,\category\:\framework\},{\id\:\2\,\name\:\Elasticsearch\,\category\:\database\});// 配置Elasticsearch节点HttpHost httpHostnew HttpHost(localhost, 9200, http);// 创建Elasticsearch SinkElasticsearchSinkString sinknew Elasticsearch7SinkBuilderString().setBulkFlushMaxActions(10) // 批量操作数量.setBulkFlushInterval(5000) // 批量刷新间隔毫秒.setHosts(httpHost).setConnectionRequestTimeout(60000) // 连接请求超时时间.setConnectionTimeout(60000) // 连接超时时间.setSocketTimeout(60000) // Socket 超时时间.setEmitter((element, context, indexer) - {try {MapString, Object json objectMapper.readValue(element, Map.class);IndexRequest request Requests.indexRequest().index(flink_documents).id((String) json.get(id)).source(json);indexer.add(request);} catch (Exception e) {// 处理解析异常System.err.println(Failed to parse JSON: element);}}).build();// 添加Sinkstream.sinkTo(sink);env.execute(Elasticsearch Sink Demo);}}使用post工具查看数据wechat_2025-09-29_180718_2793. JDBC Sink使用JDBC Sink可以将数据写入各种关系型数据库package com.cn.daimajiangxin.flink.sink;import org.apache.flink.connector.jdbc.JdbcConnectionOptions;import org.apache.flink.connector.jdbc.JdbcExecutionOptions;import org.apache.flink.connector.jdbc.JdbcStatementBuilder;import org.apache.flink.connector.jdbc.core.datastream.Jdbc;import org.apache.flink.connector.jdbc.core.datastream.sink.JdbcSink;import org.apache.flink.connector.jdbc.datasource.statements.SimpleJdbcQueryStatement;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.Arrays;import java.util.List;public class JdbcSinkDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.enableCheckpointing(5000);ListUser userList Arrays.asList( new User(1, Alice, 25,alice),new User(2, Bob, 30,bob),new User(3, Charlie, 35,charlie));// 模拟用户数据DataStreamUser userStream env.fromData(userList);JdbcExecutionOptions jdbcExecutionOptions JdbcExecutionOptions.builder().withBatchSize(1000).withBatchIntervalMs(200).withMaxRetries(5).build();JdbcConnectionOptions connectionOptions new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withUrl(jdbc:mysql://localhost:3306/test).withDriverName(com.mysql.cj.jdbc.Driver).withUsername(username).withPassword(password).build();String insertSql INSERT INTO user (id, name, age, user_name) VALUES (?, ?, ?, ?);JdbcStatementBuilderUser statementBuilder (statement, user) - {statement.setInt(1, user.getId());statement.setString(2, user.getName());statement.setInt(3, user.getAge());statement.setString(4, user.getUserName());};// 创建JDBC SinkJdbcSinkUser jdbcSink new Jdbc().UsersinkBuilder().withQueryStatement( new SimpleJdbcQueryStatementUser(insertSql,statementBuilder)).withExecutionOptions(jdbcExecutionOptions).buildAtLeastOnce(connectionOptions);// 添加SinkuserStream.sinkTo(jdbcSink);env.execute(JDBC Sink Demo);}// 用户实体类public static class User {private int id;private String name;private String userName;private int age;public User(int id, String name, int age,String userName) {this.id id;this.name name;this.age age;this.userNameuserName;}public int getId() {return id;}public String getName() {return name;}public int getAge() {return age;}public String getUserName() {return userName;}}}登录mysql客户端查看数据20250930113343五、Sink的可靠性保证机制1. 检查点与保存点Flink的检查点Checkpoint机制是实现精确一次语义的基础。当开启检查点后Flink会定期将作业的状态保存到持久化存储中。如果作业失败Flink可以从最近的检查点恢复确保数据不会丢失。// 配置检查点StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();// 启用检查点间隔5000msenv.enableCheckpointing(5000);// 配置检查点模式为EXACTLY_ONCE默认env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);// 设置检查点超时时间env.getCheckpointConfig().setCheckpointTimeout(60000);// 设置最大并行检查点数量env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);// 开启外部化检查点作业失败时保留检查点env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);2. 事务与二阶段提交对于支持事务的外部系统Flink使用二阶段提交Two-Phase Commit协议来实现精确一次语义第一阶段预提交Flink将数据写入外部系统的预提交区域但不提交第二阶段提交所有算子完成预提交后Flink通知外部系统提交数据这种机制确保了即使在作业失败或恢复的情况下数据也不会被重复写入或丢失。3. 不同Sink的语义保证级别不同的Sink连接器支持不同级别的语义保证支持精确一次Exactly-onceKafka、Elasticsearch版本支持、文件系统预写日志模式支持至少一次At-least-onceJDBC、Redis、RabbitMQ最多一次At-most-once简单的无状态输出六、自定义Sink实现当Flink内置的Sink连接器不能满足需求时我们可以通过实现SinkFunction接口来自定义Sinkpackage com.cn.daimajiangxin.flink.sink;import org.apache.flink.api.common.functions.RuntimeContext;import org.apache.flink.api.connector.sink2.Sink;import org.apache.flink.api.connector.sink2.SinkWriter;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;import java.io.IOException;public class CustomSinkDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();DataStreamString stream env.fromElements(Custom, Sink, Example);// 使用自定义Sinkstream.sinkTo(new CustomSink());env.execute(Custom Sink Demo);}// 自定义Sink实现 - 使用新APIpublic static class CustomSink implements SinkString {Overridepublic SinkWriterString createWriter(InitContext context) {return new CustomSinkWriter();}// SinkWriter负责实际的数据写入逻辑private static class CustomSinkWriter implements SinkWriterString {// 初始化资源public CustomSinkWriter() {// 初始化连接、客户端等资源System.out.println(CustomSink initialized);}// 处理每个元素Overridepublic void write(String value, Context context) throws IOException, InterruptedException {// 实际的写入逻辑System.out.println(Writing to custom sink: value);}// 刷新缓冲区Overridepublic void flush(boolean endOfInput) {// 刷新逻辑如果需要}// 清理资源Overridepublic void close() throws Exception {// 关闭连接、客户端等资源System.out.println(CustomSink closed);}}}}sad20251006111134七、实战案例实时数据处理流水线下面我们将构建一个完整的实时数据处理流水线从Kafka读取数据进行转换处理然后输出到多个目标系统1. 系统架构Kafka Source - Flink Processing - Multiple Sinks|- Kafka Sink|- Elasticsearch Sink|- JDBC Sink2. 数据模型我们将使用日志数据模型定义一个LogEntry类来表示日志条目package com.cn.daimajiangxin.flink.sink;public class LogEntry {private String timestamp;private String logLevel;private String source;private String message;public String getTimestamp() {return timestamp;}public void setTimestamp(String timestamp) {this.timestamp timestamp;}public String getLogLevel() {return logLevel;}public void setLogLevel(String logLevel) {this.logLevel logLevel;}public String getSource() {return source;}public void setSource(String source) {this.source source;}public String getMessage() {return message;}public void setMessage(String message) {this.message message;}Overridepublic String toString() {return String.format(LogEntry{timestamp%s, logLevel%s, source%s, message%s},timestamp, logLevel, source, message);}}定义一个日志统计实体类LogStats用于表示每个源的日志统计信息package com.cn.daimajiangxin.flink.sink;public class LogStats {private String source;private long count;public LogStats() {}public LogStats(String source, long count) {this.source source;this.count count;}public String getSource() {return source;}public void setSource(String source) {this.source source;}public long getCount() {return count;}public void setCount(long count) {this.count count;}Overridepublic String toString() {return String.format(LogStats{source%s, count%d}, source, count);}}3. 完整实现代码package com.cn.daimajiangxin.flink.sink;import org.apache.flink.api.common.eventtime.WatermarkStrategy;import org.apache.flink.api.common.serialization.SimpleStringSchema;import org.apache.flink.connector.jdbc.JdbcConnectionOptions;import org.apache.flink.connector.jdbc.JdbcExecutionOptions;import org.apache.flink.connector.jdbc.JdbcStatementBuilder;import org.apache.flink.connector.jdbc.core.datastream.Jdbc;import org.apache.flink.connector.jdbc.core.datastream.sink.JdbcSink;import org.apache.flink.connector.jdbc.datasource.statements.SimpleJdbcQueryStatement;import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;import org.apache.flink.connector.kafka.sink.KafkaSink;import org.apache.flink.connector.kafka.source.KafkaSource;import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.connector.elasticsearch.sink.Elasticsearch7SinkBuilder;import org.apache.flink.connector.elasticsearch.sink.ElasticsearchSink;import org.apache.http.HttpHost;import org.elasticsearch.action.index.IndexRequest;import org.elasticsearch.client.Requests;import java.sql.PreparedStatement;import java.time.LocalDateTime;import java.util.ArrayList;import java.util.HashMap;import java.util.List;import java.util.Map;import java.util.Properties;public class MultiSinkPipeline {public static void main(String[] args) throws Exception {// 1. 创建执行环境并配置检查点StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();env.enableCheckpointing(5000);// 2. 创建Kafka SourceKafkaSourceString source KafkaSource.Stringbuilder().setBootstrapServers(localhost:9092).setTopics(logs-input-topic).setGroupId(flink-group).setStartingOffsets(OffsetsInitializer.earliest()).setValueOnlyDeserializer(new SimpleStringSchema()).build();// 3. 读取数据并解析DataStreamString kafkaStream env.fromSource(source, WatermarkStrategy.noWatermarks(), Kafka Source);// 解析日志数据DataStreamLogEntry logStream kafkaStream.map(line - {