大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1:消费Kafka写到Redis

本文主要是介绍大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1:消费Kafka写到Redis,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

点一下关注吧!!!非常感谢!!持续更新!!!

目前已经更新到了:

  • Hadoop(已更完)
  • HDFS(已更完)
  • MapReduce(已更完)
  • Hive(已更完)
  • Flume(已更完)
  • Sqoop(已更完)
  • Zookeeper(已更完)
  • HBase(已更完)
  • Redis (已更完)
  • Kafka(已更完)
  • Spark(已更完)
  • Flink(正在更新!)

章节内容

上节我们完成了如下的内容:

  • Flink DataStream Transformation
  • FlatMap Window Aggregations Reduce 等等等函数

在这里插入图片描述

Sink

Flink 的 Sink 是指数据流处理过程中最终输出数据的组件。在 Apache Flink 中,数据流从 Source 读取后经过一系列的转换操作,最后会被写入到 Sink 中。Sink 是 Flink 流式处理应用的终点,决定了处理后的数据如何保存或传输。

基本概念

Flink 的 Sink 是用来将流处理的数据写入外部存储系统的,比如数据库、文件系统、消息队列等。Sink 接口提供了一种灵活的方式来定义数据的输出格式和存储目标。Flink 提供了多个内置的 Sink 连接器,用户也可以根据需求自定义 Sink。

常见类型

Flink 提供了多种内置的 Sink,可以将数据输出到多种不同的系统中。以下是一些常见的 Flink Sink:

  • File Sink:将数据输出到文件系统,支持多种文件格式,如文本文件、CSV、Parquet 等。
  • Kafka Sink:将数据输出到 Kafka 主题,用于构建流式数据管道。
  • Elasticsearch Sink:将数据写入 Elasticsearch 索引,适用于实时数据搜索和分析。
  • JDBC Sink:将数据写入关系型数据库,如 MySQL、PostgreSQL 等。
  • HDFS Sink:将数据存储在 Hadoop 分布式文件系统中,适用于大规模数据的长期存储。
  • Cassandra Sink:将数据写入 Cassandra 数据库,适用于大规模的 NoSQL 数据存储

配置与使用

要在 Flink 应用中使用 Sink,需要通过 DataStream 的 addSink 方法来配置和添加 Sink。例如,将数据写入 Kafka 的简单配置如下:

DataStream<String> dataStream = // 数据处理逻辑
dataStream.addSink(new FlinkKafkaProducer<>("localhost:9092",         // Kafka broker 地址"output-topic",           // 输出的 Kafka 主题new SimpleStringSchema()   // 数据序列化格式
));

同样,配置 JDBC Sink 的方式如下:

dataStream.addSink(JdbcSink.sink("INSERT INTO my_table (column1, column2) VALUES (?, ?)",(statement, value) -> {statement.setString(1, value.f0);statement.setInt(2, value.f1);},JdbcExecutionOptions.builder().withBatchSize(1000).withBatchIntervalMs(200).build(),new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withUrl("jdbc:mysql://localhost:3306/mydb").withDriverName("com.mysql.jdbc.Driver").withUsername("user").withPassword("password").build()
));

自定义 Sink

除了使用内置的 Sink,Flink 还允许开发者实现自定义 Sink。通过实现 SinkFunction 接口或扩展 RichSinkFunction 类,开发者可以定义自己所需的 Sink。自定义 Sink 通常用于需要特殊处理或集成尚不支持的外部系统。

例如,自定义一个简单的控制台打印 Sink:

public class PrintSinkFunction<T> extends RichSinkFunction<T> {@Overridepublic void invoke(T value, Context context) {System.out.println(value);}
}

Sink 的容错机制

Flink 提供了精确一次 (Exactly-Once) 和至少一次 (At-Least-Once) 的容错语义,具体取决于 Sink 的类型及其配置。例如,Kafka Sink 通常支持精确一次语义,而某些文件系统 Sink 可能只支持至少一次语义。通过启用 Flink 的 Checkpointing 机制,Sink 可以在发生故障时从最近的检查点恢复,从而保证数据的一致性。

Sink 的并行度

Flink 的 Sink 通常是并行的,默认情况下与上游操作的并行度一致。用户可以通过 setParallelism 方法来手动调整 Sink 的并行度。注意,对于一些 Sink,如文件系统 Sink,并行度越高,生成的文件数也越多。

生命周期

Flink 的 Sink 在执行时会经历以下几个阶段:

  • 打开 (open):初始化资源,如数据库连接、文件句柄等。
  • 写入 (invoke):将每一条数据写入目标存储系统。
  • 关闭 (close):关闭资源,确保数据完整写入和资源的正确释放。

简单示例

以下是一个将处理后的数据流写入文本文件的完整示例:

DataStream<String> dataStream = // 数据处理逻辑
StreamingFileSink<String> sink = StreamingFileSink.forRowFormat(new Path("/output/path"), new SimpleStringEncoder<String>("UTF-8")).build();dataStream.addSink(sink);

案例1:数据写入Redis

添加依赖

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-redis_2.11</artifactId><version>1.1.5</version>
</dependency>

编写代码

消费Kafka 计算之后 写入到 Redis中。
Source(Kafka) -> Sink(Redis)

