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

2024-09-03 13:44
文章标签 配置 类型 数据 使用 常见 redis 概念 原理 附带 案例 flink kafka sink 消费 写到 datastream 116

本文主要是介绍大数据-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。

相关文章

Python如何使用seleniumwire接管Chrome查看控制台中参数

Python如何使用seleniumwire接管Chrome查看控制台中参数

《Python如何使用seleniumwire接管Chrome查看控制台中参数》文章介绍了如何使用Python的seleniumwire库来接管Chrome浏览器,并通过控制台查看接口参数,本文给大家... 1、cmd打开控制台，启动谷歌并制定端口号，找不到文件的加环境变量chrome.exe --rem
阅读更多...
Oracle数据库使用 listagg去重删除重复数据的方法汇总

Oracle数据库使用 listagg去重删除重复数据的方法汇总

《Oracle数据库使用listagg去重删除重复数据的方法汇总》文章介绍了在Oracle数据库中使用LISTAGG和XMLAGG函数进行字符串聚合并去重的方法,包括去重聚合、使用XML解析和CLO... 目录案例表第一种：使用wm_concat() + distinct去重聚合第二种：使用listagg，
阅读更多...
使用C#代码计算数学表达式实例

使用C#代码计算数学表达式实例

《使用C#代码计算数学表达式实例》这段文字主要讲述了如何使用C#语言来计算数学表达式,该程序通过使用Dictionary保存变量,定义了运算符优先级,并实现了EvaluateExpression方法来... 目录C#代码计算数学表达式该方法很长，因此我将分段描述下面的代码片段显示了下一步以下代码显示该方法如
阅读更多...
龙蜥操作系统Anolis OS-23.x安装配置图解教程(保姆级)

龙蜥操作系统Anolis OS-23.x安装配置图解教程(保姆级)

《龙蜥操作系统AnolisOS-23.x安装配置图解教程(保姆级)》：本文主要介绍了安装和配置AnolisOS23.2系统，包括分区、软件选择、设置root密码、网络配置、主机名设置和禁用SELinux的步骤，详细内容请阅读本文，希望能对你有所帮助... ‌AnolisOS‌是由阿里云推出的开源操作系统，旨
阅读更多...
Go语言使用Buffer实现高性能处理字节和字符

Go语言使用Buffer实现高性能处理字节和字符

《Go语言使用Buffer实现高性能处理字节和字符》在Go中,bytes.Buffer是一个非常高效的类型,用于处理字节数据的读写操作,本文将详细介绍一下如何使用Buffer实现高性能处理字节和... 目录1. bytes.Buffer 的基本用法1.1. 创建和初始化 Buffer1.2. 使用 Writ
阅读更多...
Redis主从/哨兵机制原理分析

Redis主从/哨兵机制原理分析

《Redis主从/哨兵机制原理分析》本文介绍了Redis的主从复制和哨兵机制,主从复制实现了数据的热备份和负载均衡,而哨兵机制可以监控Redis集群,实现自动故障转移,哨兵机制通过监控、下线、选举和故... 目录一、主从复制1.1 什么是主从复制1.2 主从复制的作用1.3 主从复制原理1.3.1 全量复制
阅读更多...
Redis延迟队列的实现示例

Redis延迟队列的实现示例

《Redis延迟队列的实现示例》Redis延迟队列是一种使用Redis实现的消息队列,本文主要介绍了Redis延迟队列的实现示例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习... 目录一、什么是 Redis 延迟队列二、实现原理三、Java 代码示例四、注意事项五、使用 Redi
阅读更多...
Redis缓存问题与缓存更新机制详解

Redis缓存问题与缓存更新机制详解

《Redis缓存问题与缓存更新机制详解》本文主要介绍了缓存问题及其解决方案,包括缓存穿透、缓存击穿、缓存雪崩等问题的成因以及相应的预防和解决方法,同时,还详细探讨了缓存更新机制,包括不同情况下的缓存更... 目录一、缓存问题1.1 缓存穿透1.1.1 问题来源1.1.2 解决方案1.2 缓存击穿1.2.1
阅读更多...
redis-cli命令行工具的使用小结

redis-cli命令行工具的使用小结

《redis-cli命令行工具的使用小结》redis-cli是Redis的命令行客户端,支持多种参数用于连接、操作和管理Redis数据库,本文给大家介绍redis-cli命令行工具的使用小结,感兴趣的... 目录基本连接参数基本连接方式连接远程服务器带密码连接操作与格式参数-r参数重复执行命令-i参数指定命
阅读更多...
PyTorch使用教程之Tensor包详解

PyTorch使用教程之Tensor包详解

《PyTorch使用教程之Tensor包详解》这篇文章介绍了PyTorch中的张量（Tensor）数据结构,包括张量的数据类型、初始化、常用操作、属性等,张量是PyTorch框架中的核心数据结构,支持... 目录1、张量Tensor2、数据类型3、初始化（构造张量）4、常用操作5、常用属性5.1 存储（st
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2