package icu.wzk;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;import java.util.Properties;public class StreamFromKafka {public static void main(String[] args) {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 配置信息Properties properties = new Properties();properties.setProperty("bootstrap.servers", "h121.wzk.icu:9092");// KafkaFlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("flink_test",new SimpleStringSchema(),properties);DataStreamSource<String> data = env.getJavaEnv().addSource(consumer);SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = data.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {@Overridepublic void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {String[] words = value.split(" ");for (String word: words) {out.collect(new Tuple2<>(word, 1));}}});SingleOutputStreamOperator<Tuple2<String, Integer>> result = wordAndOne.keyBy(new KeySelector<Tuple2<String, Integer>, Object>() {@Overridepublic Object getKey(Tuple2<String, Integer> value) throws Exception {return value.f0;}}).sum(1);result.print();env.execute("StreamFromKafka");}}

启动Kafka

在这里插入图片描述

启动Redis

在这里插入图片描述

运行代码

在这里插入图片描述

写入数据

在这里插入图片描述

查看结果

在这里插入图片描述

这篇关于大数据-116 - Flink DataStream Sink 原理、概念、常见Sink类型 配置与使用 附带案例1:消费Kafka写到Redis的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1133117

相关文章

postgresql使用UUID函数的方法

《postgresql使用UUID函数的方法》本文给大家介绍postgresql使用UUID函数的方法,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录PostgreSQL有两种生成uuid的方法。可以先通过sql查看是否已安装扩展函数,和可以安装的扩展函数

SpringBoot改造MCP服务器的详细说明(StreamableHTTP 类型)

《SpringBoot改造MCP服务器的详细说明(StreamableHTTP类型)》本文介绍了SpringBoot如何实现MCPStreamableHTTP服务器,并且使用CherryStudio... 目录SpringBoot改造MCP服务器(StreamableHTTP)1 项目说明2 使用说明2.1

Spring Boot Maven 插件如何构建可执行 JAR 的核心配置

《SpringBootMaven插件如何构建可执行JAR的核心配置》SpringBoot核心Maven插件,用于生成可执行JAR/WAR,内置服务器简化部署,支持热部署、多环境配置及依赖管理... 目录前言一、插件的核心功能与目标1.1 插件的定位1.2 插件的 Goals(目标)1.3 插件定位1.4 核

如何使用Lombok进行spring 注入

《如何使用Lombok进行spring注入》本文介绍如何用Lombok简化Spring注入,推荐优先使用setter注入,通过注解自动生成getter/setter及构造器,减少冗余代码,提升开发效... Lombok为了开发环境简化代码,好处不用多说。spring 注入方式为2种,构造器注入和setter

MySQL中比较运算符的具体使用

《MySQL中比较运算符的具体使用》本文介绍了SQL中常用的符号类型和非符号类型运算符,符号类型运算符包括等于(=)、安全等于(=)、不等于(/!=)、大小比较(,=,,=)等,感兴趣的可以了解一下... 目录符号类型运算符1. 等于运算符=2. 安全等于运算符<=>3. 不等于运算符<>或!=4. 小于运

使用zip4j实现Java中的ZIP文件加密压缩的操作方法

《使用zip4j实现Java中的ZIP文件加密压缩的操作方法》本文介绍如何通过Maven集成zip4j1.3.2库创建带密码保护的ZIP文件,涵盖依赖配置、代码示例及加密原理,确保数据安全性,感兴趣的... 目录1. zip4j库介绍和版本1.1 zip4j库概述1.2 zip4j的版本演变1.3 zip4

RabbitMQ消息总线方式刷新配置服务全过程

《RabbitMQ消息总线方式刷新配置服务全过程》SpringCloudBus通过消息总线与MQ实现微服务配置统一刷新,结合GitWebhooks自动触发更新,避免手动重启,提升效率与可靠性,适用于配... 目录前言介绍环境准备代码示例测试验证总结前言介绍在微服务架构中,为了更方便的向微服务实例广播消息,

Python 字典 (Dictionary)使用详解

《Python字典(Dictionary)使用详解》字典是python中最重要,最常用的数据结构之一,它提供了高效的键值对存储和查找能力,:本文主要介绍Python字典(Dictionary)... 目录字典1.基本特性2.创建字典3.访问元素4.修改字典5.删除元素6.字典遍历7.字典的高级特性默认字典

使用Python构建一个高效的日志处理系统

《使用Python构建一个高效的日志处理系统》这篇文章主要为大家详细讲解了如何使用Python开发一个专业的日志分析工具,能够自动化处理、分析和可视化各类日志文件,大幅提升运维效率,需要的可以了解下... 目录环境准备工具功能概述完整代码实现代码深度解析1. 类设计与初始化2. 日志解析核心逻辑3. 文件处

一文详解如何使用Java获取PDF页面信息

《一文详解如何使用Java获取PDF页面信息》了解PDF页面属性是我们在处理文档、内容提取、打印设置或页面重组等任务时不可或缺的一环,下面我们就来看看如何使用Java语言获取这些信息吧... 目录引言一、安装和引入PDF处理库引入依赖二、获取 PDF 页数三、获取页面尺寸(宽高)四、获取页面旋转角度五、判断